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

refactor(session): pass ClientInfo as first params to APIs of emqx_session

This commit is contained in:
Shawn 2022-02-08 11:07:03 +08:00
parent b2f027bcf7
commit 06168f7080
4 changed files with 145 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
apps/emqx/src/emqx_channel.erl
View File

@ -365,7 +365,7 @@ handle_in(Packet = ?PUBLISH_PACKET(_QoS), Channel) ->
handle_in(?PUBACK_PACKET(PacketId, _ReasonCode, Properties), Channel handle_in(?PUBACK_PACKET(PacketId, _ReasonCode, Properties), Channel
= #channel{clientinfo = ClientInfo, session = Session}) -> = #channel{clientinfo = ClientInfo, session = Session}) ->
case emqx_session:puback(PacketId, Session) of case emqx_session:puback(ClientInfo, PacketId, Session) of
{ok, Msg, NSession} -> {ok, Msg, NSession} ->
ok = after_message_acked(ClientInfo, Msg, Properties), ok = after_message_acked(ClientInfo, Msg, Properties),
{ok, set_session(NSession, Channel)}; {ok, set_session(NSession, Channel)};
@ -384,7 +384,7 @@ handle_in(?PUBACK_PACKET(PacketId, _ReasonCode, Properties), Channel
handle_in(?PUBREC_PACKET(PacketId, _ReasonCode, Properties), Channel handle_in(?PUBREC_PACKET(PacketId, _ReasonCode, Properties), Channel
= #channel{clientinfo = ClientInfo, session = Session}) -> = #channel{clientinfo = ClientInfo, session = Session}) ->
case emqx_session:pubrec(PacketId, Session) of case emqx_session:pubrec(ClientInfo, PacketId, Session) of
{ok, Msg, NSession} -> {ok, Msg, NSession} ->
ok = after_message_acked(ClientInfo, Msg, Properties), ok = after_message_acked(ClientInfo, Msg, Properties),
NChannel = set_session(NSession, Channel), NChannel = set_session(NSession, Channel),
@ -399,8 +399,9 @@ handle_in(?PUBREC_PACKET(PacketId, _ReasonCode, Properties), Channel
handle_out(pubrel, {PacketId, RC}, Channel) handle_out(pubrel, {PacketId, RC}, Channel)
end; end;
handle_in(?PUBREL_PACKET(PacketId, _ReasonCode), Channel = #channel{session = Session}) -> handle_in(?PUBREL_PACKET(PacketId, _ReasonCode), Channel = #channel{clientinfo = ClientInfo,
case emqx_session:pubrel(PacketId, Session) of session = Session}) ->
case emqx_session:pubrel(ClientInfo, PacketId, Session) of
{ok, NSession} -> {ok, NSession} ->
NChannel = set_session(NSession, Channel), NChannel = set_session(NSession, Channel),
handle_out(pubcomp, {PacketId, ?RC_SUCCESS}, NChannel); handle_out(pubcomp, {PacketId, ?RC_SUCCESS}, NChannel);
@ -410,8 +411,9 @@ handle_in(?PUBREL_PACKET(PacketId, _ReasonCode), Channel = #channel{session = Se
handle_out(pubcomp, {PacketId, RC}, Channel) handle_out(pubcomp, {PacketId, RC}, Channel)
end; end;
handle_in(?PUBCOMP_PACKET(PacketId, _ReasonCode), Channel = #channel{session = Session}) -> handle_in(?PUBCOMP_PACKET(PacketId, _ReasonCode), Channel = #channel{
case emqx_session:pubcomp(PacketId, Session) of clientinfo = ClientInfo, session = Session}) ->
case emqx_session:pubcomp(ClientInfo, PacketId, Session) of
{ok, NSession} -> {ok, NSession} ->
{ok, set_session(NSession, Channel)}; {ok, set_session(NSession, Channel)};
{ok, Publishes, NSession} -> {ok, Publishes, NSession} ->
@ -617,8 +619,8 @@ do_publish(PacketId, Msg = #message{qos = ?QOS_1}, Channel) ->
handle_out(puback, {PacketId, RC}, NChannel); handle_out(puback, {PacketId, RC}, NChannel);
do_publish(PacketId, Msg = #message{qos = ?QOS_2}, do_publish(PacketId, Msg = #message{qos = ?QOS_2},
Channel = #channel{session = Session}) -> Channel = #channel{clientinfo = ClientInfo, session = Session}) ->
case emqx_session:publish(PacketId, Msg, Session) of case emqx_session:publish(ClientInfo, PacketId, Msg, Session) of
{ok, PubRes, NSession} -> {ok, PubRes, NSession} ->
RC = puback_reason_code(PubRes), RC = puback_reason_code(PubRes),
NChannel0 = set_session(NSession, Channel), NChannel0 = set_session(NSession, Channel),
@ -779,22 +781,22 @@ maybe_update_expiry_interval(_Properties, Channel) -> Channel.
handle_deliver(Delivers, Channel = #channel{takeover = true, handle_deliver(Delivers, Channel = #channel{takeover = true,
pendings = Pendings, pendings = Pendings,
session = Session, session = Session,
clientinfo = #{clientid := ClientId}}) -> clientinfo = #{clientid := ClientId} = ClientInfo}) ->
%% NOTE: Order is important here. While the takeover is in %% NOTE: Order is important here. While the takeover is in
%% progress, the session cannot enqueue messages, since it already %% progress, the session cannot enqueue messages, since it already
%% passed on the queue to the new connection in the session state. %% passed on the queue to the new connection in the session state.
NPendings = lists:append( NPendings = lists:append(
Pendings, Pendings,
emqx_session:ignore_local(maybe_nack(Delivers), ClientId, Session)), emqx_session:ignore_local(ClientInfo, maybe_nack(Delivers), ClientId, Session)),
{ok, Channel#channel{pendings = NPendings}}; {ok, Channel#channel{pendings = NPendings}};
handle_deliver(Delivers, Channel = #channel{conn_state = disconnected, handle_deliver(Delivers, Channel = #channel{conn_state = disconnected,
takeover = false, takeover = false,
session = Session, session = Session,
clientinfo = #{clientid := ClientId}}) -> clientinfo = #{clientid := ClientId} = ClientInfo}) ->
Delivers1 = maybe_nack(Delivers), Delivers1 = maybe_nack(Delivers),
Delivers2 = emqx_session:ignore_local(Delivers1, ClientId, Session), Delivers2 = emqx_session:ignore_local(ClientInfo, Delivers1, ClientId, Session),
NSession = emqx_session:enqueue(Delivers2, Session), NSession = emqx_session:enqueue(ClientInfo, Delivers2, Session),
NChannel = set_session(NSession, Channel), NChannel = set_session(NSession, Channel),
%% We consider queued/dropped messages as delivered since they are now in the session state. %% We consider queued/dropped messages as delivered since they are now in the session state.
maybe_mark_as_delivered(Session, Delivers), maybe_mark_as_delivered(Session, Delivers),
@ -802,9 +804,10 @@ handle_deliver(Delivers, Channel = #channel{conn_state = disconnected,
handle_deliver(Delivers, Channel = #channel{session = Session, handle_deliver(Delivers, Channel = #channel{session = Session,
takeover = false, takeover = false,
clientinfo = #{clientid := ClientId} clientinfo = #{clientid := ClientId} = ClientInfo
}) -> }) ->
case emqx_session:deliver(emqx_session:ignore_local(Delivers, ClientId, Session), Session) of case emqx_session:deliver(ClientInfo,
emqx_session:ignore_local(ClientInfo, Delivers, ClientId, Session), Session) of
{ok, Publishes, NSession} -> {ok, Publishes, NSession} ->
NChannel = set_session(NSession, Channel), NChannel = set_session(NSession, Channel),
maybe_mark_as_delivered(NSession, Delivers), maybe_mark_as_delivered(NSession, Delivers),
@ -1111,8 +1114,8 @@ handle_timeout(_TRef, retry_delivery,
Channel = #channel{conn_state = disconnected}) -> Channel = #channel{conn_state = disconnected}) ->
{ok, Channel}; {ok, Channel};
handle_timeout(_TRef, retry_delivery, handle_timeout(_TRef, retry_delivery,
Channel = #channel{session = Session}) -> Channel = #channel{session = Session, clientinfo = ClientInfo}) ->
case emqx_session:retry(Session) of case emqx_session:retry(ClientInfo, Session) of
{ok, NSession} -> {ok, NSession} ->
{ok, clean_timer(retry_timer, set_session(NSession, Channel))}; {ok, clean_timer(retry_timer, set_session(NSession, Channel))};
{ok, Publishes, Timeout, NSession} -> {ok, Publishes, Timeout, NSession} ->
@ -1124,8 +1127,8 @@ handle_timeout(_TRef, expire_awaiting_rel,
Channel = #channel{conn_state = disconnected}) -> Channel = #channel{conn_state = disconnected}) ->
{ok, Channel}; {ok, Channel};
handle_timeout(_TRef, expire_awaiting_rel, handle_timeout(_TRef, expire_awaiting_rel,
Channel = #channel{session = Session}) -> Channel = #channel{session = Session, clientinfo = ClientInfo}) ->
case emqx_session:expire(awaiting_rel, Session) of case emqx_session:expire(ClientInfo, awaiting_rel, Session) of
{ok, NSession} -> {ok, NSession} ->
{ok, clean_timer(await_timer, set_session(NSession, Channel))}; {ok, clean_timer(await_timer, set_session(NSession, Channel))};
{ok, Timeout, NSession} -> {ok, Timeout, NSession} ->
@ -1690,11 +1693,11 @@ maybe_resume_session(#channel{resuming = false}) ->
maybe_resume_session(#channel{session = Session, maybe_resume_session(#channel{session = Session,
resuming = true, resuming = true,
pendings = Pendings, pendings = Pendings,
clientinfo = #{clientid := ClientId}}) -> clientinfo = #{clientid := ClientId} = ClientInfo}) ->
{ok, Publishes, Session1} = emqx_session:replay(Session), {ok, Publishes, Session1} = emqx_session:replay(ClientInfo, Session),
%% We consider queued/dropped messages as delivered since they are now in the session state. %% We consider queued/dropped messages as delivered since they are now in the session state.
emqx_persistent_session:mark_as_delivered(ClientId, Pendings), emqx_persistent_session:mark_as_delivered(ClientId, Pendings),
case emqx_session:deliver(Pendings, Session1) of case emqx_session:deliver(ClientInfo, Pendings, Session1) of
{ok, Session2} -> {ok, Session2} ->
{ok, Publishes, Session2}; {ok, Publishes, Session2};
{ok, More, Session2} -> {ok, More, Session2} ->

3
apps/emqx/src/emqx_connection.erl
View File

@ -135,14 +135,13 @@
, 'recv_msg.qos0' , 'recv_msg.qos0'
, 'recv_msg.qos1' , 'recv_msg.qos1'
, 'recv_msg.qos2' , 'recv_msg.qos2'
, 'recv_msg.dropped'
, 'recv_msg.dropped.expired'
, send_pkt , send_pkt
, send_msg , send_msg
, 'send_msg.qos0' , 'send_msg.qos0'
, 'send_msg.qos1' , 'send_msg.qos1'
, 'send_msg.qos2' , 'send_msg.qos2'
, 'send_msg.dropped' , 'send_msg.dropped'
, 'send_msg.dropped.await_pubrel_timeout'
, 'send_msg.dropped.expired' , 'send_msg.dropped.expired'
, 'send_msg.dropped.queue_full' , 'send_msg.dropped.queue_full'
, 'send_msg.dropped.too_large' , 'send_msg.dropped.too_large'

227
apps/emqx/src/emqx_session.erl
View File

@ -66,27 +66,27 @@
, unsubscribe/4 , unsubscribe/4
]). ]).
-export([ publish/3 -export([ publish/4
, puback/2 , puback/3
, pubrec/2 , pubrec/3
, pubrel/2 , pubrel/3
, pubcomp/2 , pubcomp/3
]). ]).
-export([ deliver/2 -export([ deliver/3
, enqueue/2 , enqueue/3
, dequeue/1 , dequeue/2
, ignore_local/3 , ignore_local/4
, retry/1 , retry/2
, terminate/3 , terminate/3
]). ]).
-export([ takeover/1 -export([ takeover/1
, resume/2 , resume/2
, replay/1 , replay/2
]). ]).
-export([expire/2]). -export([expire/3]).
%% Export for CT %% Export for CT
-export([set_field/3]). -export([set_field/3]).
@ -277,18 +277,19 @@ stats(Session) -> info(?STATS_KEYS, Session).
%% Ignore local messages %% Ignore local messages
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
ignore_local(Delivers, Subscriber, Session) -> ignore_local(ClientInfo, Delivers, Subscriber, Session) ->
Subs = info(subscriptions, Session), Subs = info(subscriptions, Session),
lists:dropwhile(fun({deliver, Topic, #message{from = Publisher}}) -> lists:dropwhile(fun({deliver, Topic, #message{from = Publisher} = Msg}) ->
case maps:find(Topic, Subs) of case maps:find(Topic, Subs) of
{ok, #{nl := 1}} when Subscriber =:= Publisher -> {ok, #{nl := 1}} when Subscriber =:= Publisher ->
ok = emqx_metrics:inc('delivery.dropped'), ok = emqx_hooks:run('delivery.dropped', [ClientInfo, Msg, no_local]),
ok = emqx_metrics:inc('delivery.dropped.no_local'), ok = emqx_metrics:inc('delivery.dropped'),
true; ok = emqx_metrics:inc('delivery.dropped.no_local'),
_ -> true;
false _ ->
end false
end, Delivers). end
end, Delivers).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Client -> Broker: SUBSCRIBE %% Client -> Broker: SUBSCRIBE
@ -310,7 +311,6 @@ subscribe(ClientInfo = #{clientid := ClientId}, TopicFilter, SubOpts,
true -> {error, ?RC_QUOTA_EXCEEDED} true -> {error, ?RC_QUOTA_EXCEEDED}
end. end.
-compile({inline, [is_subscriptions_full/1]}).
is_subscriptions_full(#session{max_subscriptions = infinity}) -> is_subscriptions_full(#session{max_subscriptions = infinity}) ->
false; false;
is_subscriptions_full(#session{subscriptions = Subs, is_subscriptions_full(#session{subscriptions = Subs,
@ -340,10 +340,10 @@ unsubscribe(ClientInfo, TopicFilter, UnSubOpts,
%% Client -> Broker: PUBLISH %% Client -> Broker: PUBLISH
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(publish(emqx_types:packet_id(), emqx_types:message(), session()) -spec(publish(emqx_types:clientinfo(), emqx_types:packet_id(), emqx_types:message(), session())
-> {ok, emqx_types:publish_result(), session()} -> {ok, emqx_types:publish_result(), session()}
| {error, emqx_types:reason_code()}). | {error, emqx_types:reason_code()}).
publish(PacketId, Msg = #message{qos = ?QOS_2, timestamp = Ts}, publish(_ClientInfo, PacketId, Msg = #message{qos = ?QOS_2, timestamp = Ts},
Session = #session{awaiting_rel = AwaitingRel}) -> Session = #session{awaiting_rel = AwaitingRel}) ->
case is_awaiting_full(Session) of case is_awaiting_full(Session) of
false -> false ->
@ -359,10 +359,9 @@ publish(PacketId, Msg = #message{qos = ?QOS_2, timestamp = Ts},
end; end;
%% Publish QoS0/1 directly %% Publish QoS0/1 directly
publish(_PacketId, Msg, Session) -> publish(_ClientInfo, _PacketId, Msg, Session) ->
{ok, emqx_broker:publish(Msg), Session}. {ok, emqx_broker:publish(Msg), Session}.
-compile({inline, [is_awaiting_full/1]}).
is_awaiting_full(#session{max_awaiting_rel = infinity}) -> is_awaiting_full(#session{max_awaiting_rel = infinity}) ->
false; false;
is_awaiting_full(#session{awaiting_rel = AwaitingRel, is_awaiting_full(#session{awaiting_rel = AwaitingRel,
@ -373,23 +372,22 @@ is_awaiting_full(#session{awaiting_rel = AwaitingRel,
%% Client -> Broker: PUBACK %% Client -> Broker: PUBACK
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(puback(emqx_types:packet_id(), session()) -spec(puback(emqx_types:clientinfo(), emqx_types:packet_id(), session())
-> {ok, emqx_types:message(), session()} -> {ok, emqx_types:message(), session()}
| {ok, emqx_types:message(), replies(), session()} | {ok, emqx_types:message(), replies(), session()}
| {error, emqx_types:reason_code()}). | {error, emqx_types:reason_code()}).
puback(PacketId, Session = #session{inflight = Inflight}) -> puback(ClientInfo, PacketId, Session = #session{inflight = Inflight}) ->
case emqx_inflight:lookup(PacketId, Inflight) of case emqx_inflight:lookup(PacketId, Inflight) of
{value, {Msg, _Ts}} when is_record(Msg, message) -> {value, {Msg, _Ts}} when is_record(Msg, message) ->
Inflight1 = emqx_inflight:delete(PacketId, Inflight), Inflight1 = emqx_inflight:delete(PacketId, Inflight),
Session2 = update_latency(Msg, Session), Session2 = update_latency(Msg, Session),
return_with(Msg, dequeue(Session2#session{inflight = Inflight1})); return_with(Msg, dequeue(ClientInfo, Session2#session{inflight = Inflight1}));
{value, {_Pubrel, _Ts}} -> {value, {_Pubrel, _Ts}} ->
{error, ?RC_PACKET_IDENTIFIER_IN_USE}; {error, ?RC_PACKET_IDENTIFIER_IN_USE};
none -> none ->
{error, ?RC_PACKET_IDENTIFIER_NOT_FOUND} {error, ?RC_PACKET_IDENTIFIER_NOT_FOUND}
end. end.
-compile({inline, [return_with/2]}).
return_with(Msg, {ok, Session}) -> return_with(Msg, {ok, Session}) ->
{ok, Msg, Session}; {ok, Msg, Session};
return_with(Msg, {ok, Publishes, Session}) -> return_with(Msg, {ok, Publishes, Session}) ->
@ -399,10 +397,10 @@ return_with(Msg, {ok, Publishes, Session}) ->
%% Client -> Broker: PUBREC %% Client -> Broker: PUBREC
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(pubrec(emqx_types:packet_id(), session()) -spec(pubrec(emqx_types:clientinfo(), emqx_types:packet_id(), session())
-> {ok, emqx_types:message(), session()} -> {ok, emqx_types:message(), session()}
| {error, emqx_types:reason_code()}). | {error, emqx_types:reason_code()}).
pubrec(PacketId, Session = #session{inflight = Inflight}) -> pubrec(_ClientInfo, PacketId, Session = #session{inflight = Inflight}) ->
case emqx_inflight:lookup(PacketId, Inflight) of case emqx_inflight:lookup(PacketId, Inflight) of
{value, {Msg, _Ts}} when is_record(Msg, message) -> {value, {Msg, _Ts}} when is_record(Msg, message) ->
Update = with_ts(#pubrel_await{timestamp = Msg#message.timestamp}), Update = with_ts(#pubrel_await{timestamp = Msg#message.timestamp}),
@ -418,9 +416,9 @@ pubrec(PacketId, Session = #session{inflight = Inflight}) ->
%% Client -> Broker: PUBREL %% Client -> Broker: PUBREL
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(pubrel(emqx_types:packet_id(), session()) -spec(pubrel(emqx_types:clientinfo(), emqx_types:packet_id(), session())
-> {ok, session()} | {error, emqx_types:reason_code()}). -> {ok, session()} | {error, emqx_types:reason_code()}).
pubrel(PacketId, Session = #session{awaiting_rel = AwaitingRel}) -> pubrel(_ClientInfo, PacketId, Session = #session{awaiting_rel = AwaitingRel}) ->
case maps:take(PacketId, AwaitingRel) of case maps:take(PacketId, AwaitingRel) of
{_Ts, AwaitingRel1} -> {_Ts, AwaitingRel1} ->
{ok, Session#session{awaiting_rel = AwaitingRel1}}; {ok, Session#session{awaiting_rel = AwaitingRel1}};
@ -432,15 +430,15 @@ pubrel(PacketId, Session = #session{awaiting_rel = AwaitingRel}) ->
%% Client -> Broker: PUBCOMP %% Client -> Broker: PUBCOMP
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(pubcomp(emqx_types:packet_id(), session()) -spec(pubcomp(emqx_types:clientinfo(), emqx_types:packet_id(), session())
-> {ok, session()} | {ok, replies(), session()} -> {ok, session()} | {ok, replies(), session()}
| {error, emqx_types:reason_code()}). | {error, emqx_types:reason_code()}).
pubcomp(PacketId, Session = #session{inflight = Inflight}) -> pubcomp(ClientInfo, PacketId, Session = #session{inflight = Inflight}) ->
case emqx_inflight:lookup(PacketId, Inflight) of case emqx_inflight:lookup(PacketId, Inflight) of
{value, {Pubrel, _Ts}} when is_record(Pubrel, pubrel_await) -> {value, {Pubrel, _Ts}} when is_record(Pubrel, pubrel_await) ->
Session2 = update_latency(Pubrel, Session), Session2 = update_latency(Pubrel, Session),
Inflight1 = emqx_inflight:delete(PacketId, Inflight), Inflight1 = emqx_inflight:delete(PacketId, Inflight),
dequeue(Session2#session{inflight = Inflight1}); dequeue(ClientInfo, Session2#session{inflight = Inflight1});
{value, _Other} -> {value, _Other} ->
{error, ?RC_PACKET_IDENTIFIER_IN_USE}; {error, ?RC_PACKET_IDENTIFIER_IN_USE};
none -> none ->
@ -451,29 +449,30 @@ pubcomp(PacketId, Session = #session{inflight = Inflight}) ->
%% Dequeue Msgs %% Dequeue Msgs
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
dequeue(Session = #session{inflight = Inflight, mqueue = Q}) -> dequeue(ClientInfo, Session = #session{inflight = Inflight, mqueue = Q}) ->
case emqx_mqueue:is_empty(Q) of case emqx_mqueue:is_empty(Q) of
true -> {ok, Session}; true -> {ok, Session};
false -> false ->
{Msgs, Q1} = dequeue(batch_n(Inflight), [], Q), {Msgs, Q1} = dequeue(ClientInfo, batch_n(Inflight), [], Q),
deliver(Msgs, [], Session#session{mqueue = Q1}) deliver(Msgs, [], Session#session{mqueue = Q1})
end. end.
dequeue(0, Msgs, Q) -> dequeue(_ClientInfo, 0, Msgs, Q) ->
{lists:reverse(Msgs), Q}; {lists:reverse(Msgs), Q};
dequeue(Cnt, Msgs, Q) -> dequeue(ClientInfo, Cnt, Msgs, Q) ->
case emqx_mqueue:out(Q) of case emqx_mqueue:out(Q) of
{empty, _Q} -> dequeue(0, Msgs, Q); {empty, _Q} -> dequeue(ClientInfo, 0, Msgs, Q);
{{value, Msg}, Q1} -> {{value, Msg}, Q1} ->
case emqx_message:is_expired(Msg) of case emqx_message:is_expired(Msg) of
true -> ok = inc_expired_cnt(delivery), true ->
dequeue(Cnt, Msgs, Q1); ok = emqx_hooks:run('delivery.dropped', [ClientInfo, Msg, expired]),
false -> dequeue(acc_cnt(Msg, Cnt), [Msg|Msgs], Q1) ok = inc_delivery_expired_cnt(),
dequeue(ClientInfo, Cnt, Msgs, Q1);
false -> dequeue(ClientInfo, acc_cnt(Msg, Cnt), [Msg|Msgs], Q1)
end end
end. end.
-compile({inline, [acc_cnt/2]}).
acc_cnt(#message{qos = ?QOS_0}, Cnt) -> Cnt; acc_cnt(#message{qos = ?QOS_0}, Cnt) -> Cnt;
acc_cnt(_Msg, Cnt) -> Cnt - 1. acc_cnt(_Msg, Cnt) -> Cnt - 1.
@ -481,38 +480,38 @@ acc_cnt(_Msg, Cnt) -> Cnt - 1.
%% Broker -> Client: Deliver %% Broker -> Client: Deliver
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(deliver(list(emqx_types:deliver()), session()) -spec(deliver(emqx_types:clientinfo(), list(emqx_types:deliver()), session())
-> {ok, session()} | {ok, replies(), session()}). -> {ok, session()} | {ok, replies(), session()}).
deliver([Deliver], Session) -> %% Optimize deliver(ClientInfo, [Deliver], Session) -> %% Optimize
Enrich = enrich_fun(Session), Msg = enrich_deliver(Deliver, Session),
deliver_msg(Enrich(Deliver), Session); deliver_msg(ClientInfo, Msg, Session);
deliver(Delivers, Session) -> deliver(ClientInfo, Delivers, Session) ->
Msgs = lists:map(enrich_fun(Session), Delivers), Msgs = [enrich_deliver(D, Session) || D <- Delivers],
deliver(Msgs, [], Session). do_deliver(ClientInfo, Msgs, [], Session).
deliver([], Publishes, Session) -> do_deliver(_ClientInfo, [], Publishes, Session) ->
{ok, lists:reverse(Publishes), Session}; {ok, lists:reverse(Publishes), Session};
deliver([Msg | More], Acc, Session) -> do_deliver(ClientInfo, [Msg | More], Acc, Session) ->
case deliver_msg(Msg, Session) of case deliver_msg(ClientInfo, Msg, Session) of
{ok, Session1} -> {ok, Session1} ->
deliver(More, Acc, Session1); do_deliver(ClientInfo, More, Acc, Session1);
{ok, [Publish], Session1} -> {ok, [Publish], Session1} ->
deliver(More, [Publish|Acc], Session1) do_deliver(ClientInfo, More, [Publish|Acc], Session1)
end. end.
deliver_msg(Msg = #message{qos = ?QOS_0}, Session) -> deliver_msg(_ClientInfo, Msg = #message{qos = ?QOS_0}, Session) ->
{ok, [{undefined, maybe_ack(Msg)}], Session}; {ok, [{undefined, maybe_ack(Msg)}], Session};
deliver_msg(Msg = #message{qos = QoS}, Session = deliver_msg(ClientInfo, Msg = #message{qos = QoS}, Session =
#session{next_pkt_id = PacketId, inflight = Inflight}) #session{next_pkt_id = PacketId, inflight = Inflight})
when QoS =:= ?QOS_1 orelse QoS =:= ?QOS_2 -> when QoS =:= ?QOS_1 orelse QoS =:= ?QOS_2 ->
case emqx_inflight:is_full(Inflight) of case emqx_inflight:is_full(Inflight) of
true -> true ->
Session1 = case maybe_nack(Msg) of Session1 = case maybe_nack(Msg) of
true -> Session; true -> Session;
false -> enqueue(Msg, Session) false -> enqueue(ClientInfo, Msg, Session)
end, end,
{ok, Session1}; {ok, Session1};
false -> false ->
@ -521,32 +520,34 @@ deliver_msg(Msg = #message{qos = QoS}, Session =
{ok, [Publish], next_pkt_id(Session1)} {ok, [Publish], next_pkt_id(Session1)}
end. end.
-spec(enqueue(list(emqx_types:deliver())|emqx_types:message(), -spec(enqueue(emqx_types:clientinfo(), list(emqx_types:deliver())|emqx_types:message(),
session()) -> session()). session()) -> session()).
enqueue([Deliver], Session) -> %% Optimize enqueue(ClientInfo, Delivers, Session) when is_list(Delivers) ->
Enrich = enrich_fun(Session), lists:foldl(fun(Deliver, Session0) ->
enqueue(Enrich(Deliver), Session); Msg = enrich_deliver(Deliver, Session),
enqueue(ClientInfo, Msg, Session0)
end, Session, Delivers);
enqueue(Delivers, Session) when is_list(Delivers) -> enqueue(ClientInfo, #message{} = Msg, Session = #session{mqueue = Q}) ->
Msgs = lists:map(enrich_fun(Session), Delivers),
lists:foldl(fun enqueue/2, Session, Msgs);
enqueue(Msg, Session = #session{mqueue = Q}) when is_record(Msg, message) ->
{Dropped, NewQ} = emqx_mqueue:in(Msg, Q), {Dropped, NewQ} = emqx_mqueue:in(Msg, Q),
(Dropped =/= undefined) andalso log_dropped(Dropped, Session), (Dropped =/= undefined) andalso handle_dropped(ClientInfo, Dropped, Session),
Session#session{mqueue = NewQ}. Session#session{mqueue = NewQ}.
log_dropped(Msg = #message{qos = QoS, topic = Topic}, #session{mqueue = Q}) -> handle_dropped(ClientInfo, Msg = #message{qos = QoS, topic = Topic}, #session{mqueue = Q}) ->
Payload = emqx_message:to_log_map(Msg), Payload = emqx_message:to_log_map(Msg),
#{store_qos0 := StoreQos0} = QueueInfo = emqx_mqueue:info(Q), #{store_qos0 := StoreQos0} = QueueInfo = emqx_mqueue:info(Q),
case (QoS == ?QOS_0) andalso (not StoreQos0) of case (QoS == ?QOS_0) andalso (not StoreQos0) of
true -> true ->
ok = emqx_hooks:run('delivery.dropped', [ClientInfo, Msg, qos0_msg]),
ok = emqx_metrics:inc('delivery.dropped'),
ok = emqx_metrics:inc('delivery.dropped.qos0_msg'), ok = emqx_metrics:inc('delivery.dropped.qos0_msg'),
ok = inc_pd('send_msg.dropped'), ok = inc_pd('send_msg.dropped'),
?SLOG(warning, #{msg => "dropped_qos0_msg", ?SLOG(warning, #{msg => "dropped_qos0_msg",
queue => QueueInfo, queue => QueueInfo,
payload => Payload}, #{topic => Topic}); payload => Payload}, #{topic => Topic});
false -> false ->
ok = emqx_hooks:run('delivery.dropped', [ClientInfo, Msg, queue_full]),
ok = emqx_metrics:inc('delivery.dropped'),
ok = emqx_metrics:inc('delivery.dropped.queue_full'), ok = emqx_metrics:inc('delivery.dropped.queue_full'),
ok = inc_pd('send_msg.dropped'), ok = inc_pd('send_msg.dropped'),
ok = inc_pd('send_msg.dropped.queue_full'), ok = inc_pd('send_msg.dropped.queue_full'),
@ -555,10 +556,8 @@ log_dropped(Msg = #message{qos = QoS, topic = Topic}, #session{mqueue = Q}) ->
payload => Payload}, #{topic => Topic}) payload => Payload}, #{topic => Topic})
end. end.
enrich_fun(Session = #session{subscriptions = Subs}) -> enrich_deliver({deliver, Topic, Msg}, Session = #session{subscriptions = Subs}) ->
fun({deliver, Topic, Msg}) -> enrich_subopts(get_subopts(Topic, Subs), Msg, Session).
enrich_subopts(get_subopts(Topic, Subs), Msg, Session)
end.
maybe_ack(Msg) -> maybe_ack(Msg) ->
case emqx_shared_sub:is_ack_required(Msg) of case emqx_shared_sub:is_ack_required(Msg) of
@ -613,53 +612,54 @@ await(PacketId, Msg, Session = #session{inflight = Inflight}) ->
%% Retry Delivery %% Retry Delivery
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(retry(session()) -> {ok, session()} | {ok, replies(), timeout(), session()}). -spec(retry(emqx_types:clientinfo(), session()) ->
retry(Session = #session{inflight = Inflight, retry_interval = RetryInterval}) -> {ok, session()} | {ok, replies(), timeout(), session()}).
retry(ClientInfo, Session = #session{inflight = Inflight, retry_interval = RetryInterval}) ->
case emqx_inflight:is_empty(Inflight) of case emqx_inflight:is_empty(Inflight) of
true -> {ok, Session}; true -> {ok, Session};
false -> false ->
Now = erlang:system_time(millisecond), Now = erlang:system_time(millisecond),
Session2 = check_expire_latency(Now, RetryInterval, Session), Session2 = check_expire_latency(Now, RetryInterval, Session),
retry_delivery(emqx_inflight:to_list(sort_fun(), Inflight), retry_delivery(emqx_inflight:to_list(sort_fun(), Inflight), [], Now,
[], Session2, ClientInfo)
Now,
Session2)
end. end.
retry_delivery([], Acc, _Now, Session = #session{retry_interval = Interval}) -> retry_delivery([], Acc, _Now, Session = #session{retry_interval = Interval}, _ClientInfo) ->
{ok, lists:reverse(Acc), Interval, Session}; {ok, lists:reverse(Acc), Interval, Session};
retry_delivery([{PacketId, {Msg, Ts}}|More], Acc, Now, Session = retry_delivery([{PacketId, {Msg, Ts}}|More], Acc, Now, Session =
#session{retry_interval = Interval, inflight = Inflight}) -> #session{retry_interval = Interval, inflight = Inflight}, ClientInfo) ->
case (Age = age(Now, Ts)) >= Interval of case (Age = age(Now, Ts)) >= Interval of
true -> true ->
{Acc1, Inflight1} = retry_delivery(PacketId, Msg, Now, Acc, Inflight), {Acc1, Inflight1} = do_retry_delivery(PacketId, Msg, Now, Acc, Inflight, ClientInfo),
retry_delivery(More, Acc1, Now, Session#session{inflight = Inflight1}); retry_delivery(More, Acc1, Now, Session#session{inflight = Inflight1}, ClientInfo);
false -> false ->
{ok, lists:reverse(Acc), Interval - max(0, Age), Session} {ok, lists:reverse(Acc), Interval - max(0, Age), Session}
end. end.
retry_delivery(PacketId, Msg, Now, Acc, Inflight) when is_record(Msg, message) -> do_retry_delivery(PacketId, pubrel, Now, Acc, Inflight, _) ->
Inflight1 = emqx_inflight:update(PacketId, {pubrel, Now}, Inflight),
{[{pubrel, PacketId}|Acc], Inflight1};
do_retry_delivery(PacketId, #message{} = Msg, Now, Acc, Inflight, ClientInfo) ->
case emqx_message:is_expired(Msg) of case emqx_message:is_expired(Msg) of
true -> true ->
ok = inc_expired_cnt(delivery), ok = emqx_hooks:run('delivery.dropped', [ClientInfo, Msg, expired]),
ok = inc_delivery_expired_cnt(),
{Acc, emqx_inflight:delete(PacketId, Inflight)}; {Acc, emqx_inflight:delete(PacketId, Inflight)};
false -> false ->
Msg1 = emqx_message:set_flag(dup, true, Msg), Msg1 = emqx_message:set_flag(dup, true, Msg),
Inflight1 = emqx_inflight:update(PacketId, {Msg1, Now}, Inflight), Inflight1 = emqx_inflight:update(PacketId, {Msg1, Now}, Inflight),
{[{PacketId, Msg1}|Acc], Inflight1} {[{PacketId, Msg1}|Acc], Inflight1}
end; end.
retry_delivery(PacketId, Pubrel, Now, Acc, Inflight) ->
Inflight1 = emqx_inflight:update(PacketId, {Pubrel, Now}, Inflight),
{[{pubrel, PacketId}|Acc], Inflight1}.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Expire Awaiting Rel %% Expire Awaiting Rel
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-spec(expire(awaiting_rel, session()) -> {ok, session()} | {ok, timeout(), session()}). -spec(expire(emqx_types:clientinfo(), awaiting_rel, session()) ->
expire(awaiting_rel, Session = #session{awaiting_rel = AwaitingRel}) -> {ok, session()} | {ok, timeout(), session()}).
expire(_ClientInfo, awaiting_rel, Session = #session{awaiting_rel = AwaitingRel}) ->
case maps:size(AwaitingRel) of case maps:size(AwaitingRel) of
0 -> {ok, Session}; 0 -> {ok, Session};
_ -> expire_awaiting_rel(erlang:system_time(millisecond), Session) _ -> expire_awaiting_rel(erlang:system_time(millisecond), Session)
@ -670,7 +670,7 @@ expire_awaiting_rel(Now, Session = #session{awaiting_rel = AwaitingRel,
NotExpired = fun(_PacketId, Ts) -> age(Now, Ts) < Timeout end, NotExpired = fun(_PacketId, Ts) -> age(Now, Ts) < Timeout end,
AwaitingRel1 = maps:filter(NotExpired, AwaitingRel), AwaitingRel1 = maps:filter(NotExpired, AwaitingRel),
ExpiredCnt = maps:size(AwaitingRel) - maps:size(AwaitingRel1), ExpiredCnt = maps:size(AwaitingRel) - maps:size(AwaitingRel1),
(ExpiredCnt > 0) andalso inc_expired_cnt(message, ExpiredCnt), (ExpiredCnt > 0) andalso inc_await_pubrel_timeout(ExpiredCnt),
NSession = Session#session{awaiting_rel = AwaitingRel1}, NSession = Session#session{awaiting_rel = AwaitingRel1},
case maps:size(AwaitingRel1) of case maps:size(AwaitingRel1) of
0 -> {ok, NSession}; 0 -> {ok, NSession};
@ -693,14 +693,14 @@ resume(ClientInfo = #{clientid := ClientId}, Session = #session{subscriptions =
ok = emqx_metrics:inc('session.resumed'), ok = emqx_metrics:inc('session.resumed'),
emqx_hooks:run('session.resumed', [ClientInfo, info(Session)]). emqx_hooks:run('session.resumed', [ClientInfo, info(Session)]).
-spec(replay(session()) -> {ok, replies(), session()}). -spec(replay(emqx_types:clientinfo(), session()) -> {ok, replies(), session()}).
replay(Session = #session{inflight = Inflight}) -> replay(ClientInfo, Session = #session{inflight = Inflight}) ->
Pubs = lists:map(fun({PacketId, {Pubrel, _Ts}}) when is_record(Pubrel, pubrel_await) -> Pubs = lists:map(fun({PacketId, {Pubrel, _Ts}}) when is_record(Pubrel, pubrel_await) ->
{pubrel, PacketId}; {pubrel, PacketId};
({PacketId, {Msg, _Ts}}) -> ({PacketId, {Msg, _Ts}}) ->
{PacketId, emqx_message:set_flag(dup, true, Msg)} {PacketId, emqx_message:set_flag(dup, true, Msg)}
end, emqx_inflight:to_list(Inflight)), end, emqx_inflight:to_list(Inflight)),
case dequeue(Session) of case dequeue(ClientInfo, Session) of
{ok, NSession} -> {ok, Pubs, NSession}; {ok, NSession} -> {ok, Pubs, NSession};
{ok, More, NSession} -> {ok, More, NSession} ->
{ok, lists:append(Pubs, More), NSession} {ok, lists:append(Pubs, More), NSession}
@ -714,29 +714,26 @@ terminate(ClientInfo, takenover, Session) ->
terminate(ClientInfo, Reason, Session) -> terminate(ClientInfo, Reason, Session) ->
run_hook('session.terminated', [ClientInfo, Reason, info(Session)]). run_hook('session.terminated', [ClientInfo, Reason, info(Session)]).
-compile({inline, [run_hook/2]}).
run_hook(Name, Args) -> run_hook(Name, Args) ->
ok = emqx_metrics:inc(Name), emqx_hooks:run(Name, Args). ok = emqx_metrics:inc(Name), emqx_hooks:run(Name, Args).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Inc message/delivery expired counter %% Inc message/delivery expired counter
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
inc_delivery_expired_cnt() ->
inc_delivery_expired_cnt(1).
-compile({inline, [inc_expired_cnt/1, inc_expired_cnt/2]}). inc_delivery_expired_cnt(N) ->
inc_expired_cnt(K) -> inc_expired_cnt(K, 1).
inc_expired_cnt(delivery, N) ->
ok = inc_pd('send_msg.dropped', N), ok = inc_pd('send_msg.dropped', N),
ok = inc_pd('send_msg.dropped.expired', N), ok = inc_pd('send_msg.dropped.expired', N),
ok = emqx_metrics:inc('delivery.dropped', N), ok = emqx_metrics:inc('delivery.dropped', N),
emqx_metrics:inc('delivery.dropped.expired', N); emqx_metrics:inc('delivery.dropped.expired', N).
inc_expired_cnt(message, N) -> inc_await_pubrel_timeout(N) ->
ok = inc_pd('recv_msg.dropped', N), ok = inc_pd('send_msg.dropped', N),
ok = inc_pd('recv_msg.dropped.expired', N), ok = inc_pd('send_msg.dropped.await_pubrel_timeout', N),
ok = emqx_metrics:inc('messages.dropped', N), ok = emqx_metrics:inc('messages.dropped', N),
emqx_metrics:inc('messages.dropped.expired', N). emqx_metrics:inc('messages.dropped.await_pubrel_timeout', N).
inc_pd(Key) -> inc_pd(Key) ->
inc_pd(Key, 1). inc_pd(Key, 1).
@ -747,9 +744,6 @@ inc_pd(Key, Inc) ->
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Next Packet Id %% Next Packet Id
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-compile({inline, [next_pkt_id/1]}).
next_pkt_id(Session = #session{next_pkt_id = ?MAX_PACKET_ID}) -> next_pkt_id(Session = #session{next_pkt_id = ?MAX_PACKET_ID}) ->
Session#session{next_pkt_id = 1}; Session#session{next_pkt_id = 1};
@ -788,9 +782,6 @@ get_birth_timestamp(_, _) ->
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Helper functions %% Helper functions
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-compile({inline, [sort_fun/0, batch_n/1, with_ts/1, age/2]}).
sort_fun() -> sort_fun() ->
fun({_, {_, Ts1}}, {_, {_, Ts2}}) -> Ts1 =< Ts2 end. fun({_, {_, Ts1}}, {_, {_, Ts2}}) -> Ts1 =< Ts2 end.

22
apps/emqx_management/src/emqx_mgmt_api_clients.erl
View File

@ -152,10 +152,6 @@ properties(client) ->
<<"Number of PUBLISH QoS1 packets received">>}, <<"Number of PUBLISH QoS1 packets received">>},
{'recv_msg.qos2', integer, {'recv_msg.qos2', integer,
<<"Number of PUBLISH QoS2 packets received">>}, <<"Number of PUBLISH QoS2 packets received">>},
{'recv_msg.dropped', integer,
<<"Number of dropped PUBLISH packets">>},
{'recv_msg.dropped.expired', integer,
<<"Number of dropped PUBLISH packets due to expired">>},
{recv_oct, integer, {recv_oct, integer,
<<"Number of bytes received by EMQ X Broker (the same below)">>}, <<"Number of bytes received by EMQ X Broker (the same below)">>},
{recv_pkt, integer, {recv_pkt, integer,
@ -165,21 +161,23 @@ properties(client) ->
{send_cnt, integer, {send_cnt, integer,
<<"Number of TCP packets sent">>}, <<"Number of TCP packets sent">>},
{send_msg, integer, {send_msg, integer,
<<"Number of PUBLISH packets sent">>}, <<"Number of PUBLISH messages sent">>},
{'send_msg.qos0', integer, {'send_msg.qos0', integer,
<<"Number of PUBLISH QoS0 packets sent">>}, <<"Number of PUBLISH QoS0 messages sent">>},
{'send_msg.qos1', integer, {'send_msg.qos1', integer,
<<"Number of PUBLISH QoS1 packets sent">>}, <<"Number of PUBLISH QoS1 messages sent">>},
{'send_msg.qos2', integer, {'send_msg.qos2', integer,
<<"Number of PUBLISH QoS2 packets sent">>}, <<"Number of PUBLISH QoS2 messages sent">>},
{'send_msg.dropped', integer, {'send_msg.dropped', integer,
<<"Number of dropped PUBLISH packets">>}, <<"Number of dropped PUBLISH messages">>},
{'send_msg.dropped.await_pubrel_timeout', integer,
<<"Number of dropped PUBLISH packets due to waiting PUBREL timeout">>},
{'send_msg.dropped.expired', integer, {'send_msg.dropped.expired', integer,
<<"Number of dropped PUBLISH packets due to expired">>}, <<"Number of dropped PUBLISH messages due to expired">>},
{'send_msg.dropped.queue_full', integer, {'send_msg.dropped.queue_full', integer,
<<"Number of dropped PUBLISH packets due to queue full">>}, <<"Number of dropped PUBLISH messages due to queue full">>},
{'send_msg.dropped.too_large', integer, {'send_msg.dropped.too_large', integer,
<<"Number of dropped PUBLISH packets due to packet length too large">>}, <<"Number of dropped PUBLISH messages due to packet length too large">>},
{send_oct, integer, {send_oct, integer,
<<"Number of bytes sent">>}, <<"Number of bytes sent">>},
{send_pkt, integer, {send_pkt, integer,