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

fix issue #244

This commit is contained in:
Feng Lee 2015-08-14 12:16:41 +08:00
parent 322f2f61f5
commit d23bf74d1e
2 changed files with 23 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/emqttd_mqueue.erl
View File

@ -59,7 +59,8 @@
-export([new/3, name/1, -export([new/3, name/1,
is_empty/1, is_full/1, is_empty/1, is_full/1,
len/1, max_len/1, len/1, max_len/1,
in/2, out/1]). in/2, out/1,
stats/1]).
-define(LOW_WM, 0.2). -define(LOW_WM, 0.2).
@ -72,6 +73,7 @@
high_wm = ?HIGH_WM, high_wm = ?HIGH_WM,
max_len = ?MAX_LEN, max_len = ?MAX_LEN,
qos0 = false, qos0 = false,
dropped = 0,
alarm_fun}). alarm_fun}).
-type mqueue() :: #mqueue{}. -type mqueue() :: #mqueue{}.
@ -111,6 +113,9 @@ len(#mqueue{len = Len}) -> Len.
max_len(#mqueue{max_len= MaxLen}) -> MaxLen. max_len(#mqueue{max_len= MaxLen}) -> MaxLen.
stats(#mqueue{max_len = MaxLen, len = Len, dropped = Dropped}) ->
[{max_len, MaxLen}, {len, Len}, {dropped, Dropped}].
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% @doc Queue one message. %% @doc Queue one message.
%% @end %% @end
@ -122,11 +127,11 @@ in(#mqtt_message{qos = ?QOS_0}, MQ = #mqueue{qos0 = false}) ->
MQ; MQ;
%% simply drop the oldest one if queue is full, improve later %% simply drop the oldest one if queue is full, improve later
in(Msg, MQ = #mqueue{name = Name, q = Q, len = Len, max_len = MaxLen}) in(Msg, MQ = #mqueue{q = Q, len = Len, max_len = MaxLen, dropped = Dropped})
when Len =:= MaxLen -> when Len =:= MaxLen ->
{{value, OldMsg}, Q2} = queue:out(Q), {{value, _OldMsg}, Q2} = queue:out(Q),
lager:error("MQueue(~s) drop ~s", [Name, emqttd_message:format(OldMsg)]), %lager:error("MQueue(~s) drop ~s", [Name, emqttd_message:format(OldMsg)]),
MQ#mqueue{q = queue:in(Msg, Q2)}; MQ#mqueue{q = queue:in(Msg, Q2), dropped = Dropped +1};
in(Msg, MQ = #mqueue{q = Q, len = Len}) -> in(Msg, MQ = #mqueue{q = Q, len = Len}) ->
maybe_set_alarm(MQ#mqueue{q = queue:in(Msg, Q), len = Len + 1}). maybe_set_alarm(MQ#mqueue{q = queue:in(Msg, Q), len = Len + 1}).

22
src/emqttd_session.erl
View File

@ -647,21 +647,23 @@ start_collector(Session = #session{collect_interval = Interval}) ->
TRef = erlang:send_after(Interval * 1000, self(), collect_info), TRef = erlang:send_after(Interval * 1000, self(), collect_info),
Session#session{collect_timer = TRef}. Session#session{collect_timer = TRef}.
info(#session{clean_sess = CleanSess, info(#session{clean_sess = CleanSess,
subscriptions = Subscriptions, subscriptions = Subscriptions,
inflight_queue = InflightQueue, inflight_queue = InflightQueue,
max_inflight = MaxInflight, max_inflight = MaxInflight,
message_queue = MessageQueue, message_queue = MessageQueue,
awaiting_rel = AwaitingRel, awaiting_rel = AwaitingRel,
awaiting_ack = AwaitingAck, awaiting_ack = AwaitingAck,
awaiting_comp = AwaitingComp, awaiting_comp = AwaitingComp,
timestamp = CreatedAt}) -> timestamp = CreatedAt}) ->
Stats = emqttd_mqueue:stats(MessageQueue),
[{pid, self()}, [{pid, self()},
{clean_sess, CleanSess}, {clean_sess, CleanSess},
{subscriptions, Subscriptions}, {subscriptions, Subscriptions},
{max_inflight, MaxInflight}, {max_inflight, MaxInflight},
{inflight_queue, length(InflightQueue)}, {inflight_queue, length(InflightQueue)},
{message_queue, emqttd_mqueue:len(MessageQueue)}, {message_queue, proplists:get_value(len, Stats)},
{message_dropped, proplists:get_value(dropped, Stats)},
{awaiting_rel, maps:size(AwaitingRel)}, {awaiting_rel, maps:size(AwaitingRel)},
{awaiting_ack, maps:size(AwaitingAck)}, {awaiting_ack, maps:size(AwaitingAck)},
{awaiting_comp, maps:size(AwaitingComp)}, {awaiting_comp, maps:size(AwaitingComp)},