anlicheng/ekfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix subscription

Browse Source
This commit is contained in:
anlicheng 2026-04-20 20:56:29 +08:00
parent 3f2aa00f35
commit 910987e8e5
1 changed files with 32 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
src/efka_subscription.erl
View File

@ -88,12 +88,22 @@ handle_call({subscribe, Topic, SubscriberPid}, _From, State = #state{subscribers
Components = of_components(Topic),
case is_valid_components(Components) of
true ->
Sub = #subscriber{topic = Topic, subscriber_pid = SubscriberPid, components = Components, order = order_num(Components)},
%% 建立到SubscriberPid的monitor，进程退出需要清理订阅
erlang:monitor(process, SubscriberPid),
%% 处理遗留的消息
RestRemainingMessages = dispatch_remaining_messages(Sub, RemainingMessages),
{reply, ok, State#state{subscribers = Subscribers ++ [Sub], remaining_messages = RestRemainingMessages}};
case has_subscription(Topic, SubscriberPid, Subscribers) of
true ->
{reply, ok, State};
false ->
Sub = #subscriber{topic = Topic, subscriber_pid = SubscriberPid, components = Components, order = order_num(Components)},
%% 只有首次看到该pid时才建立monitor，避免重复monitor
case has_subscriber_pid(SubscriberPid, Subscribers) of
true ->
ok;
false ->
erlang:monitor(process, SubscriberPid)
end,
%% 处理遗留的消息
RestRemainingMessages = dispatch_remaining_messages(Sub, RemainingMessages),
{reply, ok, State#state{subscribers = Subscribers ++ [Sub], remaining_messages = RestRemainingMessages}}
end;
false ->
{reply, {error, <<"invalid topic name">>}, State}
end;
@ -114,13 +124,13 @@ handle_call(debug_info, _From, State = #state{subscribers = Subscribers, remaini
handle_cast({publish, Topic, Qos, Content}, State = #state{subscribers = Subscribers, remaining_messages = RemainingMessages}) ->
MatchedSubscribers = match_subscribers(Subscribers, Topic),
logger:debug("[efka_subscription] topic: ~p, content: ~p, match subscribers: ~p", [Topic, Content, MatchedSubscribers]),
case length(MatchedSubscribers) > 0 of
true ->
case MatchedSubscribers of
[_|_] ->
broadcast(Topic, Content, MatchedSubscribers),
{noreply, State};
false when Qos =:= 0 ->
[] when Qos =:= 0 ->
{noreply, State};
false ->
[] ->
{noreply, State#state{remaining_messages = [{Topic, Content}|RemainingMessages]}}
end.
@ -178,6 +188,17 @@ match_subscribers(Subscribers, Topic) when is_list(Subscribers), is_binary(Topic
contain_channel(Pid, Subscribers) when is_pid(Pid), is_list(Subscribers) ->
lists:search(fun(#subscriber{subscriber_pid = Pid0}) -> Pid == Pid0 end, Subscribers) /= false.
-spec has_subscriber_pid(pid(), [#subscriber{}]) -> boolean().
has_subscriber_pid(SubscriberPid, Subscribers) when is_pid(SubscriberPid), is_list(Subscribers) ->
lists:any(fun(#subscriber{subscriber_pid = SubscriberPid0}) -> SubscriberPid =:= SubscriberPid0 end, Subscribers).
-spec has_subscription(binary(), pid(), [#subscriber{}]) -> boolean().
has_subscription(Topic, SubscriberPid, Subscribers)
when is_binary(Topic), is_pid(SubscriberPid), is_list(Subscribers) ->
lists:any(fun(#subscriber{topic = Topic0, subscriber_pid = SubscriberPid0}) ->
Topic =:= Topic0 andalso SubscriberPid =:= SubscriberPid0
end, Subscribers).
%% 开始对比订阅的topic和发布的topic的Components信息
%% *表示单级匹配，+表示多级匹配；+只能出现一次，并且只能在末尾
-spec match_components(list(), list()) -> boolean().
@ -227,7 +248,7 @@ dispatch_remaining_messages(#subscriber{subscriber_pid = SubscriberPid, componen
%% 处理遗留的消息
lists:foldl(fun({Topic0, Content0}, Acc) ->
Components0 = of_components(Topic0),
case match_components(Components0, Components) of
case match_components(Components, Components0) of
true ->
SubscriberPid ! {topic_broadcast, Topic0, Content0},
Acc;