From f8614196ac1e69ba0e2ca3eba571eb47829ecddb Mon Sep 17 00:00:00 2001 From: JianBo He Date: Fri, 2 Sep 2022 14:50:17 +0800 Subject: [PATCH] test: ensure udp client keepalive value getting right value --- .../src/bhvrs/emqx_gateway_conn.erl | 19 ++++--- .../src/exproto/emqx_exproto_channel.erl | 9 ++- apps/emqx_gateway/test/emqx_exproto_SUITE.erl | 56 ++++++++++++++----- 3 files changed, 59 insertions(+), 25 deletions(-) diff --git a/apps/emqx_gateway/src/bhvrs/emqx_gateway_conn.erl b/apps/emqx_gateway/src/bhvrs/emqx_gateway_conn.erl index 786142cdd..24d63f02c 100644 --- a/apps/emqx_gateway/src/bhvrs/emqx_gateway_conn.erl +++ b/apps/emqx_gateway/src/bhvrs/emqx_gateway_conn.erl @@ -19,6 +19,7 @@ -include_lib("emqx/include/types.hrl"). -include_lib("emqx/include/logger.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). %% API -export([ @@ -51,6 +52,9 @@ %% Internal callback -export([wakeup_from_hib/2, recvloop/2]). +%% for channel module +-export([keepalive_stats/1]). + -record(state, { %% TCP/SSL/UDP/DTLS Wrapped Socket socket :: {esockd_transport, esockd:socket()} | {udp, _, _}, @@ -573,9 +577,15 @@ terminate( channel = Channel } ) -> - ?SLOG(debug, #{msg => "conn_process_terminated", reason => Reason}), _ = ChannMod:terminate(Reason, Channel), _ = close_socket(State), + ClientId = + try ChannMod:info(clientid, Channel) of + Id -> Id + catch + _:_ -> undefined + end, + ?tp(debug, conn_process_terminated, #{reason => Reason, clientid => ClientId}), exit(Reason). %%-------------------------------------------------------------------- @@ -655,12 +665,7 @@ handle_timeout( disconnected -> {ok, State}; _ -> - case keepalive_stats(Stat) of - {ok, Oct} -> - handle_timeout(TRef, {Keepalive, Oct}, State); - {error, Reason} -> - handle_info({sock_error, Reason}, State) - end + handle_timeout(TRef, {Keepalive, keepalive_stats(Stat)}, State) end; handle_timeout( _TRef, diff --git a/apps/emqx_gateway/src/exproto/emqx_exproto_channel.erl b/apps/emqx_gateway/src/exproto/emqx_exproto_channel.erl index 3380f35be..94ca031aa 100644 --- a/apps/emqx_gateway/src/exproto/emqx_exproto_channel.erl +++ b/apps/emqx_gateway/src/exproto/emqx_exproto_channel.erl @@ -84,8 +84,6 @@ -define(INFO_KEYS, [conninfo, conn_state, clientinfo, session, will_msg]). --define(DEFAULT_IDLE_TIMEOUT, 30000). - %%-------------------------------------------------------------------- %% Info, Attrs and Caps %%-------------------------------------------------------------------- @@ -154,7 +152,7 @@ init( Ctx = maps:get(ctx, Options), GRpcChann = maps:get(handler, Options), PoolName = maps:get(pool_name, Options), - IdleTimeout = proplists:get_value(idle_timeout, Options, ?DEFAULT_IDLE_TIMEOUT), + IdleTimeout = emqx_gateway_utils:idle_timeout(Options), NConnInfo = default_conninfo(ConnInfo#{idle_timeout => IdleTimeout}), ListenerId = @@ -301,7 +299,7 @@ handle_timeout( Req = #{type => 'KEEPALIVE'}, NChannel = clean_timer(alive_timer, Channel), %% close connection if keepalive timeout - Replies = [{event, disconnected}, {close, normal}], + Replies = [{event, disconnected}, {close, keepalive_timeout}], {ok, Replies, try_dispatch(on_timer_timeout, wrap(Req), NChannel)} end; handle_timeout(_TRef, force_close, Channel = #channel{closed_reason = Reason}) -> @@ -667,7 +665,8 @@ ensure_keepalive(Channel = #channel{clientinfo = ClientInfo}) -> ensure_keepalive_timer(Interval, Channel) when Interval =< 0 -> Channel; ensure_keepalive_timer(Interval, Channel) -> - Keepalive = emqx_keepalive:init(timer:seconds(Interval)), + StatVal = emqx_gateway_conn:keepalive_stats(recv_oct), + Keepalive = emqx_keepalive:init(StatVal, timer:seconds(Interval)), ensure_timer(alive_timer, Channel#channel{keepalive = Keepalive}). ensure_timer(Name, Channel = #channel{timers = Timers}) -> diff --git a/apps/emqx_gateway/test/emqx_exproto_SUITE.erl b/apps/emqx_gateway/test/emqx_exproto_SUITE.erl index 65725fd19..73dc2ad45 100644 --- a/apps/emqx_gateway/test/emqx_exproto_SUITE.erl +++ b/apps/emqx_gateway/test/emqx_exproto_SUITE.erl @@ -20,6 +20,7 @@ -compile(nowarn_export_all). -include_lib("emqx/include/emqx_hooks.hrl"). +-include_lib("eunit/include/eunit.hrl"). -import( emqx_exproto_echo_svr, @@ -38,6 +39,7 @@ -include_lib("emqx/include/emqx.hrl"). -include_lib("emqx/include/emqx_mqtt.hrl"). +-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -define(TCPOPTS, [binary, {active, false}]). -define(DTLSOPTS, [binary, {active, false}, {protocol, dtls}]). @@ -223,14 +225,16 @@ t_acl_deny(Cfg) -> close(Sock). t_keepalive_timeout(Cfg) -> + ok = snabbkaffe:start_trace(), SockType = proplists:get_value(listener_type, Cfg), Sock = open(SockType), + ClientId1 = <<"keepalive_test_client1">>, Client = #{ proto_name => <<"demo">>, proto_ver => <<"v0.1">>, - clientid => <<"test_client_1">>, - keepalive => 2 + clientid => ClientId1, + keepalive => 5 }, Password = <<"123456">>, @@ -238,18 +242,41 @@ t_keepalive_timeout(Cfg) -> ConnAckBin = frame_connack(0), send(Sock, ConnBin), - {ok, ConnAckBin} = recv(Sock, 5000), + {ok, ConnAckBin} = recv(Sock), - %% Timed out connections are closed immediately, - %% so there may not be a disconnect message here - %%DisconnectBin = frame_disconnect(), - %%{ok, DisconnectBin} = recv(Sock, 10000), - - SockType =/= udp andalso - begin - {error, closed} = recv(Sock, 5000) - end, - ok. + case SockType of + udp -> + %% another udp client should not affect the first + %% udp client keepalive check + timer:sleep(4000), + Sock2 = open(SockType), + ConnBin2 = frame_connect( + Client#{clientid => <<"keepalive_test_client2">>}, + Password + ), + send(Sock2, ConnBin2), + %% first client will be keepalive timeouted in 6s + ?assertMatch( + {ok, #{ + clientid := ClientId1, + reason := {shutdown, {sock_closed, keepalive_timeout}} + }}, + ?block_until(#{?snk_kind := conn_process_terminated}, 8000) + ); + _ -> + ?assertMatch( + {ok, #{ + clientid := ClientId1, + reason := {shutdown, {sock_closed, keepalive_timeout}} + }}, + ?block_until(#{?snk_kind := conn_process_terminated}, 12000) + ), + Trace = snabbkaffe:collect_trace(), + %% conn process should be terminated + ?assertEqual(1, length(?of_kind(conn_process_terminated, Trace))), + %% socket port should be closed + ?assertEqual({error, closed}, recv(Sock, 5000)) + end. t_hook_connected_disconnected(Cfg) -> SockType = proplists:get_value(listener_type, Cfg), @@ -424,6 +451,9 @@ send({ssl, Sock}, Bin) -> send({dtls, Sock}, Bin) -> ssl:send(Sock, Bin). +recv(Sock) -> + recv(Sock, infinity). + recv({tcp, Sock}, Ts) -> gen_tcp:recv(Sock, 0, Ts); recv({udp, Sock}, Ts) ->