yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore(shared): comments

This commit is contained in:
Georgy Sychev 2022-05-25 13:59:57 +04:00
parent 56093cb5be
commit b069a09674
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/emqx_shared_sub.erl
View File

@ -185,11 +185,15 @@ dispatch_with_ack(SubPid, Group, Topic, Msg, Type) ->
Ref = erlang:monitor(process, SubPid), Ref = erlang:monitor(process, SubPid),
Sender = self(), Sender = self(),
SubPid ! {deliver, Topic, with_group_ack(Msg, Group, Type, Sender, Ref)}, SubPid ! {deliver, Topic, with_group_ack(Msg, Group, Type, Sender, Ref)},
OldRef = old_ref(Type, Group, Ref),
Timeout = case Msg#message.qos of Timeout = case Msg#message.qos of
?QOS_1 -> timer:seconds(?SHARED_SUB_QOS1_DISPATCH_TIMEOUT_SECONDS); ?QOS_1 -> timer:seconds(?SHARED_SUB_QOS1_DISPATCH_TIMEOUT_SECONDS);
?QOS_2 -> infinity ?QOS_2 -> infinity
end, end,
%% This OldRef is a forward compatibilty workaround.
%% When an old version receives {Sender, OldRef} tuple, it acks (or nacks) back
%% with this OldRef structure
OldRef = old_ref(Type, Group, Ref),
try try
receive receive
{ReceivedRef, ?ACK} when ReceivedRef =:= Ref; ReceivedRef =:= OldRef -> {ReceivedRef, ?ACK} when ReceivedRef =:= Ref; ReceivedRef =:= OldRef ->
@ -237,7 +241,7 @@ maybe_nack_dropped(Msg) ->
?NO_ACK -> false; ?NO_ACK -> false;
{Sender, {fresh, _Group, Ref}} -> ok == nack(Sender, Ref, dropped); {Sender, {fresh, _Group, Ref}} -> ok == nack(Sender, Ref, dropped);
{_Sender, {retry, _Group, _Ref}} -> maybe_ack(Msg), false; {_Sender, {retry, _Group, _Ref}} -> maybe_ack(Msg), false;
%% This clause is for backward compatability %% This clause is for backward compatibility
Ack -> Ack ->
{Sender, Ref} = fetch_sender_ref(Ack), {Sender, Ref} = fetch_sender_ref(Ack),
ok == nack(Sender, Ref, dropped) ok == nack(Sender, Ref, dropped)
@ -268,7 +272,7 @@ maybe_ack(Msg) ->
end. end.
fetch_sender_ref({Sender, {_Type, _Group, Ref}}) -> {Sender, Ref}; fetch_sender_ref({Sender, {_Type, _Group, Ref}}) -> {Sender, Ref};
%% These clauses are for backward compatability %% These clauses are for backward compatibility
fetch_sender_ref({Sender, {_Group, Ref}}) -> {Sender, Ref}; fetch_sender_ref({Sender, {_Group, Ref}}) -> {Sender, Ref};
fetch_sender_ref({Sender, Ref}) -> {Sender, Ref}. fetch_sender_ref({Sender, Ref}) -> {Sender, Ref}.