feat: add shutdown count

Fixes https://emqx.atlassian.net/browse/EMQX-9008
Fixes https://emqx.atlassian.net/browse/EMQX-9219

apps/emqx/src/emqx_listeners.erl

@@ -35,7 +35,8 @@
     current_conns/2,
     max_conns/2,
     id_example/0,
-    default_max_conn/0
+    default_max_conn/0,
+    shutdown_count/2
 ]).
 
 -export([
@@ -195,6 +196,17 @@ max_conns(Type, Name, _ListenOn) when Type =:= ws; Type =:= wss ->
 max_conns(_, _, _) ->
     {error, not_support}.
 
+shutdown_count(ID, ListenOn) ->
+    {ok, #{type := Type, name := Name}} = parse_listener_id(ID),
+    shutdown_count(Type, Name, ListenOn).
+
+shutdown_count(Type, Name, ListenOn) when Type == tcp; Type == ssl ->
+    esockd:get_shutdown_count({listener_id(Type, Name), ListenOn});
+shutdown_count(Type, _Name, _ListenOn) when Type =:= ws; Type =:= wss ->
+    [];
+shutdown_count(_, _, _) ->
+    {error, not_support}.
+
 %% @doc Start all listeners.
 -spec start() -> ok.
 start() ->

apps/emqx/test/emqx_client_SUITE.erl

@@ -67,7 +67,8 @@ groups() ->
             %% t_keepalive,
             %% t_redelivery_on_reconnect,
             %% subscribe_failure_test,
-            t_dollar_topics
+            t_dollar_topics,
+            t_sub_non_utf8_topic
         ]},
         {mqttv5, [non_parallel_tests], [t_basic_with_props_v5]},
         {others, [non_parallel_tests], [
@@ -297,6 +298,36 @@ t_dollar_topics(_) ->
     ok = emqtt:disconnect(C),
     ct:pal("$ topics test succeeded").
 
+t_sub_non_utf8_topic(_) ->
+    {ok, Socket} = gen_tcp:connect({127, 0, 0, 1}, 1883, [{active, true}, binary]),
+    ConnPacket = emqx_frame:serialize(#mqtt_packet{
+        header = #mqtt_packet_header{type = 1},
+        variable = #mqtt_packet_connect{
+            clientid = <<"abcdefg">>
+        }
+    }),
+    ok = gen_tcp:send(Socket, ConnPacket),
+    receive
+        {tcp, _, _ConnAck = <<32, 2, 0, 0>>} -> ok
+    after 3000 -> ct:fail({connect_ack_not_recv, process_info(self(), messages)})
+    end,
+    SubHeader = <<130, 18, 25, 178>>,
+    SubTopicLen = <<0, 13>>,
+    %% this is not a valid utf8 topic
+    SubTopic = <<128, 10, 10, 12, 178, 159, 162, 47, 115, 1, 1, 1, 1>>,
+    SubQoS = <<1>>,
+    SubPacket = <<SubHeader/binary, SubTopicLen/binary, SubTopic/binary, SubQoS/binary>>,
+    ok = gen_tcp:send(Socket, SubPacket),
+    receive
+        {tcp_closed, _} -> ok
+    after 3000 -> ct:fail({should_get_disconnected, process_info(self(), messages)})
+    end,
+    timer:sleep(1000),
+    ListenerCounts = emqx_listeners:shutdown_count('tcp:default', {{0, 0, 0, 0}, 1883}),
+    TopicInvalidCount = proplists:get_value(topic_filter_invalid, ListenerCounts),
+    ?assert(is_integer(TopicInvalidCount) andalso TopicInvalidCount > 0),
+    ok.
+
 %%--------------------------------------------------------------------
 %% Test cases for MQTT v5
 %%--------------------------------------------------------------------

apps/emqx_management/src/emqx_mgmt_cli.erl

@@ -615,13 +615,18 @@ listeners([]) ->
                     {error, _} -> [];
                     MC -> [{max_conns, MC}]
                 end,
+            ShutdownCount =
+                case emqx_listeners:shutdown_count(ID, Bind) of
+                    {error, _} -> [];
+                    SC -> [{shutdown_count, SC}]
+                end,
             Info =
                 [
                     {listen_on, {string, emqx_listeners:format_bind(Bind)}},
                     {acceptors, Acceptors},
                     {proxy_protocol, ProxyProtocol},
                     {running, Running}
-                ] ++ CurrentConns ++ MaxConn,
+                ] ++ CurrentConns ++ MaxConn ++ ShutdownCount,
             emqx_ctl:print("~ts~n", [ID]),
             lists:foreach(fun indent_print/1, Info)
         end,

changes/ce/feat-10568.en.md

@@ -0,0 +1 @@
+Add shutdown counter information to `emqx ctl listeners` command