Merge branch 'develop' into remove-protocol-module
This commit is contained in:
Feng Lee
GitHub
commit
1a3261b186
|
|
@ -334,30 +334,26 @@ handle_in(Packet = ?UNSUBSCRIBE_PACKET(PacketId, Properties, TopicFilters),
|
|||
handle_in(?PACKET(?PINGREQ), Channel) ->
|
||||
{ok, ?PACKET(?PINGRESP), Channel};
|
||||
|
||||
handle_in(?DISCONNECT_PACKET(RC, Props),
|
||||
Channel = #channel{conninfo = ConnInfo = #{expiry_interval := OldInterval}}) ->
|
||||
handle_in(?DISCONNECT_PACKET(ReasonCode, Properties), Channel = #channel{session = Session,
|
||||
conninfo = ConnInfo = #{expiry_interval := OldInterval}}) ->
|
||||
OldInterval = emqx_session:info(expiry_interval, Session),
|
||||
Interval = emqx_mqtt_props:get('Session-Expiry-Interval', Props, OldInterval),
|
||||
case OldInterval =:= 0 andalso Interval =/= OldInterval of
|
||||
true ->
|
||||
handle_out({disconnect, ?RC_PROTOCOL_ERROR}, Channel);
|
||||
false ->
|
||||
Channel1 = case RC of
|
||||
?RC_SUCCESS -> Channel#channel{will_msg = undefined};
|
||||
_ -> Channel
|
||||
end,
|
||||
Channel2 = Channel1#channel{conninfo = ConnInfo#{expiry_interval => Interval}},
|
||||
case Interval of
|
||||
?UINT_MAX ->
|
||||
{ok, ensure_timer(will_timer, Channel2)};
|
||||
Int when Int > 0 ->
|
||||
{ok, ensure_timer([will_timer, expire_timer], Channel2)};
|
||||
_Other ->
|
||||
Reason = case RC of
|
||||
?RC_SUCCESS -> normal;
|
||||
_ -> emqx_reason_codes:name(RC, maps:get(proto_ver, ConnInfo))
|
||||
end,
|
||||
{stop, {shutdown, Reason}, Channel2}
|
||||
end
|
||||
Reason = case ReasonCode of
|
||||
?RC_SUCCESS -> normal;
|
||||
_ ->
|
||||
ProtoVer = emqx_protocol:info(proto_ver, Protocol),
|
||||
emqx_reason_codes:name(ReasonCode, ProtoVer)
|
||||
end,
|
||||
{wait_session_expire, {shutdown, Reason},
|
||||
Channel#channel{session = emqx_session:update_expiry_interval(Interval, Session),
|
||||
protocol = case ReasonCode of
|
||||
?RC_SUCCESS -> emqx_protocol:clear_will_msg(Protocol);
|
||||
_ -> Protocol
|
||||
end}}
|
||||
end;
|
||||
|
||||
handle_in(?AUTH_PACKET(), Channel) ->
|
||||
|
|
@ -366,7 +362,7 @@ handle_in(?AUTH_PACKET(), Channel) ->
|
|||
|
||||
handle_in(Packet, Channel) ->
|
||||
?LOG(error, "Unexpected incoming: ~p", [Packet]),
|
||||
{stop, {shutdown, unexpected_incoming_packet}, Channel}.
|
||||
handle_out({disconnect, ?RC_MALFORMED_PACKET}, Channel).
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
%% Process Connect
|
||||
|
|
@ -562,9 +558,6 @@ handle_out({deliver, Delivers}, Channel = #channel{session = Session}) ->
|
|||
{ok, Channel#channel{session = NSession}}
|
||||
end;
|
||||
|
||||
handle_out({publish, [Publish]}, Channel) ->
|
||||
handle_out(Publish, Channel);
|
||||
|
||||
handle_out({publish, Publishes}, Channel) when is_list(Publishes) ->
|
||||
Packets = lists:foldl(
|
||||
fun(Publish, Acc) ->
|
||||
|
|
@ -621,10 +614,10 @@ handle_out({disconnect, ReasonCode}, Channel = #channel{conninfo = ConnInfo}) ->
|
|||
?MQTT_PROTO_V5 ->
|
||||
Reason = emqx_reason_codes:name(ReasonCode),
|
||||
Packet = ?DISCONNECT_PACKET(ReasonCode),
|
||||
{stop, {shutdown, Reason}, Packet, Channel};
|
||||
{wait_session_expire, {shutdown, Reason}, Packet, Channel};
|
||||
ProtoVer ->
|
||||
Reason = emqx_reason_codes:name(ReasonCode, ProtoVer),
|
||||
{stop, {shutdown, Reason}, Channel}
|
||||
{wait_session_expire, {shutdown, Reason}, Channel}
|
||||
end;
|
||||
|
||||
handle_out({Type, Data}, Channel) ->
|
||||
|
|
@ -696,17 +689,29 @@ handle_info({unsubscribe, TopicFilters}, Channel = #channel{client = ClientInfo}
|
|||
handle_info(disconnected, Channel = #channel{connected = undefined}) ->
|
||||
shutdown(closed, Channel);
|
||||
|
||||
handle_info(disconnected, Channel = #channel{conninfo = #{expiry_interval := Interval},
|
||||
handle_info(disconnected, Channel = #channel{connected = false}) ->
|
||||
{ok, Channel};
|
||||
|
||||
handle_info(disconnected, Channel = #channel{conninfo = #{expiry_interval := ExpiryInterval},
|
||||
client = ClientInfo = #{zone := Zone},
|
||||
session = Session,
|
||||
will_msg = WillMsg}) ->
|
||||
%% TODO: Why handle will_msg here?
|
||||
publish_will_msg(WillMsg),
|
||||
NChannel = Channel#channel{will_msg = undefined},
|
||||
case Interval of
|
||||
emqx_zone:enable_flapping_detect(Zone) andalso emqx_flapping:detect(ClientInfo),
|
||||
Channel1 = ensure_disconnected(Channel),
|
||||
Channel2 = case timer:seconds(will_delay_interval(WillMsg)) of
|
||||
0 ->
|
||||
publish_will_msg(WillMsg),
|
||||
Channel1#channel{will_msg = undefined};
|
||||
_ ->
|
||||
ensure_timer(will_timer, Channel1)
|
||||
end,
|
||||
case ExpiryInterval of
|
||||
?UINT_MAX ->
|
||||
{ok, ensure_disconnected(NChannel)};
|
||||
{ok, Channel2};
|
||||
Int when Int > 0 ->
|
||||
{ok, ensure_timer(expire_timer, ensure_disconnected(NChannel))};
|
||||
_Other -> shutdown(closed, NChannel)
|
||||
{ok, ensure_timer(expire_timer, Channel2)};
|
||||
_Other ->
|
||||
shutdown(closed, Channel2)
|
||||
end;
|
||||
|
||||
handle_info(Info, Channel) ->
|
||||
|
|
@ -735,7 +740,7 @@ handle_timeout(TRef, {keepalive, StatVal},
|
|||
NChannel = Channel#channel{keepalive = NKeepalive},
|
||||
{ok, reset_timer(alive_timer, NChannel)};
|
||||
{error, timeout} ->
|
||||
{stop, {shutdown, keepalive_timeout}, Channel}
|
||||
{wait_session_expire, {shutdown, keepalive_timeout}, Channel}
|
||||
end;
|
||||
|
||||
handle_timeout(TRef, retry_delivery,
|
||||
|
|
@ -818,7 +823,11 @@ interval(expire_timer, #channel{conninfo = ConnInfo}) ->
|
|||
timer:seconds(maps:get(expiry_interval, ConnInfo));
|
||||
interval(will_timer, #channel{will_msg = WillMsg}) ->
|
||||
%% TODO: Ensure the header exists.
|
||||
timer:seconds(emqx_message:get_header('Will-Delay-Interval', WillMsg)).
|
||||
timer:seconds(will_delay_interval(WillMsg)).
|
||||
|
||||
will_delay_interval(undefined) -> 0;
|
||||
will_delay_interval(WillMsg) ->
|
||||
emqx_message:get_header('Will-Delay-Interval', WillMsg, 0).
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
%% Terminate
|
||||
|
|
@ -826,9 +835,16 @@ interval(will_timer, #channel{will_msg = WillMsg}) ->
|
|||
|
||||
terminate(normal, #channel{conninfo = ConnInfo, client = ClientInfo}) ->
|
||||
ok = emqx_hooks:run('client.disconnected', [ClientInfo, normal, ConnInfo]);
|
||||
terminate({shutdown, Reason}, #channel{conninfo = ConnInfo, client = ClientInfo,})
|
||||
when Reason =:= kicked orelse Reason =:= discarded orelse Reason =:= takeovered ->
|
||||
ok = emqx_hooks:run('client.disconnected', [ClientInfo, Reason, ConnInfo]);
|
||||
terminate(Reason, #channel{conninfo = ConnInfo, client = ClientInfo, will_msg = WillMsg}) ->
|
||||
publish_will_msg(WillMsg),
|
||||
ok = emqx_hooks:run('client.disconnected', [ClientInfo, Reason, ConnInfo]).
|
||||
if
|
||||
Protocol == undefined -> ok;
|
||||
true -> publish_will_msg(emqx_protocol:info(will_msg, Protocol))
|
||||
end.
|
||||
|
||||
-spec(received(pos_integer(), channel()) -> channel()).
|
||||
received(Oct, Channel) ->
|
||||
|
|
|
|||
|
|
@ -224,10 +224,13 @@ idle(cast, {incoming, Packet = ?CONNECT_PACKET(ConnPkt)}, State) ->
|
|||
SuccFun = fun(NewSt) -> {next_state, connected, NewSt} end,
|
||||
handle_incoming(Packet, SuccFun, NState);
|
||||
|
||||
idle(cast, {incoming, Packet}, State) ->
|
||||
idle(cast, {incoming, Packet}, State) when is_record(Packet, mqtt_packet) ->
|
||||
?LOG(warning, "Unexpected incoming: ~p", [Packet]),
|
||||
shutdown(unexpected_incoming_packet, State);
|
||||
|
||||
idle(cast, {incoming, {error, Reason}}, State) ->
|
||||
shutdown(Reason, State);
|
||||
|
||||
idle(EventType, Content, State) ->
|
||||
?HANDLE(EventType, Content, State).
|
||||
|
||||
|
|
@ -241,6 +244,17 @@ connected(enter, _PrevSt, State) ->
|
|||
connected(cast, {incoming, Packet}, State) when is_record(Packet, mqtt_packet) ->
|
||||
handle_incoming(Packet, fun keep_state/1, State);
|
||||
|
||||
connected(cast, {incoming, {error, Reason}}, State = #connection{chan_state = ChanState}) ->
|
||||
case emqx_channel:handle_out({disconnect, emqx_reason_codes:mqtt_frame_error(Reason)}, ChanState) of
|
||||
{wait_session_expire, _, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
{next_state, disconnected, State#connection{chan_state= NChanState}};
|
||||
{wait_session_expire, _, OutPackets, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
NState = State#connection{chan_state= NChanState},
|
||||
{next_state, disconnected, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState)}
|
||||
end;
|
||||
|
||||
connected(info, Deliver = {deliver, _Topic, _Msg}, State) ->
|
||||
handle_deliver(emqx_misc:drain_deliver([Deliver]), State);
|
||||
|
||||
|
|
@ -408,8 +422,7 @@ process_incoming(Data, State) ->
|
|||
process_incoming(<<>>, Packets, State) ->
|
||||
{keep_state, State, next_incoming_events(Packets)};
|
||||
|
||||
process_incoming(Data, Packets, State = #connection{parse_state = ParseState,
|
||||
chan_state = ChanState}) ->
|
||||
process_incoming(Data, Packets, State = #connection{parse_state = ParseState}) ->
|
||||
try emqx_frame:parse(Data, ParseState) of
|
||||
{more, NParseState} ->
|
||||
NState = State#connection{parse_state = NParseState},
|
||||
|
|
@ -418,32 +431,16 @@ process_incoming(Data, Packets, State = #connection{parse_state = ParseState,
|
|||
NState = State#connection{parse_state = NParseState},
|
||||
process_incoming(Rest, [Packet|Packets], NState);
|
||||
{error, Reason} ->
|
||||
shutdown(Reason, State)
|
||||
{keep_state, State, next_incoming_events({error, Reason})}
|
||||
catch
|
||||
error:Reason:Stk ->
|
||||
?LOG(error, "Parse failed for ~p~nStacktrace:~p~nError data:~p", [Reason, Stk, Data]),
|
||||
Result =
|
||||
case emqx_channel:info(connected, ChanState) of
|
||||
undefined ->
|
||||
emqx_channel:handle_out({connack, emqx_reason_codes:mqtt_frame_error(Reason)}, ChanState);
|
||||
true ->
|
||||
emqx_channel:handle_out({disconnect, emqx_reason_codes:mqtt_frame_error(Reason)}, ChanState);
|
||||
_ ->
|
||||
ignore
|
||||
end,
|
||||
case Result of
|
||||
{stop, Reason0, OutPackets, NChanState} ->
|
||||
Shutdown = fun(NewSt) -> stop(Reason0, NewSt) end,
|
||||
NState = State#connection{chan_state = NChanState},
|
||||
handle_outgoing(OutPackets, Shutdown, NState);
|
||||
{stop, Reason0, NChanState} ->
|
||||
stop(Reason0, State#connection{chan_state = NChanState});
|
||||
ignore ->
|
||||
keep_state(State)
|
||||
end
|
||||
?LOG(error, "~nParse failed for ~p~nStacktrace: ~p~nError data:~p", [Reason, Stk, Data]),
|
||||
{keep_state, State, next_incoming_events({error, Reason})}
|
||||
end.
|
||||
|
||||
-compile({inline, [next_incoming_events/1]}).
|
||||
next_incoming_events({error, Reason}) ->
|
||||
[next_event(cast, {incoming, {error, Reason}})];
|
||||
next_incoming_events(Packets) ->
|
||||
[next_event(cast, {incoming, Packet}) || Packet <- Packets].
|
||||
|
||||
|
|
@ -459,14 +456,19 @@ handle_incoming(Packet = ?PACKET(Type), SuccFun,
|
|||
{ok, NChanState} ->
|
||||
SuccFun(State#connection{chan_state= NChanState});
|
||||
{ok, OutPackets, NChanState} ->
|
||||
handle_outgoing(OutPackets, SuccFun,
|
||||
State#connection{chan_state = NChanState});
|
||||
handle_outgoing(OutPackets, SuccFun, State#connection{chan_state = NChanState});
|
||||
{wait_session_expire, Reason, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
{next_state, disconnected, State#connection{chan_state = NChanState}};
|
||||
{wait_session_expire, Reason, OutPackets, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
NState = State#connection{chan_state= NChanState},
|
||||
{next_state, disconnected, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState)};
|
||||
{stop, Reason, NChanState} ->
|
||||
stop(Reason, State#connection{chan_state = NChanState});
|
||||
{stop, Reason, OutPackets, NChanState} ->
|
||||
Shutdown = fun(NewSt) -> stop(Reason, NewSt) end,
|
||||
NState = State#connection{chan_state = NChanState},
|
||||
handle_outgoing(OutPackets, Shutdown, NState)
|
||||
NState = State#connection{chan_state= NChanState},
|
||||
stop(Reason, handle_outgoing(OutPackets, fun(NewSt) -> NewSt end, NState))
|
||||
end.
|
||||
|
||||
%%-------------------------------------------------------------------
|
||||
|
|
@ -477,10 +479,7 @@ handle_deliver(Delivers, State = #connection{chan_state = ChanState}) ->
|
|||
{ok, NChanState} ->
|
||||
keep_state(State#connection{chan_state = NChanState});
|
||||
{ok, Packets, NChanState} ->
|
||||
NState = State#connection{chan_state = NChanState},
|
||||
handle_outgoing(Packets, fun keep_state/1, NState);
|
||||
{stop, Reason, NChanState} ->
|
||||
stop(Reason, State#connection{chan_state = NChanState})
|
||||
handle_outgoing(Packets, fun keep_state/1, State#connection{chan_state = NChanState})
|
||||
end.
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
|
|
@ -534,8 +533,10 @@ handle_timeout(TRef, Msg, State = #connection{chan_state = ChanState}) ->
|
|||
{ok, NChanState} ->
|
||||
keep_state(State#connection{chan_state = NChanState});
|
||||
{ok, Packets, NChanState} ->
|
||||
handle_outgoing(Packets, fun keep_state/1,
|
||||
State#connection{chan_state = NChanState});
|
||||
handle_outgoing(Packets, fun keep_state/1, State#connection{chan_state = NChanState});
|
||||
{wait_session_expire, Reason, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
{next_state, disconnected, State#connection{chan_state = NChanState}};
|
||||
{stop, Reason, NChanState} ->
|
||||
stop(Reason, State#connection{chan_state = NChanState})
|
||||
end.
|
||||
|
|
|
|||
|
|
@ -141,11 +141,11 @@ handle_cast({detected, Flapping = #flapping{client_id = ClientId,
|
|||
%% Log first
|
||||
?LOG(error, "Flapping detected: ~s(~s) disconnected ~w times in ~wms",
|
||||
[ClientId, esockd_net:ntoa(PeerHost), DetectCnt, Duration]),
|
||||
%% TODO: Send Alarm
|
||||
%% Banned.
|
||||
BannedFlapping = Flapping#flapping{client_id = {banned, ClientId},
|
||||
banned_at = emqx_time:now_ms()
|
||||
},
|
||||
alarm_handler:set_alarm({{flapping_detected, ClientId}, BannedFlapping}),
|
||||
ets:insert(?FLAPPING_TAB, BannedFlapping);
|
||||
false ->
|
||||
?LOG(warning, "~s(~s) disconnected ~w times in ~wms",
|
||||
|
|
@ -189,9 +189,17 @@ with_flapping_tab(Fun, Args) ->
|
|||
end.
|
||||
|
||||
expire_flapping(NowTime, #{duration := Duration, banned_interval := Interval}) ->
|
||||
ets:select_delete(?FLAPPING_TAB,
|
||||
[{#flapping{started_at = '$1', banned_at = undefined, _ = '_'},
|
||||
[{'<', '$1', NowTime-Duration}], [true]},
|
||||
{#flapping{client_id = {banned, '_'}, banned_at = '$1', _ = '_'},
|
||||
[{'<', '$1', NowTime-Interval}], [true]}]).
|
||||
case ets:select(?FLAPPING_TAB,
|
||||
[{#flapping{started_at = '$1', banned_at = undefined, _ = '_'},
|
||||
[{'<', '$1', NowTime-Duration}], ['$_']},
|
||||
{#flapping{client_id = {banned, '_'}, banned_at = '$1', _ = '_'},
|
||||
[{'<', '$1', NowTime-Interval}], ['$_']}]) of
|
||||
[] -> ok;
|
||||
Flappings ->
|
||||
lists:foreach(fun(Flapping = #flapping{client_id = {banned, ClientId}}) ->
|
||||
ets:delete_object(?FLAPPING_TAB, Flapping),
|
||||
alarm_handler:clear_alarm({flapping_detected, ClientId});
|
||||
(_) -> ok
|
||||
end, Flappings)
|
||||
end.
|
||||
|
||||
|
|
|
|||
|
|
@ -110,6 +110,10 @@ check(#mqtt_packet{variable = SubPkt}) when is_record(SubPkt, mqtt_packet_subscr
|
|||
check(#mqtt_packet{variable = UnsubPkt}) when is_record(UnsubPkt, mqtt_packet_unsubscribe) ->
|
||||
check(UnsubPkt);
|
||||
|
||||
check(#mqtt_packet_publish{topic_name = <<>>, properties = #{'Topic-Alias':= _TopicAlias}}) ->
|
||||
ok;
|
||||
check(#mqtt_packet_publish{topic_name = <<>>, properties = #{}}) ->
|
||||
{error, ?RC_PROTOCOL_ERROR};
|
||||
check(#mqtt_packet_publish{topic_name = TopicName, properties = Props}) ->
|
||||
try emqx_topic:validate(name, TopicName) of
|
||||
true -> check_pub_props(Props)
|
||||
|
|
|
|||
|
|
@ -426,7 +426,13 @@ dequeue(Cnt, Msgs, Q) ->
|
|||
case emqx_mqueue:out(Q) of
|
||||
{empty, _Q} -> {Msgs, Q};
|
||||
{{value, Msg}, Q1} ->
|
||||
dequeue(Cnt-1, [Msg|Msgs], Q1)
|
||||
case emqx_message:is_expired(Msg) of
|
||||
true ->
|
||||
ok = emqx_metrics:inc('messages.expired'),
|
||||
dequeue(Cnt-1, Msgs, Q1);
|
||||
false ->
|
||||
dequeue(Cnt-1, [Msg|Msgs], Q1)
|
||||
end
|
||||
end.
|
||||
|
||||
batch_n(Inflight) ->
|
||||
|
|
|
|||
|
|
@ -166,7 +166,7 @@ code_change(_OldVsn, State, _Extra) ->
|
|||
|
||||
handle_partition_event({partition, {occurred, Node}}) ->
|
||||
alarm_handler:set_alarm({partitioned, Node});
|
||||
handle_partition_event({partition, {healed, Node}}) ->
|
||||
handle_partition_event({partition, {healed, _Node}}) ->
|
||||
alarm_handler:clear_alarm(partitioned).
|
||||
|
||||
suppress(Key, SuccFun, State = #{events := Events}) ->
|
||||
|
|
|
|||
|
|
@ -254,6 +254,22 @@ websocket_info({cast, Msg}, State = #ws_connection{chan_state = ChanState}) ->
|
|||
stop(Reason, State#ws_connection{chan_state = NChanState})
|
||||
end;
|
||||
|
||||
websocket_info({incoming, {error, Reason}}, State = #ws_connection{fsm_state = idle}) ->
|
||||
stop({shutdown, Reason}, State);
|
||||
|
||||
websocket_info({incoming, {error, Reason}}, State = #ws_connection{fsm_state = connected, chan_state = ChanState}) ->
|
||||
case emqx_channel:handle_out({disconnect, emqx_reason_codes:mqtt_frame_error(Reason)}, ChanState) of
|
||||
{wait_session_expire, _, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
disconnected(State#ws_connection{chan_state= NChanState});
|
||||
{wait_session_expire, _, OutPackets, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
disconnected(enqueue(OutPackets, State#ws_connection{chan_state = NChanState}))
|
||||
end;
|
||||
|
||||
websocket_info({incoming, {error, _Reason}}, State = #ws_connection{fsm_state = disconnected}) ->
|
||||
reply(State);
|
||||
|
||||
websocket_info({incoming, Packet = ?CONNECT_PACKET(ConnPkt)},
|
||||
State = #ws_connection{fsm_state = idle}) ->
|
||||
#mqtt_packet_connect{proto_ver = ProtoVer, properties = Properties} = ConnPkt,
|
||||
|
|
@ -276,9 +292,7 @@ websocket_info(Deliver = {deliver, _Topic, _Msg},
|
|||
{ok, NChanState} ->
|
||||
reply(State#ws_connection{chan_state = NChanState});
|
||||
{ok, Packets, NChanState} ->
|
||||
reply(enqueue(Packets, State#ws_connection{chan_state = NChanState}));
|
||||
{stop, Reason, NChanState} ->
|
||||
stop(Reason, State#ws_connection{chan_state = NChanState})
|
||||
reply(enqueue(Packets, State#ws_connection{chan_state = NChanState}))
|
||||
end;
|
||||
|
||||
websocket_info({timeout, TRef, keepalive}, State) when is_reference(TRef) ->
|
||||
|
|
@ -307,8 +321,7 @@ websocket_info(Info, State = #ws_connection{chan_state = ChanState}) ->
|
|||
|
||||
terminate(SockError, _Req, #ws_connection{chan_state = ChanState,
|
||||
stop_reason = Reason}) ->
|
||||
?LOG(debug, "Terminated for ~p, sockerror: ~p",
|
||||
[Reason, SockError]),
|
||||
?LOG(debug, "Terminated for ~p, sockerror: ~p", [Reason, SockError]),
|
||||
emqx_channel:terminate(Reason, ChanState).
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
|
|
@ -318,6 +331,12 @@ connected(State = #ws_connection{chan_state = ChanState}) ->
|
|||
ok = emqx_channel:handle_cast({register, attrs(State), stats(State)}, ChanState),
|
||||
reply(State#ws_connection{fsm_state = connected}).
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
%% Disconnected callback
|
||||
|
||||
disconnected(State) ->
|
||||
reply(State#ws_connection{fsm_state = disconnected}).
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
%% Handle timeout
|
||||
|
||||
|
|
@ -328,6 +347,9 @@ handle_timeout(TRef, Msg, State = #ws_connection{chan_state = ChanState}) ->
|
|||
{ok, Packets, NChanState} ->
|
||||
NState = State#ws_connection{chan_state = NChanState},
|
||||
reply(enqueue(Packets, NState));
|
||||
{wait_session_expire, Reason, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
disconnected(State#ws_connection{chan_state = NChanState});
|
||||
{stop, Reason, NChanState} ->
|
||||
stop(Reason, State#ws_connection{chan_state = NChanState})
|
||||
end.
|
||||
|
|
@ -347,29 +369,13 @@ process_incoming(Data, State = #ws_connection{parse_state = ParseState,
|
|||
self() ! {incoming, Packet},
|
||||
process_incoming(Rest, State#ws_connection{parse_state = NParseState});
|
||||
{error, Reason} ->
|
||||
?LOG(error, "Frame error: ~p", [Reason]),
|
||||
stop(Reason, State)
|
||||
self() ! {incoming, {error, Reason}},
|
||||
{ok, State}
|
||||
catch
|
||||
error:Reason:Stk ->
|
||||
?LOG(error, "Parse failed for ~p~nStacktrace:~p~nFrame data: ~p", [Reason, Stk, Data]),
|
||||
Result =
|
||||
case emqx_channel:info(connected, ChanState) of
|
||||
undefined ->
|
||||
emqx_channel:handle_out({connack, emqx_reason_codes:mqtt_frame_error(Reason)}, ChanState);
|
||||
true ->
|
||||
emqx_channel:handle_out({disconnect, emqx_reason_codes:mqtt_frame_error(Reason)}, ChanState);
|
||||
_ ->
|
||||
ignore
|
||||
end,
|
||||
case Result of
|
||||
{stop, Reason0, OutPackets, NChanState} ->
|
||||
NState = State#ws_connection{chan_state = NChanState},
|
||||
stop(Reason0, enqueue(OutPackets, NState));
|
||||
{stop, Reason0, NChanState} ->
|
||||
stop(Reason0, State#ws_connection{chan_state = NChanState});
|
||||
ignore ->
|
||||
{ok, State}
|
||||
end
|
||||
?LOG(error, "~nParse failed for ~p~nStacktrace: ~p~nFrame data: ~p", [Reason, Stk, Data]),
|
||||
self() ! {incoming, {error, Reason}},
|
||||
{ok, State}
|
||||
end.
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
|
|
@ -386,11 +392,17 @@ handle_incoming(Packet = ?PACKET(Type), SuccFun,
|
|||
{ok, OutPackets, NChanState} ->
|
||||
NState = State#ws_connection{chan_state= NChanState},
|
||||
SuccFun(enqueue(OutPackets, NState));
|
||||
{wait_session_expire, Reason, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
disconnected(State#ws_connection{chan_state = NChanState});
|
||||
{wait_session_expire, Reason, OutPackets, NChanState} ->
|
||||
?LOG(debug, "Disconnect and wait for session to expire due to ~p", [Reason]),
|
||||
disconnected(enqueue(OutPackets, State#ws_connection{chan_state = NChanState}));
|
||||
{stop, Reason, NChanState} ->
|
||||
stop(Reason, State#ws_connection{chan_state= NChanState});
|
||||
{stop, Reason, OutPacket, NChanState} ->
|
||||
stop(Reason, State#ws_connection{chan_state = NChanState});
|
||||
{stop, Reason, OutPackets, NChanState} ->
|
||||
NState = State#ws_connection{chan_state= NChanState},
|
||||
stop(Reason, enqueue(OutPacket, NState))
|
||||
stop(Reason, enqueue(OutPackets, NState))
|
||||
end.
|
||||
|
||||
%%--------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -50,6 +50,7 @@ t_emqx_pubsub_api(_) ->
|
|||
Payload = <<"Hello World">>,
|
||||
Topic1 = <<"mytopic1">>,
|
||||
emqx:subscribe(Topic, ClientId),
|
||||
ct:sleep(100),
|
||||
?assertEqual([Topic], emqx:topics()),
|
||||
?assertEqual([self()], emqx:subscribers(Topic)),
|
||||
?assertEqual([{Topic,#{qos => 0,subid => ClientId}}], emqx:subscriptions(self())),
|
||||
|
|
|
|||
|
|
@ -144,7 +144,7 @@ t_handle_pingreq(_) ->
|
|||
t_handle_disconnect(_) ->
|
||||
with_channel(
|
||||
fun(Channel) ->
|
||||
{stop, {shutdown, normal}, Channel1} = handle_in(?DISCONNECT_PACKET(?RC_SUCCESS), Channel),
|
||||
{wait_session_expire, {shutdown, normal}, Channel1} = handle_in(?DISCONNECT_PACKET(?RC_SUCCESS), Channel),
|
||||
?assertMatch(#{will_msg := undefined}, emqx_channel:info(protocol, Channel1))
|
||||
end).
|
||||
|
||||
|
|
|
|||
|
|
@ -97,6 +97,7 @@ t_cm(_) ->
|
|||
ClientId = <<"myclient">>,
|
||||
{ok, C} = emqtt:start_link([{client_id, ClientId}]),
|
||||
{ok, _} = emqtt:connect(C),
|
||||
ct:sleep(50),
|
||||
#{client := #{client_id := ClientId}} = emqx_cm:get_chan_attrs(ClientId),
|
||||
emqtt:subscribe(C, <<"mytopic">>, 0),
|
||||
ct:sleep(1200),
|
||||
|
|
|
|||
|
|
@ -22,22 +22,24 @@
|
|||
all() -> emqx_ct:all(?MODULE).
|
||||
|
||||
init_per_suite(Config) ->
|
||||
prepare_env(),
|
||||
emqx_ct_helpers:boot_modules(all),
|
||||
emqx_ct_helpers:start_apps([], fun set_special_configs/1),
|
||||
Config.
|
||||
|
||||
prepare_env() ->
|
||||
set_special_configs(emqx) ->
|
||||
emqx_zone:set_env(external, enable_flapping_detect, true),
|
||||
application:set_env(emqx, flapping_detect_policy,
|
||||
#{threshold => 3,
|
||||
duration => 100,
|
||||
banned_interval => 200
|
||||
}).
|
||||
});
|
||||
set_special_configs(_App) -> ok.
|
||||
|
||||
end_per_suite(_Config) ->
|
||||
emqx_ct_helpers:stop_apps([]),
|
||||
ok.
|
||||
|
||||
t_detect_check(_) ->
|
||||
{ok, _Pid} = emqx_flapping:start_link(),
|
||||
ClientInfo = #{zone => external,
|
||||
client_id => <<"clientid">>,
|
||||
peerhost => {127,0,0,1}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,92 @@
|
|||
%%--------------------------------------------------------------------
|
||||
%% Copyright (c) 2019 EMQ Technologies Co., Ltd. All Rights Reserved.
|
||||
%%
|
||||
%% Licensed under the Apache License, Version 2.0 (the "License");
|
||||
%% you may not use this file except in compliance with the License.
|
||||
%% You may obtain a copy of the License at
|
||||
%%
|
||||
%% http://www.apache.org/licenses/LICENSE-2.0
|
||||
%%
|
||||
%% Unless required by applicable law or agreed to in writing, software
|
||||
%% distributed under the License is distributed on an "AS IS" BASIS,
|
||||
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
%% See the License for the specific language governing permissions and
|
||||
%% limitations under the License.
|
||||
%%--------------------------------------------------------------------
|
||||
|
||||
-module(emqx_msg_expiry_interval_SUITE).
|
||||
|
||||
-compile(export_all).
|
||||
-compile(nowarn_export_all).
|
||||
|
||||
-include_lib("eunit/include/eunit.hrl").
|
||||
|
||||
all() -> emqx_ct:all(?MODULE).
|
||||
|
||||
init_per_suite(Config) ->
|
||||
emqx_ct_helpers:boot_modules(all),
|
||||
emqx_ct_helpers:start_apps([]),
|
||||
Config.
|
||||
|
||||
end_per_suite(_Config) ->
|
||||
emqx_ct_helpers:stop_apps([]).
|
||||
|
||||
t_message_expiry_interval_1(_) ->
|
||||
ClientA = message_expiry_interval_init(),
|
||||
[message_expiry_interval_exipred(ClientA, QoS) || QoS <- [0,1,2]].
|
||||
|
||||
t_message_expiry_interval_2(_) ->
|
||||
ClientA = message_expiry_interval_init(),
|
||||
[message_expiry_interval_not_exipred(ClientA, QoS) || QoS <- [0,1,2]].
|
||||
|
||||
message_expiry_interval_init() ->
|
||||
{ok, ClientA} = emqtt:start_link([{proto_ver,v5}, {client_id, <<"client-a">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, ClientB} = emqtt:start_link([{proto_ver,v5}, {client_id, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, _} = emqtt:connect(ClientA),
|
||||
{ok, _} = emqtt:connect(ClientB),
|
||||
%% subscribe and disconnect client-b
|
||||
emqtt:subscribe(ClientB, <<"t/a">>, 1),
|
||||
emqtt:stop(ClientB),
|
||||
ClientA.
|
||||
|
||||
message_expiry_interval_exipred(ClientA, QoS) ->
|
||||
ct:pal("~p ~p", [?FUNCTION_NAME, QoS]),
|
||||
%% publish to t/a and waiting for the message expired
|
||||
emqtt:publish(ClientA, <<"t/a">>, #{'Message-Expiry-Interval' => 1}, <<"this will be purged in 1s">>, [{qos, QoS}]),
|
||||
ct:sleep(1000),
|
||||
|
||||
%% resume the session for client-b
|
||||
{ok, ClientB1} = emqtt:start_link([{proto_ver,v5}, {client_id, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, _} = emqtt:connect(ClientB1),
|
||||
|
||||
%% verify client-b could not receive the publish message
|
||||
receive
|
||||
{publish,#{client_pid := ClientB1, topic := <<"t/a">>}} ->
|
||||
ct:fail(should_have_expired)
|
||||
after 300 ->
|
||||
ok
|
||||
end,
|
||||
emqtt:stop(ClientB1).
|
||||
|
||||
message_expiry_interval_not_exipred(ClientA, QoS) ->
|
||||
ct:pal("~p ~p", [?FUNCTION_NAME, QoS]),
|
||||
%% publish to t/a
|
||||
emqtt:publish(ClientA, <<"t/a">>, #{'Message-Expiry-Interval' => 20}, <<"this will be purged in 1s">>, [{qos, QoS}]),
|
||||
|
||||
%% wait for 1s and then resume the session for client-b, the message should not expires
|
||||
%% as Message-Expiry-Interval = 20s
|
||||
ct:sleep(1000),
|
||||
{ok, ClientB1} = emqtt:start_link([{proto_ver,v5}, {client_id, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, _} = emqtt:connect(ClientB1),
|
||||
|
||||
%% verify client-b could receive the publish message and the Message-Expiry-Interval is set
|
||||
receive
|
||||
{publish,#{client_pid := ClientB1, topic := <<"t/a">>,
|
||||
properties := #{'Message-Expiry-Interval' := MsgExpItvl}}}
|
||||
when MsgExpItvl < 20 -> ok;
|
||||
{publish, _} = Msg ->
|
||||
ct:fail({incorrect_publish, Msg})
|
||||
after 300 ->
|
||||
ct:fail(no_publish_received)
|
||||
end,
|
||||
emqtt:stop(ClientB1).
|
||||
Loading…
Reference in New Issue