Prevent not_found on list web mqtt connections

Fixes rabbitmq#9302

deps/rabbitmq_web_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand.erl

@@ -0,0 +1,86 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+
+-module('Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand').
+
+-include_lib("rabbitmq_mqtt/include/rabbit_mqtt.hrl").
+
+-behaviour('Elixir.RabbitMQ.CLI.CommandBehaviour').
+
+-export([formatter/0,
+         scopes/0,
+         switches/0,
+         aliases/0,
+         usage/0,
+         usage_additional/0,
+         usage_doc_guides/0,
+         banner/2,
+         validate/2,
+         merge_defaults/2,
+         run/2,
+         output/2,
+         description/0,
+         help_section/0]).
+
+formatter() -> 'Elixir.RabbitMQ.CLI.Formatters.Table'.
+scopes() -> [ctl, diagnostics].
+switches() -> [{verbose, boolean}].
+aliases() -> [{'V', verbose}].
+
+description() -> <<"Lists all Web MQTT connections">>.
+
+help_section() ->
+    {plugin, web_mqtt}.
+
+validate(Args, _) ->
+    InfoItems = lists:map(fun atom_to_list/1, ?INFO_ITEMS),
+    case 'Elixir.RabbitMQ.CLI.Ctl.InfoKeys':validate_info_keys(Args,
+                                                               InfoItems) of
+        {ok, _} -> ok;
+        Error   -> Error
+    end.
+
+merge_defaults([], Opts) ->
+    merge_defaults([<<"client_id">>, <<"conn_name">>], Opts);
+merge_defaults(Args, Opts) ->
+    {Args, maps:merge(#{verbose => false}, Opts)}.
+
+usage() ->
+    <<"list_web_mqtt_connections [<column> ...]">>.
+
+usage_additional() ->
+    Prefix = <<" must be one of ">>,
+    InfoItems = 'Elixir.Enum':join(lists:usort(?INFO_ITEMS), <<", ">>),
+    [
+      {<<"<column>">>, <<Prefix/binary, InfoItems/binary>>}
+    ].
+
+usage_doc_guides() ->
+    [?MQTT_GUIDE_URL].
+
+run(Args, #{node := NodeName,
+            timeout := Timeout,
+            verbose := Verbose}) ->
+    InfoKeys = case Verbose of
+        true  -> ?INFO_ITEMS;
+        false -> 'Elixir.RabbitMQ.CLI.Ctl.InfoKeys':prepare_info_keys(Args)
+    end,
+
+    Nodes = 'Elixir.RabbitMQ.CLI.Core.Helpers':nodes_in_cluster(NodeName),
+
+    'Elixir.RabbitMQ.CLI.Ctl.RpcStream':receive_list_items(
+        NodeName,
+        rabbit_web_mqtt_app,
+        emit_connection_info_all,
+        [Nodes, InfoKeys],
+        Timeout,
+        InfoKeys,
+        length(Nodes)).
+
+banner(_, _) -> <<"Listing Web MQTT connections ...">>.
+
+output(Result, _Opts) ->
+    'Elixir.RabbitMQ.CLI.DefaultOutput':output(Result).

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl

@@ -12,7 +12,9 @@
     start/2,
     prep_stop/1,
     stop/1,
-    list_connections/0
+    list_connections/0,
+    emit_connection_info_all/4,
+    emit_connection_info_local/3
 ]).
 
 %% Dummy supervisor - see Ulf Wiger's comment at
@@ -52,23 +54,35 @@ list_connections() ->
     TLSPids   = connection_pids_of_protocol(?TLS_PROTOCOL),
     PlainPids ++ TLSPids.
 
+-spec emit_connection_info_all([node()], rabbit_types:info_keys(), reference(), pid()) -> term().
+emit_connection_info_all(Nodes, Items, Ref, AggregatorPid) ->
+    Pids = [spawn_link(Node, ?MODULE, emit_connection_info_local,
+                       [Items, Ref, AggregatorPid])
+            || Node <- Nodes],
+
+    rabbit_control_misc:await_emitters_termination(Pids).
+
+-spec emit_connection_info_local(rabbit_types:info_keys(), reference(), pid()) -> ok.
+emit_connection_info_local(Items, Ref, AggregatorPid) ->
+    LocalPids = list_connections(),
+    emit_connection_info(Items, Ref, AggregatorPid, LocalPids).
+
+emit_connection_info(Items, Ref, AggregatorPid, Pids) ->
+    rabbit_control_misc:emitting_map_with_exit_handler(
+      AggregatorPid, Ref,
+      fun(Pid) ->
+              rabbit_web_mqtt_handler:info(Pid, Items)
+      end, Pids).
 %%
 %% Implementation
 %%
 
 connection_pids_of_protocol(Protocol) ->
     case rabbit_networking:ranch_ref_of_protocol(Protocol) of
         undefined   -> [];
-        AcceptorRef ->
-            lists:map(fun cowboy_ws_connection_pid/1, ranch:procs(AcceptorRef, connections))
+        AcceptorRef -> ranch:procs(AcceptorRef, connections)
     end.
 
--spec cowboy_ws_connection_pid(pid()) -> pid().
-cowboy_ws_connection_pid(RanchConnPid) ->
-    Children = supervisor:which_children(RanchConnPid),
-    {cowboy_clear, Pid, _, _} = lists:keyfind(cowboy_clear, 1, Children),
-    Pid.
-
 mqtt_init() ->
     CowboyOpts0  = maps:from_list(get_env(cowboy_opts, [])),
     CowboyWsOpts = maps:from_list(get_env(cowboy_ws_opts, [])),

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl

@@ -23,6 +23,7 @@
 ]).
 
 -export([conserve_resources/3]).
+-export([info/2]).
 
 %% cowboy_sub_protocol
 -export([upgrade/4,
@@ -94,6 +95,19 @@ init(Req, Opts) ->
             end
     end.
 
+%% We cannot use a gen_server call, because the handler process is a
+%% special cowboy_websocket process (not a gen_server) which assumes
+%% all gen_server calls are supervisor calls, and does not pass on the
+%% request to this callback module. (see cowboy_websocket:loop/3 and
+%% cowboy_children:handle_supervisor_call/4) However using a generic
+%% gen:call with a special label ?MODULE works fine.
+-spec info(pid(), rabbit_types:info_keys()) ->
+    rabbit_types:infos().
+info(Pid, all) ->
+    info(Pid, ?INFO_ITEMS);
+info(Pid, Items) ->
+    {ok, Res} = gen:call(Pid, ?MODULE, {info, Items}),
+    Res.
 -spec websocket_init(state()) ->
     {cowboy_websocket:commands(), state()} |
     {cowboy_websocket:commands(), state(), hibernate}.
@@ -244,6 +258,10 @@ websocket_info(connection_created, State) ->
     rabbit_core_metrics:connection_created(self(), Infos),
     rabbit_event:notify(connection_created, Infos),
     {[], State, hibernate};
+websocket_info({?MODULE, From, {info, Items}}, State) ->
+    Infos = infos(Items, State),
+    gen:reply(From, Infos),
+    {[], State, hibernate};
 websocket_info(Msg, State) ->
     ?LOG_WARNING("Web MQTT: unexpected message ~tp", [Msg]),
     {[], State, hibernate}.