diff --git a/src/efka_subscription.erl b/src/efka_subscription.erl index 0bf7770..b21a492 100644 --- a/src/efka_subscription.erl +++ b/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) -> %% 处理遗留的消息 diff --git a/src/transport/service_channel.erl b/src/transport/service_channel.erl index df37c13..efd545f 100644 --- a/src/transport/service_channel.erl +++ b/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}. @@ -147,4 +150,4 @@ error_reply_packet(PacketId, Code, Message) when is_integer(PacketId), is_intege packet_id = PacketId, reply = {error, #'ServiceReply.Error'{code = Code, message = Message}} }), - <>. \ No newline at end of file + <>.