feat(sn): introduce subs_resume option

As the mqtt-sn v1.2 spec metioned, the gateway will be able to sync the
subscriptions topic-name registry to client when the client resume
it's session

port from: https://github.com/emqx/emqx/pull/7300

apps/emqx_gateway/src/emqx_gateway_schema.erl

@@ -142,6 +142,13 @@ which do not support any other features except this one.<br>
 There is no connection setup nor tear down, no registration nor subscription.<br>
 The client just sends its 'PUBLISH' messages to a GW"
            })}
+    , {subs_resume,
+       sc(boolean(),
+          #{ default => false
+           , desc =>
+"Whether to initiate all subscribed topic name registration messages to the
+client after the Session has been taken over by a new channel."
+           })}
     , {predefined,
        sc(hoconsc:array(ref(mqttsn_predefined)),
           #{ default => []

apps/emqx_gateway/src/mqttsn/emqx_sn_channel.erl

@@ -114,7 +114,12 @@
 
 -define(NEG_QOS_CLIENT_ID, <<"NegQoS-Client">>).
 
--define(REGISTER_TIMEOUT, 10000). % 10s
+-define(REGISTER_INFLIGHT(TopicId, TopicName),
+        #channel{register_inflight = {TopicId, _, TopicName}}).
+
+-define(MAX_RETRY_TIMES, 3).
+
+-define(REGISTER_TIMEOUT, 5000). % 5s
 -define(DEFAULT_SESSION_EXPIRY, 7200000). %% 2h
 
 %%--------------------------------------------------------------------
@@ -546,10 +551,8 @@ handle_in(?SN_REGISTER_MSG(_TopicId, MsgId, TopicName),
             {ok, {outgoing, AckPacket}, Channel}
     end;
 
-handle_in(?SN_REGACK_MSG(TopicId, MsgId, ?SN_RC_ACCEPTED),
+handle_in(?SN_REGACK_MSG(TopicId, _MsgId, ?SN_RC_ACCEPTED),
-          Channel = #channel{register_inflight = Inflight}) ->
+          Channel = ?REGISTER_INFLIGHT(TopicId, TopicName)) ->
-    case Inflight of
-        {TopicId, _, TopicName} ->
     ?SLOG(debug, #{ msg => "register_topic_name_to_client_succesfully"
                   , topic_id => TopicId
                   , topic_name => TopicName
@@ -558,16 +561,9 @@ handle_in(?SN_REGACK_MSG(TopicId, MsgId, ?SN_RC_ACCEPTED),
                  register_timer,
                  Channel#channel{register_inflight = undefined}),
     send_next_register_or_replay_publish(TopicName, NChannel);
-        _ ->
-            ?SLOG(error, #{ msg => "unexpected_regack_msg"
-                          , msg_id => MsgId
-                          , topic_id => TopicId
-                          , current_inflight => Inflight
-                          }),
-            {ok, Channel}
-    end;
 
-handle_in(?SN_REGACK_MSG(_TopicId, _MsgId, Reason), Channel) ->
+handle_in(?SN_REGACK_MSG(TopicId, _MsgId, Reason),
+          Channel = ?REGISTER_INFLIGHT(TopicId, TopicName)) ->
     case Reason of
         ?SN_RC_CONGESTION ->
             %% TODO: a or b?
@@ -575,11 +571,28 @@ handle_in(?SN_REGACK_MSG(_TopicId, _MsgId, Reason), Channel) ->
             %% b. re-new the re-transmit timer
             {ok, Channel};
         _ ->
-            %% disconnect this client, if the reason is
+            %% skipp this topic-name register, if the reason is
             %% ?SN_RC_NOT_SUPPORTED, ?SN_RC_INVALID_TOPIC_ID, etc.
-            handle_out(disconnect, ?SN_RC_NOT_SUPPORTED, Channel)
+            ?SLOG(warning, #{ msg => "skipp_register_topic_name_to_client"
+                            , topic_id => TopicId
+                            , topic_name => TopicName
+                            }),
+            NChannel = cancel_timer(
+                         register_timer,
+                         Channel#channel{register_inflight = undefined}),
+            send_next_register_or_replay_publish(TopicName, NChannel)
     end;
 
+handle_in(?SN_REGACK_MSG(TopicId, MsgId, Reason),
+          Channel = #channel{register_inflight = Inflight}) ->
+    ?SLOG(error, #{ msg => "unexpected_regack_msg"
+                  , acked_msg_id => MsgId
+                  , acked_topic_id => TopicId
+                  , acked_reason => Reason
+                  , current_inflight => Inflight
+                  }),
+    {ok, Channel};
+
 handle_in(PubPkt = ?SN_PUBLISH_MSG(_Flags, TopicId0, MsgId, _Data), Channel) ->
     TopicId = case is_integer(TopicId0) of
                   true -> TopicId0;
@@ -642,7 +655,7 @@ handle_in(?SN_PUBACK_MSG(TopicId, MsgId, ReturnCode),
                     %% involving the predefined topic name in register to
                     %% enhance the gateway's robustness even inconsistent
                     %% with MQTT-SN channels
-                    handle_out(register, {TopicId, MsgId, TopicName}, Channel)
+                    handle_out(register, {TopicId, TopicName}, Channel)
             end;
         _ ->
             ?SLOG(error, #{ msg => "cannt_handle_PUBACK"
@@ -779,12 +792,10 @@ handle_in(?SN_PINGREQ_MSG(ClientId),
     awake(ClientId, Channel);
 
 handle_in(?SN_PINGREQ_MSG(ClientId),
-          Channel = #channel{conn_state = ConnState}) ->
+          Channel = #channel{
-    ?SLOG(error, #{ msg => "awake_pingreq_in_bad_conn_state"
+                       conn_state = connected,
-                  , conn_state => ConnState
+                       clientinfo = #{clientid := ClientId}}) ->
-                  , clientid => ClientId
+    {ok, {outgoing, ?SN_PINGRESP_MSG()}, Channel};
-                  }),
-    handle_out(disconnect, protocol_error, Channel);
 
 handle_in(?SN_DISCONNECT_MSG(_Duration = undefined), Channel) ->
     handle_out(disconnect, normal, Channel);
@@ -833,30 +844,18 @@ after_message_acked(ClientInfo, Msg, #channel{ctx = Ctx}) ->
 outgoing_and_update(Pkt) ->
     [{outgoing, Pkt}, {event, update}].
 
-send_next_register_or_replay_publish(TopicName,
+send_next_register_or_replay_publish(
-                                     Channel = #channel{
+  _TopicName,
-                                                  session = Session,
+  Channel = #channel{ register_awaiting_queue = []}) ->
-                                                  register_awaiting_queue = []}) ->
+    {Outgoing, NChannel} = resume_or_replay_messages(Channel),
-    case emqx_inflight:to_list(emqx_session:info(inflight, Session)) of
+    {ok, Outgoing, NChannel};
-        [] -> {ok, Channel};
+
-        [{PktId, {inflight_data, _, Msg, _}}] ->
+send_next_register_or_replay_publish(
-            case TopicName =:= emqx_message:topic(Msg) of
+  _TopicName,
-                false ->
+  Channel = #channel{register_awaiting_queue = RAQueue}) ->
-                    ?SLOG(warning, #{ msg => "replay_inflight_message_failed"
-                                    , acked_topic_name => TopicName
-                                    , inflight_message => Msg
-                                    }),
-                    {ok, Channel};
-                true ->
-                    NMsg = emqx_message:set_flag(dup, true, Msg),
-                    handle_out(publish, {PktId, NMsg}, Channel)
-            end
-    end;
-send_next_register_or_replay_publish(_TopicName,
-                                     Channel = #channel{
-                                                  register_awaiting_queue = RAQueue}) ->
     [RegisterReq | NRAQueue] = RAQueue,
-    handle_out(register, RegisterReq, Channel#channel{register_awaiting_queue = NRAQueue}).
+    handle_out(register, RegisterReq,
+               Channel#channel{register_awaiting_queue = NRAQueue}).
 
 %%--------------------------------------------------------------------
 %% Handle Publish
@@ -1263,36 +1262,32 @@ handle_out(pubrel, MsgId, Channel) ->
 handle_out(pubcomp, MsgId, Channel) ->
     {ok, {outgoing, ?SN_PUBREC_MSG(?SN_PUBCOMP, MsgId)}, Channel};
 
-handle_out(register, {TopicId, MsgId, TopicName},
+handle_out(register, {TopicId, TopicName},
-           Channel = #channel{register_inflight = undefined}) ->
+           Channel = #channel{session = Session,
+                              register_inflight = undefined}) ->
+    {MsgId, NSession} = emqx_session:obtain_next_pkt_id(Session),
     Outgoing = {outgoing, ?SN_REGISTER_MSG(TopicId, MsgId, TopicName)},
-    NChannel = Channel#channel{register_inflight = {TopicId, MsgId, TopicName}},
+    NChannel = Channel#channel{
-    {ok, Outgoing, ensure_timer(register_timer, ?REGISTER_TIMEOUT, NChannel)};
+                 session = NSession,
+                 register_inflight = {TopicId, MsgId, TopicName}},
+    {ok, Outgoing, ensure_register_timer(NChannel)};
 
-handle_out(register, {TopicId, MsgId, TopicName},
+handle_out(register, {TopicId, TopicName},
            Channel = #channel{register_inflight = Inflight,
                               register_awaiting_queue = RAQueue}) ->
-    case Inflight of
+    case enqueue_register_request({TopicId, TopicName}, Inflight, RAQueue) of
-        {_, _, TopicName} ->
+        ignore ->
-            ?SLOG(debug, #{ msg => "ingore_handle_out_register"
+            ?SLOG(debug, #{ msg => "ingore_register_request_to_client"
-                          , requested_register_msg =>
+                          , register_request =>
                              #{ topic_id => TopicId
-                              , msg_id => MsgId
                               , topic_name => TopicName
                               }
                           }),
             {ok, Channel};
-        {InflightTopicId, InflightMsgId, InflightTopicName} ->
+        NRAQueue ->
-            NRAQueue = RAQueue ++ [{TopicId, MsgId, TopicName}],
             ?SLOG(debug, #{ msg => "put_register_msg_into_awaiting_queue"
-                          , inflight_register_msg =>
+                          , register_request =>
-                             #{ topic_id => InflightTopicId
-                              , msg_id => InflightMsgId
-                              , topic_name => InflightTopicName
-                              }
-                          , queued_register_msg =>
                              #{ topic_id => TopicId
-                              , msg_id => MsgId
                               , topic_name => TopicName
                               }
                           , register_awaiting_queue_size => length(NRAQueue)
@@ -1302,7 +1297,20 @@ handle_out(register, {TopicId, MsgId, TopicName},
 
 handle_out(disconnect, RC, Channel) ->
     DisPkt = ?SN_DISCONNECT_MSG(undefined),
-    {ok, [{outgoing, DisPkt}, {close, RC}], Channel}.
+    Reason = case is_atom(RC) of
+                 true -> RC;
+                 false -> returncode_name(RC)
+             end,
+    {ok, [{outgoing, DisPkt}, {close, Reason}], Channel}.
+
+enqueue_register_request({_, TopicName}, {_, _, TopicName}, _RAQueue) ->
+    ignore;
+enqueue_register_request({TopicId, TopicName}, _, RAQueue) ->
+    HasQueued = lists:any(fun({_, T}) -> T == TopicName end, RAQueue),
+    case HasQueued of
+        true -> ignore;
+        false -> RAQueue ++ [{TopicId, TopicName}]
+    end.
 
 %%--------------------------------------------------------------------
 %% Return ConnAck
@@ -1310,35 +1318,53 @@ handle_out(disconnect, RC, Channel) ->
 
 return_connack(AckPacket, Channel) ->
     Replies1 = [{event, connected}, {outgoing, AckPacket}],
-    case maybe_resume_session(Channel) of
+    {Replies2, NChannel} = maybe_resume_session(Channel),
-        ignore -> {ok, Replies1, Channel};
+    {ok, Replies1 ++ Replies2, NChannel}.
-        {ok, Publishes, NSession} ->
-            NChannel = Channel#channel{session  = NSession,
-                                       resuming = false,
-                                       pendings = []
-                                      },
-            {Replies2, NChannel1} = outgoing_deliver_and_register(
-                                      do_deliver(Publishes, NChannel)
-                                     ),
-            {ok, Replies1 ++ Replies2, NChannel1}
-    end.
 
 %%--------------------------------------------------------------------
 %% Maybe Resume Session
 
-maybe_resume_session(#channel{resuming = false}) ->
+maybe_resume_session(Channel = #channel{resuming = false}) ->
-    ignore;
+    {[], Channel};
-maybe_resume_session(#channel{session  = Session,
+maybe_resume_session(Channel = #channel{session  = Session,
-                              resuming = true,
+                                        resuming = true}) ->
-                              pendings = Pendings, clientinfo = ClientInfo}) ->
+    Subs = emqx_session:info(subscriptions, Session),
-    {ok, Publishes, Session1} = emqx_session:replay(ClientInfo, Session),
+    case subs_resume() andalso map_size(Subs) =/= 0 of
-    case emqx_session:deliver(ClientInfo, Pendings, Session1) of
+        true ->
-        {ok, Session2} ->
+            TopicNames = lists:filter(fun(T) -> not emqx_topic:wildcard(T)
-            {ok, Publishes, Session2};
+                                      end, maps:keys(Subs)),
-        {ok, More, Session2} ->
+            Registers = lists:map(fun(T) -> {register, T} end, TopicNames),
-            {ok, lists:append(Publishes, More), Session2}
+            {Registers, Channel};
+        false ->
+            resume_or_replay_messages(Channel)
     end.
 
+resume_or_replay_messages(Channel = #channel{
+                                       resuming = Resuming,
+                                       pendings = Pendings,
+                                       session = Session,
+                                       clientinfo = ClientInfo}) ->
+    {NPendings, NChannel} =
+        case Resuming of
+            true ->
+                {Pendings, Channel#channel{resuming = false, pendings = []}};
+            false ->
+                {[], Channel}
+        end,
+    {ok, Publishes, Session1} = emqx_session:replay(ClientInfo, Session),
+    {NPublishes, NSession} =
+        case emqx_session:deliver(ClientInfo, NPendings, Session1) of
+            {ok, Session2} ->
+                {Publishes, Session2};
+            {ok, More, Session2} ->
+                {lists:append(Publishes, More), Session2}
+        end,
+    outgoing_deliver_and_register(
+      do_deliver(NPublishes, NChannel#channel{session = NSession})).
+
+subs_resume() ->
+    emqx:get_config([gateway, mqttsn, subs_resume]).
+
 %%--------------------------------------------------------------------
 %% Deliver publish: broker -> client
 %%--------------------------------------------------------------------
@@ -1539,32 +1565,16 @@ handle_info(clean_authz_cache, Channel) ->
 handle_info({subscribe, _}, Channel) ->
    {ok, Channel};
 
-handle_info({register, TopicName},
+handle_info({register, TopicName}, Channel) ->
-            Channel = #channel{
+    case ensure_registered_topic_name(TopicName, Channel) of
-                         registry = Registry,
-                         session = Session}) ->
-    ClientId = clientid(Channel),
-    case emqx_sn_registry:lookup_topic_id(Registry,  ClientId, TopicName) of
-        undefined ->
-            case emqx_sn_registry:register_topic(Registry, ClientId, TopicName) of
         {error, Reason} ->
             ?SLOG(error, #{ msg => "register_topic_failed"
                           , topic_name => TopicName
                           , reason => Reason
                           }),
             {ok, Channel};
-                TopicId ->
+        {ok, TopicId} ->
-                    {MsgId, NSession} = emqx_session:obtain_next_pkt_id(Session),
+            handle_out(register, {TopicId, TopicName}, Channel)
-                    handle_out(
-                      register,
-                      {TopicId, MsgId, TopicName},
-                      Channel#channel{session = NSession})
-            end;
-        Registered ->
-            ?SLOG(debug, #{ msg => "ignore_register_request"
-                          , registered_as => Registered
-                          }),
-            {ok, Channel}
     end;
 
 handle_info(Info, Channel) ->
@@ -1581,6 +1591,19 @@ maybe_shutdown(Reason, Channel = #channel{conninfo = ConnInfo}) ->
         _ -> shutdown(Reason, Channel)
     end.
 
+ensure_registered_topic_name(TopicName,
+                             Channel = #channel{registry = Registry}) ->
+    ClientId = clientid(Channel),
+    case emqx_sn_registry:lookup_topic_id(Registry, ClientId, TopicName) of
+        undefined ->
+            case emqx_sn_registry:register_topic(Registry, ClientId, TopicName) of
+                {error, Reason} -> {error, Reason};
+                TopicId -> {ok, TopicId}
+            end;
+        TopicId ->
+            {ok, TopicId}
+    end.
+
 %%--------------------------------------------------------------------
 %% Ensure disconnected
 
@@ -1623,12 +1646,17 @@ handle_deliver(Delivers, Channel = #channel{
                 ),
     {ok, Channel#channel{session = NSession}};
 
+%% There are two secensar need to cache delivering messages:
+%%  1. it is being takeover by other channel
+%%  2. it is being resume registered topic-names
 handle_deliver(Delivers, Channel = #channel{
                                       ctx = Ctx,
-                                      takeover = true,
+                                      takeover = Takeover,
                                       pendings = Pendings,
                                       session = Session,
-                                      clientinfo = #{clientid := ClientId}}) ->
+                                      resuming = Resuming,
+                                      clientinfo = #{clientid := ClientId}})
+  when Takeover == true; Resuming == true ->
     NPendings = lists:append(
                   Pendings,
                   ignore_local(maybe_nack(Delivers), ClientId, Session, Ctx)
@@ -1680,7 +1708,6 @@ not_nacked({deliver, _Topic, Msg}) ->
       -> {ok, channel()}
        | {ok, replies(), channel()}
        | {shutdown, Reason :: term(), channel()}.
-
 handle_timeout(_TRef, {keepalive, _StatVal},
                Channel = #channel{keepalive = undefined}) ->
     {ok, Channel};
@@ -1731,6 +1758,23 @@ handle_timeout(_TRef, expire_awaiting_rel,
             {ok, reset_timer(await_timer, Timeout, Channel#channel{session = NSession})}
     end;
 
+handle_timeout(_TRef, {retry_register, RetryTimes},
+               Channel = #channel{register_inflight = {TopicId, MsgId, TopicName}}) ->
+    case RetryTimes < ?MAX_RETRY_TIMES of
+        true ->
+            Outgoing = {outgoing, ?SN_REGISTER_MSG(TopicId, MsgId, TopicName)},
+            {ok, Outgoing, ensure_register_timer(RetryTimes + 1, Channel)};
+        false ->
+            ?SLOG(error, #{ msg => "register_topic_reached_max_retry_times"
+                          , register_request =>
+                             #{ topic_id => TopicId
+                              , msg_id => MsgId
+                              , topic_name => TopicName
+                              }
+                          }),
+            handle_out(disconnect, ?SN_RC2_REACHED_MAX_RETRY, Channel)
+    end;
+
 handle_timeout(_TRef, expire_session, Channel) ->
     shutdown(expired, Channel);
 
@@ -1788,6 +1832,14 @@ ensure_asleep_timer(Durtion, Channel) ->
     ensure_timer(asleep_timer, timer:seconds(Durtion),
                  Channel#channel{asleep_timer_duration = Durtion}).
 
+ensure_register_timer(Channel) ->
+    ensure_register_timer(0, Channel).
+
+ensure_register_timer(RetryTimes, Channel = #channel{timers = Timers}) ->
+    Msg = maps:get(register_timer, ?TIMER_TABLE),
+    TRef = emqx_misc:start_timer(?REGISTER_TIMEOUT, {Msg, RetryTimes}),
+    Channel#channel{timers = Timers#{register_timer => TRef}}.
+
 cancel_timer(Name, Channel = #channel{timers = Timers}) ->
     case maps:get(Name, Timers, undefined) of
         undefined ->
@@ -1858,4 +1910,5 @@ returncode_name(?SN_RC2_NOT_AUTHORIZE) -> rejected_not_authorize;
 returncode_name(?SN_RC2_FAILED_SESSION) -> rejected_failed_open_session;
 returncode_name(?SN_RC2_KEEPALIVE_TIMEOUT) -> rejected_keepalive_timeout;
 returncode_name(?SN_RC2_EXCEED_LIMITATION) -> rejected_exceed_limitation;
+returncode_name(?SN_RC2_REACHED_MAX_RETRY) -> reached_max_retry_times;
 returncode_name(_) -> accepted.

apps/emqx_gateway/src/mqttsn/include/emqx_sn.hrl

@@ -58,6 +58,7 @@
 -define(SN_RC2_FAILED_SESSION,    16#81).
 -define(SN_RC2_KEEPALIVE_TIMEOUT, 16#82).
 -define(SN_RC2_EXCEED_LIMITATION, 16#83).
+-define(SN_RC2_REACHED_MAX_RETRY, 16#84).
 
 -define(QOS_NEG1, 3).
 

apps/emqx_gateway/test/emqx_sn_protocol_SUITE.erl

@@ -89,12 +89,22 @@ all() ->
 
 init_per_suite(Config) ->
     ok = emqx_common_test_helpers:load_config(emqx_gateway_schema, ?CONF_DEFAULT),
-    emqx_mgmt_api_test_util:init_suite([emqx_gateway]),
+    emqx_mgmt_api_test_util:init_suite([emqx_conf, emqx_gateway]),
     Config.
 
 end_per_suite(_) ->
     {ok, _} = emqx:remove_config([gateway, mqttsn]),
-    emqx_mgmt_api_test_util:end_suite([emqx_gateway]).
+    emqx_mgmt_api_test_util:end_suite([emqx_gateway, emqx_conf]).
+
+restart_mqttsn_with_subs_resume_on() ->
+    emqx_gateway_conf:update_gateway(
+      mqttsn,
+      #{<<"subs_resume">> => <<"true">>}).
+
+restart_mqttsn_with_subs_resume_off() ->
+    emqx_gateway_conf:update_gateway(
+      mqttsn,
+      #{<<"subs_resume">> => <<"false">>}).
 
 %%--------------------------------------------------------------------
 %% Test cases
@@ -992,8 +1002,8 @@ t_delivery_qos1_register_invalid_topic_id(_) ->
     %% acked with ?SN_RC_INVALID_TOPIC_ID to
     send_puback_msg(Socket, TopicId, MsgId, ?SN_RC_INVALID_TOPIC_ID),
 
-    ?assertEqual(
+    ?assertMatch(
-       {TopicId, MsgId},
+       {TopicId, _},
        check_register_msg_on_udp(<<"ab">>, receive_response(Socket))),
     send_regack_msg(Socket, TopicId, MsgId + 1),
 
@@ -1008,111 +1018,6 @@ t_delivery_qos1_register_invalid_topic_id(_) ->
     ?assertEqual(<<2, ?SN_DISCONNECT>>, receive_response(Socket)),
     gen_udp:close(Socket).
 
-t_delivery_takeover_and_re_register(_) ->
-    MsgId = 1,
-    {ok, Socket} = gen_udp:open(0, [binary]),
-    send_connect_msg(Socket, <<"test">>, 0),
-    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
-                 receive_response(Socket)),
-
-    send_subscribe_msg_normal_topic(Socket, ?QOS_1, <<"topic-a">>, MsgId+1),
-    <<_, ?SN_SUBACK, 2#00100000,
-      TopicIdA:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
-
-    send_subscribe_msg_normal_topic(Socket, ?QOS_2, <<"topic-b">>, MsgId+2),
-    <<_, ?SN_SUBACK, 2#01000000,
-      TopicIdB:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
-
-    _ = emqx:publish(
-          emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"test-a">>)),
-    _ = emqx:publish(
-          emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"test-b">>)),
-
-    <<_, ?SN_PUBLISH, 2#00100000,
-      TopicIdA:16, MsgId1:16, "test-a">> = receive_response(Socket),
-    send_puback_msg(Socket, TopicIdA, MsgId1, ?SN_RC_ACCEPTED),
-
-    <<_, ?SN_PUBLISH, 2#01000000,
-      TopicIdB:16, MsgId2:16, "test-b">> = receive_response(Socket),
-    send_puback_msg(Socket, TopicIdB, MsgId2, ?SN_RC_ACCEPTED),
-
-    send_disconnect_msg(Socket, undefined),
-    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(Socket)),
-    gen_udp:close(Socket),
-
-    %% offline messages will be queued into the MQTT-SN session
-    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"m1">>)),
-    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"m2">>)),
-    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"m3">>)),
-    _ = emqx:publish(emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"m1">>)),
-    _ = emqx:publish(emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"m2">>)),
-    _ = emqx:publish(emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"m3">>)),
-
-    {ok, NSocket} = gen_udp:open(0, [binary]),
-    send_connect_msg(NSocket, <<"test">>, 0),
-    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
-                 receive_response(NSocket)),
-
-    %% qos1
-
-    %% received the resume messages
-    <<_, ?SN_PUBLISH, 2#00100000,
-      TopicIdA:16, MsgIdA0:16, "m1">> = receive_response(NSocket),
-    %% only one qos1/qos2 inflight
-    ?assertEqual(udp_receive_timeout, receive_response(NSocket)),
-    send_puback_msg(NSocket, TopicIdA, MsgIdA0, ?SN_RC_INVALID_TOPIC_ID),
-    %% recv register
-    <<_, ?SN_REGISTER,
-      TopicIdA:16, RegMsgIdA:16, "topic-a">> = receive_response(NSocket),
-    send_regack_msg(NSocket, TopicIdA, RegMsgIdA),
-    %% received the replay messages
-    <<_, ?SN_PUBLISH, 2#00100000,
-      TopicIdA:16, MsgIdA1:16, "m1">> = receive_response(NSocket),
-    send_puback_msg(NSocket, TopicIdA, MsgIdA1, ?SN_RC_ACCEPTED),
-
-    <<_, ?SN_PUBLISH, 2#00100000,
-      TopicIdA:16, MsgIdA2:16, "m2">> = receive_response(NSocket),
-    send_puback_msg(NSocket, TopicIdA, MsgIdA2, ?SN_RC_ACCEPTED),
-
-    <<_, ?SN_PUBLISH, 2#00100000,
-      TopicIdA:16, MsgIdA3:16, "m3">> = receive_response(NSocket),
-    send_puback_msg(NSocket, TopicIdA, MsgIdA3, ?SN_RC_ACCEPTED),
-
-    %% qos2
-    <<_, ?SN_PUBLISH, 2#01000000,
-      TopicIdB:16, MsgIdB0:16, "m1">> = receive_response(NSocket),
-    %% only one qos1/qos2 inflight
-    ?assertEqual(udp_receive_timeout, receive_response(NSocket)),
-    send_puback_msg(NSocket, TopicIdB, MsgIdB0, ?SN_RC_INVALID_TOPIC_ID),
-    %% recv register
-    <<_, ?SN_REGISTER,
-      TopicIdB:16, RegMsgIdB:16, "topic-b">> = receive_response(NSocket),
-    send_regack_msg(NSocket, TopicIdB, RegMsgIdB),
-    %% received the replay messages
-    <<_, ?SN_PUBLISH, 2#01000000,
-      TopicIdB:16, MsgIdB1:16, "m1">> = receive_response(NSocket),
-    send_pubrec_msg(NSocket, MsgIdB1),
-    <<_, ?SN_PUBREL, MsgIdB1:16>> = receive_response(NSocket),
-    send_pubcomp_msg(NSocket, MsgIdB1),
-
-    <<_, ?SN_PUBLISH, 2#01000000,
-      TopicIdB:16, MsgIdB2:16, "m2">> = receive_response(NSocket),
-    send_puback_msg(NSocket, TopicIdB, MsgIdB2, ?SN_RC_ACCEPTED),
-
-    <<_, ?SN_PUBLISH, 2#01000000,
-      TopicIdB:16, MsgIdB3:16, "m3">> = receive_response(NSocket),
-    send_puback_msg(NSocket, TopicIdB, MsgIdB3, ?SN_RC_ACCEPTED),
-
-    %% no more messages
-    ?assertEqual(udp_receive_timeout, receive_response(NSocket)),
-
-    send_disconnect_msg(NSocket, undefined),
-    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(NSocket)),
-
-    _ = emqx_gateway_cm:kick_session(mqttsn, <<"test">>),
-
-    gen_udp:close(NSocket).
-
 t_will_case01(_) ->
     QoS = 1,
     Duration = 1,
@@ -1924,6 +1829,326 @@ t_broadcast_test1(_) ->
     timer:sleep(600),
     gen_udp:close(Socket).
 
+t_register_subs_resume_on(_) ->
+    restart_mqttsn_with_subs_resume_on(),
+    MsgId = 1,
+    {ok, Socket} = gen_udp:open(0, [binary]),
+    send_connect_msg(Socket, <<"test">>, 0),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(Socket)),
+
+    send_subscribe_msg_normal_topic(Socket, ?QOS_2, <<"topic-a">>, MsgId+1),
+    <<_, ?SN_SUBACK, 2#01000000,
+      TopicIdA:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
+
+    send_subscribe_msg_normal_topic(Socket, ?QOS_2, <<"topic-b">>, MsgId+2),
+    <<_, ?SN_SUBACK, 2#01000000,
+      TopicIdB:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
+
+    _ = emqx:publish(
+          emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"test-a">>)),
+    _ = emqx:publish(
+          emqx_message:make(test, ?QOS_1, <<"topic-b">>, <<"test-b">>)),
+
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdA:16, MsgId1:16, "test-a">> = receive_response(Socket),
+    send_puback_msg(Socket, TopicIdA, MsgId1, ?SN_RC_ACCEPTED),
+
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdB:16, MsgId2:16, "test-b">> = receive_response(Socket),
+    send_puback_msg(Socket, TopicIdB, MsgId2, ?SN_RC_ACCEPTED),
+
+    send_disconnect_msg(Socket, undefined),
+    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(Socket)),
+    gen_udp:close(Socket),
+
+    %% offline messages will be queued into the MQTT-SN session
+    _ = emqx:publish(emqx_message:make(test, ?QOS_0, <<"topic-a">>, <<"m1">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"m2">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_2, <<"topic-a">>, <<"m3">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_0, <<"topic-b">>, <<"m1">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-b">>, <<"m2">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"m3">>)),
+
+    emqx_logger:set_log_level(debug),
+
+    {ok, NSocket} = gen_udp:open(0, [binary]),
+    send_connect_msg(NSocket, <<"test">>, 0),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(NSocket)),
+
+    %% receive subs register requests
+    <<_, ?SN_REGISTER,
+      TopicIdA:16, RegMsgIdA:16, "topic-a">> = receive_response(NSocket),
+    send_regack_msg(NSocket, TopicIdA, RegMsgIdA),
+
+    <<_, ?SN_REGISTER,
+      TopicIdB:16, RegMsgIdB:16, "topic-b">> = receive_response(NSocket),
+    send_regack_msg(NSocket, TopicIdB, RegMsgIdB),
+
+    %% receive the queued messages
+
+    <<_, ?SN_PUBLISH, 2#00000000,
+      TopicIdA:16, 0:16, "m1">> = receive_response(NSocket),
+
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdA:16, MsgIdA1:16, "m2">> = receive_response(NSocket),
+    send_puback_msg(NSocket, TopicIdA, MsgIdA1, ?SN_RC_ACCEPTED),
+
+    <<_, ?SN_PUBLISH, 2#01000000,
+      TopicIdA:16, MsgIdA2:16, "m3">> = receive_response(NSocket),
+    send_pubrec_msg(NSocket, MsgIdA2),
+    <<_, ?SN_PUBREL, MsgIdA2:16>> = receive_response(NSocket),
+    send_pubcomp_msg(NSocket, MsgIdA2),
+
+
+    <<_, ?SN_PUBLISH, 2#00000000,
+      TopicIdB:16, 0:16, "m1">> = receive_response(NSocket),
+
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdB:16, MsgIdB1:16, "m2">> = receive_response(NSocket),
+    send_puback_msg(NSocket, TopicIdB, MsgIdB1, ?SN_RC_ACCEPTED),
+
+    <<_, ?SN_PUBLISH, 2#01000000,
+      TopicIdB:16, MsgIdB2:16, "m3">> = receive_response(NSocket),
+    send_pubrec_msg(NSocket, MsgIdB2),
+    <<_, ?SN_PUBREL, MsgIdB2:16>> = receive_response(NSocket),
+    send_pubcomp_msg(NSocket, MsgIdB2),
+
+    %% no more messages
+    ?assertEqual(udp_receive_timeout, receive_response(NSocket)),
+
+    gen_udp:close(NSocket),
+    {ok, NSocket1} = gen_udp:open(0, [binary]),
+    send_connect_msg(NSocket1, <<"test">>),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(NSocket1)),
+    send_disconnect_msg(NSocket1, undefined),
+    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(NSocket1)),
+    gen_udp:close(NSocket1),
+    restart_mqttsn_with_subs_resume_off().
+
+t_register_subs_resume_off(_) ->
+    MsgId = 1,
+    {ok, Socket} = gen_udp:open(0, [binary]),
+    send_connect_msg(Socket, <<"test">>, 0),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(Socket)),
+
+    send_subscribe_msg_normal_topic(Socket, ?QOS_1, <<"topic-a">>, MsgId+1),
+    <<_, ?SN_SUBACK, 2#00100000,
+      TopicIdA:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
+
+    send_subscribe_msg_normal_topic(Socket, ?QOS_2, <<"topic-b">>, MsgId+2),
+    <<_, ?SN_SUBACK, 2#01000000,
+      TopicIdB:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
+
+    _ = emqx:publish(
+          emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"test-a">>)),
+    _ = emqx:publish(
+          emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"test-b">>)),
+
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdA:16, MsgId1:16, "test-a">> = receive_response(Socket),
+    send_puback_msg(Socket, TopicIdA, MsgId1, ?SN_RC_ACCEPTED),
+
+    <<_, ?SN_PUBLISH, 2#01000000,
+      TopicIdB:16, MsgId2:16, "test-b">> = receive_response(Socket),
+    send_puback_msg(Socket, TopicIdB, MsgId2, ?SN_RC_ACCEPTED),
+
+    send_disconnect_msg(Socket, undefined),
+    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(Socket)),
+    gen_udp:close(Socket),
+
+    %% offline messages will be queued into the MQTT-SN session
+    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"m1">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"m2">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"m3">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"m1">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"m2">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_2, <<"topic-b">>, <<"m3">>)),
+
+    {ok, NSocket} = gen_udp:open(0, [binary]),
+    send_connect_msg(NSocket, <<"test">>, 0),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(NSocket)),
+
+    %% qos1
+
+    %% received the resume messages
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdA:16, MsgIdA0:16, "m1">> = receive_response(NSocket),
+    %% only one qos1/qos2 inflight
+    ?assertEqual(udp_receive_timeout, receive_response(NSocket)),
+    send_puback_msg(NSocket, TopicIdA, MsgIdA0, ?SN_RC_INVALID_TOPIC_ID),
+    %% recv register
+    <<_, ?SN_REGISTER,
+      TopicIdA:16, RegMsgIdA:16, "topic-a">> = receive_response(NSocket),
+    send_regack_msg(NSocket, TopicIdA, RegMsgIdA),
+    %% received the replay messages
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdA:16, MsgIdA1:16, "m1">> = receive_response(NSocket),
+    send_puback_msg(NSocket, TopicIdA, MsgIdA1, ?SN_RC_ACCEPTED),
+
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdA:16, MsgIdA2:16, "m2">> = receive_response(NSocket),
+    send_puback_msg(NSocket, TopicIdA, MsgIdA2, ?SN_RC_ACCEPTED),
+
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdA:16, MsgIdA3:16, "m3">> = receive_response(NSocket),
+    send_puback_msg(NSocket, TopicIdA, MsgIdA3, ?SN_RC_ACCEPTED),
+
+    %% qos2
+    <<_, ?SN_PUBLISH, 2#01000000,
+      TopicIdB:16, MsgIdB0:16, "m1">> = receive_response(NSocket),
+    %% only one qos1/qos2 inflight
+    ?assertEqual(udp_receive_timeout, receive_response(NSocket)),
+    send_puback_msg(NSocket, TopicIdB, MsgIdB0, ?SN_RC_INVALID_TOPIC_ID),
+    %% recv register
+    <<_, ?SN_REGISTER,
+      TopicIdB:16, RegMsgIdB:16, "topic-b">> = receive_response(NSocket),
+    send_regack_msg(NSocket, TopicIdB, RegMsgIdB),
+    %% received the replay messages
+    <<_, ?SN_PUBLISH, 2#01000000,
+      TopicIdB:16, MsgIdB1:16, "m1">> = receive_response(NSocket),
+    send_pubrec_msg(NSocket, MsgIdB1),
+    <<_, ?SN_PUBREL, MsgIdB1:16>> = receive_response(NSocket),
+    send_pubcomp_msg(NSocket, MsgIdB1),
+
+    <<_, ?SN_PUBLISH, 2#01000000,
+      TopicIdB:16, MsgIdB2:16, "m2">> = receive_response(NSocket),
+    send_puback_msg(NSocket, TopicIdB, MsgIdB2, ?SN_RC_ACCEPTED),
+
+    <<_, ?SN_PUBLISH, 2#01000000,
+      TopicIdB:16, MsgIdB3:16, "m3">> = receive_response(NSocket),
+    send_puback_msg(NSocket, TopicIdB, MsgIdB3, ?SN_RC_ACCEPTED),
+
+    %% no more messages
+    ?assertEqual(udp_receive_timeout, receive_response(NSocket)),
+
+    gen_udp:close(NSocket),
+    {ok, NSocket1} = gen_udp:open(0, [binary]),
+    send_connect_msg(NSocket1, <<"test">>),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(NSocket1)),
+    send_disconnect_msg(NSocket1, undefined),
+    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(NSocket1)),
+    gen_udp:close(NSocket1).
+
+t_register_skip_failure_topic_name_and_reach_max_retry_times(_) ->
+    restart_mqttsn_with_subs_resume_on(),
+    MsgId = 1,
+    {ok, Socket} = gen_udp:open(0, [binary]),
+    send_connect_msg(Socket, <<"test">>, 0),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(Socket)),
+
+    send_subscribe_msg_normal_topic(Socket, ?QOS_2, <<"topic-a">>, MsgId+1),
+    <<_, ?SN_SUBACK, 2#01000000,
+      TopicIdA:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
+
+    send_subscribe_msg_normal_topic(Socket, ?QOS_2, <<"topic-b">>, MsgId+2),
+    <<_, ?SN_SUBACK, 2#01000000,
+      TopicIdB:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
+
+    send_disconnect_msg(Socket, undefined),
+    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(Socket)),
+    gen_udp:close(Socket),
+
+    emqx_logger:set_log_level(debug),
+
+    {ok, NSocket} = gen_udp:open(0, [binary]),
+    send_connect_msg(NSocket, <<"test">>, 0),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(NSocket)),
+
+    %% receive subs register requests
+
+    %% registered failured topic-name will be skipped
+    <<_, ?SN_REGISTER,
+      TopicIdA:16, RegMsgIdA:16, "topic-a">> = receive_response(NSocket),
+    send_regack_msg(NSocket, TopicIdA, RegMsgIdA, ?SN_RC_INVALID_TOPIC_ID),
+
+    %% the gateway try to shutdown this client if it reached max-retry-times
+    %%
+    %% times-0
+    <<_, ?SN_REGISTER,
+      TopicIdB:16, RegMsgIdB:16, "topic-b">> = receive_response(NSocket),
+    %% times-1
+    timer:sleep(5000), %% RETYRY_TIMEOUT
+    <<_, ?SN_REGISTER,
+      TopicIdB:16, RegMsgIdB:16, "topic-b">> = receive_response(NSocket),
+    %% times-2
+    timer:sleep(5000), %% RETYRY_TIMEOUT
+    <<_, ?SN_REGISTER,
+      TopicIdB:16, RegMsgIdB:16, "topic-b">> = receive_response(NSocket),
+    %% just a ping
+    send_pingreq_msg(NSocket, <<"test">>),
+    ?assertEqual(<<2, ?SN_PINGRESP>>, receive_response(NSocket)),
+    %% times-3
+    timer:sleep(5000), %% RETYRY_TIMEOUT
+    <<_, ?SN_REGISTER,
+      TopicIdB:16, RegMsgIdB:16, "topic-b">> = receive_response(NSocket),
+    %% shutdown due to reached max retry times
+    timer:sleep(5000), %% RETYRY_TIMEOUT
+    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(NSocket)),
+    gen_udp:close(NSocket),
+    restart_mqttsn_with_subs_resume_off().
+
+t_register_enqueue_delivering_messages(_) ->
+    restart_mqttsn_with_subs_resume_on(),
+    MsgId = 1,
+    {ok, Socket} = gen_udp:open(0, [binary]),
+    send_connect_msg(Socket, <<"test">>, 0),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(Socket)),
+
+    send_subscribe_msg_normal_topic(Socket, ?QOS_2, <<"topic-a">>, MsgId+1),
+    <<_, ?SN_SUBACK, 2#01000000,
+      TopicIdA:16, _:16, ?SN_RC_ACCEPTED>> = receive_response(Socket),
+
+    send_disconnect_msg(Socket, undefined),
+    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(Socket)),
+    gen_udp:close(Socket),
+
+    {ok, NSocket} = gen_udp:open(0, [binary]),
+    send_connect_msg(NSocket, <<"test">>, 0),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(NSocket)),
+
+    %% receive subs register requests
+
+    %% registered failured topic-name will be skipped
+    <<_, ?SN_REGISTER,
+      TopicIdA:16, RegMsgIdA:16, "topic-a">> = receive_response(NSocket),
+
+    _ = emqx:publish(emqx_message:make(test, ?QOS_0, <<"topic-a">>, <<"m1">>)),
+    _ = emqx:publish(emqx_message:make(test, ?QOS_1, <<"topic-a">>, <<"m2">>)),
+
+    send_regack_msg(NSocket, TopicIdA, RegMsgIdA, ?SN_RC_ACCEPTED),
+
+    %% receive the queued messages
+
+    <<_, ?SN_PUBLISH, 2#00000000,
+      TopicIdA:16, 0:16, "m1">> = receive_response(NSocket),
+
+    <<_, ?SN_PUBLISH, 2#00100000,
+      TopicIdA:16, MsgIdA1:16, "m2">> = receive_response(NSocket),
+    send_puback_msg(NSocket, TopicIdA, MsgIdA1, ?SN_RC_ACCEPTED),
+
+    %% no more messages
+    ?assertEqual(udp_receive_timeout, receive_response(NSocket)),
+
+    gen_udp:close(NSocket),
+    {ok, NSocket1} = gen_udp:open(0, [binary]),
+    send_connect_msg(NSocket1, <<"test">>),
+    ?assertMatch(<<_, ?SN_CONNACK, ?SN_RC_ACCEPTED>>,
+                 receive_response(NSocket1)),
+    send_disconnect_msg(NSocket1, undefined),
+    ?assertMatch(<<2, ?SN_DISCONNECT>>, receive_response(NSocket1)),
+    gen_udp:close(NSocket1),
+    restart_mqttsn_with_subs_resume_off().
+
 t_socket_passvice(_) ->
     %% TODO: test this gateway enter the passvie event
     ok.
@@ -2082,9 +2307,12 @@ send_register_msg(Socket, TopicName, MsgId) ->
     ok = gen_udp:send(Socket, ?HOST, ?PORT, RegisterPacket).
 
 send_regack_msg(Socket, TopicId, MsgId) ->
+    send_regack_msg(Socket, TopicId, MsgId, ?SN_RC_ACCEPTED).
+
+send_regack_msg(Socket, TopicId, MsgId, Rc) ->
     Length = 7,
     MsgType = ?SN_REGACK,
-    Packet = <<Length:8, MsgType:8, TopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>,
+    Packet = <<Length:8, MsgType:8, TopicId:16, MsgId:16, Rc>>,
     ok = gen_udp:send(Socket, ?HOST, ?PORT, Packet).
 
 send_publish_msg_normal_topic(Socket, QoS, MsgId, TopicId, Data) ->