迁移到ets

src/endpoint/endpoint_subscription.erl

@@ -20,6 +20,7 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
 
 -define(SERVER, ?MODULE).
+-define(SUBSCRIBER_TAB, endpoint_subscription_subscribers).
 
 %% 定义订阅者
 -record(subscriber, {
@@ -35,7 +36,7 @@
 }).
 
 -record(state, {
-    subscribers = []
+    tid :: ets:tid()
 }).
 
 %%%===================================================================
@@ -56,7 +57,18 @@ get_subscribers() ->
 
 -spec publish(RouteKey :: binary(), Content :: binary()) -> no_return().
 publish(RouteKey, Content) when is_binary(RouteKey), is_binary(Content) ->
-    gen_server:cast(?SERVER, {publish, RouteKey, Content}).
+    case ets:info(?SUBSCRIBER_TAB) of
+        undefined ->
+            ok;
+        _ ->
+            Subscribers = ets:tab2list(?SUBSCRIBER_TAB),
+            MatchedSubscribers = match_subscribers(Subscribers, RouteKey),
+            lists:foreach(fun(#subscriber{subscriber_pid = SubscriberPid}) ->
+                endpoint:forward(SubscriberPid, Content)
+            end, MatchedSubscribers),
+            logger:debug("[efka_subscription] route_key: ~p, metric: ~p, match subscribers: ~p", [RouteKey, Content, MatchedSubscribers]),
+            ok
+    end.
 
 %% @doc Spawns the server and registers the local name (unique)
 -spec(start_link() ->
@@ -75,7 +87,8 @@ start_link() ->
     {stop, Reason :: term()} | ignore).
 init([]) ->
     ok = iot_log:set_metadata(),
-    {ok, #state{}}.
+    Tid = ets:new(?SUBSCRIBER_TAB, [named_table, protected, bag, {keypos, 2}]),
+    {ok, #state{tid = Tid}}.
 
 %% @private
 %% @doc Handling call messages
@@ -88,14 +101,15 @@ init([]) ->
     {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
     {stop, Reason :: term(), NewState :: #state{}}).
 %%  同一个SubscriberPid只能订阅同一个topic一次
-handle_call(get_subscribers, _From, State = #state{subscribers = Subscribers}) ->
+handle_call(get_subscribers, _From, State = #state{tid = Tid}) ->
+    Subscribers = ets:tab2list(Tid),
     {reply, {ok, Subscribers}, State};
 
-handle_call({subscribe, Topic, SubscriberPid}, _From, State = #state{subscribers = Subscribers}) ->
+handle_call({subscribe, Topic, SubscriberPid}, _From, State = #state{tid = Tid}) ->
     Components = of_components(Topic),
     case is_valid_components(Components) of
         true ->
-            case has_subscription(Topic, SubscriberPid, Subscribers) of
+            case has_subscription(Tid, Topic, SubscriberPid) of
                 true ->
                     {reply, ok, State};
                 false ->
@@ -108,22 +122,23 @@ handle_call({subscribe, Topic, SubscriberPid}, _From, State = #state{subscribers
                         monitor_ref = MonitorRef,
                         order = order_num(Components)
                     },
-                    {reply, ok, State#state{subscribers = [Sub | Subscribers]}}
+                    true = ets:insert(Tid, Sub),
+                    {reply, ok, State}
             end;
         false ->
             {reply, {error, <<"invalid topic name">>}, State}
     end;
 
-handle_call({unsubscribe, Topic, SubscriberPid}, _From, State = #state{subscribers = Subscribers}) ->
+handle_call({unsubscribe, Topic, SubscriberPid}, _From, State = #state{tid = Tid}) ->
-    {Removed, Reserved} = lists:partition(fun(#subscriber{topic = Topic0, subscriber_pid = SubscriberPid0}) ->
+    Removed = [Sub || Sub = #subscriber{subscriber_pid = SubscriberPid0} <- ets:lookup(Tid, Topic),
-        Topic =:= Topic0 andalso SubscriberPid =:= SubscriberPid0
+        SubscriberPid =:= SubscriberPid0],
-    end, Subscribers),
     lists:foreach(fun(#subscriber{monitor_ref = MonitorRef}) when is_reference(MonitorRef) ->
         erlang:demonitor(MonitorRef, [flush]);
                      (_) ->
         ok
     end, Removed),
-    {reply, ok, State#state{subscribers = Reserved}}.
+    lists:foreach(fun(Sub) -> true = ets:delete_object(Tid, Sub) end, Removed),
+    {reply, ok, State}.
 
 %% @private
 %% @doc Handling cast messages
@@ -131,13 +146,7 @@ handle_call({unsubscribe, Topic, SubscriberPid}, _From, State = #state{subscribe
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-%% 发布消息
+handle_cast(_Request, State = #state{}) ->
-handle_cast({publish, RouteKey, Metric}, State = #state{subscribers = Subscribers}) ->
-    MatchedSubscribers = match_subscribers(Subscribers, RouteKey),
-    lists:foreach(fun(#subscriber{subscriber_pid = SubscriberPid}) ->
-        endpoint:forward(SubscriberPid, Metric)
-                  end, MatchedSubscribers),
-    logger:debug("[efka_subscription] route_key: ~p, metric: ~p, match subscribers: ~p", [RouteKey, Metric, MatchedSubscribers]),
     {noreply, State}.
 
 %% @private
@@ -146,10 +155,15 @@ handle_cast({publish, RouteKey, Metric}, State = #state{subscribers = Subscriber
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_info({'DOWN', _Ref, process, SubscriberPid, Reason}, State = #state{subscribers = Subscribers}) ->
+handle_info({'DOWN', _Ref, process, SubscriberPid, Reason}, State = #state{tid = Tid}) ->
     logger:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
-    NSubscribers = lists:filter(fun(#subscriber{subscriber_pid = Pid0}) -> SubscriberPid /= Pid0 end, Subscribers),
+    Subscribers = ets:tab2list(Tid),
-    {noreply, State#state{subscribers = NSubscribers}};
+    lists:foreach(fun(Sub = #subscriber{subscriber_pid = Pid0}) when Pid0 =:= SubscriberPid ->
+        true = ets:delete_object(Tid, Sub);
+                     (_) ->
+        ok
+    end, Subscribers),
+    {noreply, State};
 
 handle_info(Info, State = #state{}) ->
     logger:debug("[efka_subscription] get unknown info: ~p", [Info]),
@@ -226,8 +240,9 @@ order_num([<<$+>>|_]) ->
 order_num([_|Tail]) ->
     order_num(Tail).
 
--spec has_subscription(binary(), pid(), [#subscriber{}]) -> boolean().
+-spec has_subscription(ets:tid(), binary(), pid()) -> boolean().
-has_subscription(Topic, SubscriberPid, Subscribers) ->
+has_subscription(Tid, Topic, SubscriberPid) ->
+    Subscribers = ets:lookup(Tid, Topic),
     lists:any(fun(#subscriber{topic = Topic0, subscriber_pid = SubscriberPid0}) ->
         Topic =:= Topic0 andalso SubscriberPid =:= SubscriberPid0
     end, Subscribers).