fix topic

src/efka_subscription.erl

@@ -13,7 +13,7 @@
 
 %% API
 -export([start_link/0]).
--export([subscribe/2, publish/3, debug_info/0]).
+-export([subscribe/2, unsubscribe_all/1, publish/3, debug_info/0]).
 -export([match_components/2, is_valid_components/1, of_components/1]).
 
 %% gen_server callbacks
@@ -51,6 +51,10 @@ subscribe(Topic, SubscriberPid) when is_binary(Topic), is_pid(SubscriberPid) ->
 publish(Topic, Qos, Content) when is_binary(Topic), is_integer(Qos), is_binary(Content) ->
     gen_server:cast(?SERVER, {publish, Topic, Qos, Content}).
 
+-spec unsubscribe_all(pid()) -> ok.
+unsubscribe_all(SubscriberPid) when is_pid(SubscriberPid) ->
+    gen_server:call(?SERVER, {unsubscribe_all, SubscriberPid}).
+
 -spec debug_info() -> {ok, Info :: map()}.
 debug_info() ->
     gen_server:call(?SERVER, debug_info).
@@ -107,6 +111,8 @@ handle_call({subscribe, Topic, SubscriberPid}, _From, State = #state{subscribers
         false ->
             {reply, {error, <<"invalid topic name">>}, State}
     end;
+handle_call({unsubscribe_all, SubscriberPid}, _From, State = #state{subscribers = Subscribers}) ->
+    {reply, ok, State#state{subscribers = remove_subscriber_pid(SubscriberPid, Subscribers)}};
 handle_call(debug_info, _From, State = #state{subscribers = Subscribers, remaining_messages = RemainingMessages}) ->
     Info = #{
         subscribes => Subscribers,
@@ -142,7 +148,7 @@ handle_cast({publish, Topic, Qos, Content}, State = #state{subscribers = Subscri
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_info({'DOWN', _Ref, process, SubscriberPid, Reason}, State = #state{subscribers = Subscribers}) ->
     logger:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
-    NSubscribers = lists:filter(fun(#subscriber{subscriber_pid = Pid0}) -> SubscriberPid /= Pid0 end, Subscribers),
+    NSubscribers = remove_subscriber_pid(SubscriberPid, Subscribers),
     {noreply, State#state{subscribers = NSubscribers}};
 
 handle_info(Info, State = #state{}) ->
@@ -243,6 +249,10 @@ broadcast(Topic, Content, MatchedSubscribers) ->
         SubscriberPid ! {topic_broadcast, Topic, Content}
                   end, MatchedSubscribers).
 
+-spec remove_subscriber_pid(pid(), [#subscriber{}]) -> [#subscriber{}].
+remove_subscriber_pid(SubscriberPid, Subscribers) when is_pid(SubscriberPid), is_list(Subscribers) ->
+    lists:filter(fun(#subscriber{subscriber_pid = SubscriberPid0}) -> SubscriberPid =/= SubscriberPid0 end, Subscribers).
+
 -spec dispatch_remaining_messages(Subscriber :: #subscriber{}, RemainingMessages :: list()) -> RestRemainingMessages :: list().
 dispatch_remaining_messages(#subscriber{subscriber_pid = SubscriberPid, components = Components}, RemainingMessages) when is_list(RemainingMessages) ->
     %% 处理遗留的消息

src/transport/service_channel.erl

@@ -19,6 +19,7 @@
 -record(state, {
     service_id :: undefined | binary(),
     service_pid :: undefined | pid(),
+    subscribed_topics = sets:new(),
     is_registered = false :: boolean()
 }).
 
@@ -75,6 +76,7 @@ websocket_info(Info, State) ->
 
 %% 进程关闭事件
 terminate(Reason, _Req, State = #state{service_id = ServiceId, is_registered = IsRegistered}) ->
+    ok = efka_subscription:unsubscribe_all(self()),
     case IsRegistered of
         true ->
             ok = service_model:change_status(ServiceId, 0);
@@ -114,15 +116,16 @@ handle_request(#'ServiceRequest'{packet_id = PacketId, request = {register, #'Se
 
 %% 订阅事件
 handle_request(#'ServiceRequest'{packet_id = PacketId, request = {subscribe, #'ServiceRequest.Subscribe'{topic = Topic}}},
-    State = #state{is_registered = true}) ->
+    State = #state{ subscribed_topics = SubscribedTopics, is_registered = true}) ->
 
-    Packet = case efka_subscription:subscribe(Topic, self()) of
-                 ok ->
-                     result_reply_packet(PacketId, <<"ok">>);
-                 {error, Reason} ->
-                     error_reply_packet(PacketId, -1, Reason)
-             end,
-    {reply, {binary, Packet}, State};
+    case efka_subscription:subscribe(Topic, self()) of
+        ok ->
+            Packet = result_reply_packet(PacketId, <<"ok">>),
+            {reply, {binary, Packet}, State#state{subscribed_topics = sets:add_element(Topic, SubscribedTopics)}};
+        {error, Reason} ->
+            Packet = error_reply_packet(PacketId, -1, Reason),
+            {reply, {binary, Packet}, State}
+    end;
 handle_request(#'ServiceRequest'{packet_id = PacketId}, State) ->
     {reply, {binary, error_reply_packet(PacketId, -1, <<"invalid request">>)}, State}.