Mqtt: test close connection

Author: dcorbacho

deps/rabbit/src/rabbit_connection_tracking.erl

@@ -427,6 +427,10 @@ close_connection(#tracked_connection{pid = Pid, type = direct}, Message) ->
     %% Do an RPC call to the node running the direct client.
     Node = node(Pid),
     rpc:call(Node, amqp_direct_connection, server_close, [Pid, 320, Message]);
+close_connection(#tracked_connection{pid = Pid,
+                                     protocol = {'Web MQTT', _}}, Message) ->
+    % this will work for connections to web mqtt plugin
+    Pid ! {shutdown, Message};
 close_connection(#tracked_connection{pid = Pid}, Message) ->
     % best effort, this will work for connections to the stream plugin
     Node = node(Pid),

deps/rabbit/src/rabbit_networking.erl

@@ -531,9 +531,8 @@ close_connections(Pids, Explanation) ->
 
 -spec close_all_user_connections(rabbit_types:username(), string()) -> 'ok'.
 close_all_user_connections(Username, Explanation) ->
-    Pids = [Pid || #tracked_connection{pid = Pid} <- rabbit_connection_tracking:list_of_user(Username)],
-    [close_connection(Pid, Explanation) || Pid <- Pids],
-    ok.
+    Tracked = rabbit_connection_tracking:list_of_user(Username),
+    rabbit_connection_tracking:close_connections(Tracked, Explanation, 0).
 
 %% Meant to be used by tests only
 -spec close_all_connections(string()) -> 'ok'.

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -129,6 +129,8 @@ cluster_size_1_tests() ->
      ,retained_message_conversion
      ,bind_exchange_to_exchange
      ,bind_exchange_to_exchange_single_message
+     ,cli_close_all_connections
+     ,cli_close_all_user_connections
     ].
 
 cluster_size_3_tests() ->
@@ -141,6 +143,8 @@ cluster_size_3_tests() ->
      rabbit_mqtt_qos0_queue,
      rabbit_mqtt_qos0_queue_kill_node,
      cli_list_queues,
+     cli_close_all_connections,
+     cli_close_all_user_connections,
      delete_create_queue,
      session_reconnect,
      session_takeover,
@@ -207,7 +211,9 @@ end_per_group(_, Config) ->
 
 init_per_testcase(T, Config)
   when T =:= management_plugin_connection;
-       T =:= management_plugin_enable ->
+       T =:= management_plugin_enable;
+       T =:= cli_close_all_user_connections;
+       T =:= cli_close_all_connections ->
     inets:start(),
     init_per_testcase0(T, Config);
 init_per_testcase(Testcase, Config) ->
@@ -220,7 +226,9 @@ init_per_testcase0(Testcase, Config) ->
 
 end_per_testcase(T, Config)
   when T =:= management_plugin_connection;
-       T =:= management_plugin_enable ->
+       T =:= management_plugin_enable;
+       T =:= cli_close_all_user_connections;
+       T =:= cli_close_all_connections ->
     ok = inets:stop(),
     end_per_testcase0(T, Config);
 end_per_testcase(Testcase, Config) ->
@@ -1208,6 +1216,36 @@ management_plugin_enable(Config) ->
 
     ok = emqtt:disconnect(C).
 
+cli_close_all_connections(Config) ->
+    KeepaliveSecs = 99,
+    ClientId = atom_to_binary(?FUNCTION_NAME),
+
+    _ = connect(ClientId, Config, [{keepalive, KeepaliveSecs}]),
+    eventually(?_assertEqual(1, length(http_get(Config, "/connections"))), 1000, 10),
+
+    process_flag(trap_exit, true),
+    {ok, String} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["close_all_connections", "bye"]),
+    ?assertEqual(match, re:run(String, "Closing .* reason: bye", [{capture, none}])),
+
+    process_flag(trap_exit, false),
+    eventually(?_assertEqual([], http_get(Config, "/connections")),
+               1000, 10).
+
+cli_close_all_user_connections(Config) ->
+    KeepaliveSecs = 99,
+    ClientId = atom_to_binary(?FUNCTION_NAME),
+
+    _ = connect(ClientId, Config, [{keepalive, KeepaliveSecs}]),
+    eventually(?_assertEqual(1, length(http_get(Config, "/connections"))), 1000, 10),
+
+    process_flag(trap_exit, true),
+    {ok, String} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["close_all_user_connections","guest", "bye"]),
+    ?assertEqual(match, re:run(String, "Closing .* reason: bye", [{capture, none}])),
+
+    process_flag(trap_exit, false),
+    eventually(?_assertEqual([], http_get(Config, "/connections")),
+               1000, 10).
+
 %% Test that queues of type rabbit_mqtt_qos0_queue can be listed via rabbitmqctl.
 cli_list_queues(Config) ->
     C = connect(?FUNCTION_NAME, Config),