Avoid crash in stream connection

1.
Prior to this commit, closing a stream connection via:
```
./sbin/rabbitmqctl close_all_user_connections guest enough
```
crashed the stream process as follows:
```
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>   crasher:
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>     initial call: rabbit_stream_reader:init/1
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>     pid: <0.1098.0>
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>     registered_name: []
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>     exception error: no function clause matching
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>                      rabbit_stream_reader:open({call,
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>                                                 {<0.1233.0>,
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>                                                  #Ref<0.519694519.1387790337.15898>}},
2024-08-28 13:00:18.969931+02:00 [error] <0.1098.0>                                                {shutdown,<<"enough">>},
```

This commit fixes this crash.

2.
Both CLI commands and management plugin use the same way
to close MQTT, Web MQTT, and Stream connections: They all send a message
via `Pid ! {shutdown, Reason}` to the connection.

3.
This commit avoids making `rabbit` core app to know about
'Web MQTT'.

4
This commit simplifies rabbit_mqtt_reader by avoiding another
handle_call clause

Author: ansd

deps/rabbit/src/rabbit_connection_tracking.erl

@@ -427,11 +427,7 @@ 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),
-    rpc:call(Node, gen_server, call, [Pid, {shutdown, Message}, infinity]).
+    %% Best effort will work for following plugins:
+    %% rabbitmq_stream, rabbitmq_mqtt, rabbitmq_web_mqtt
+    Pid ! {shutdown, Message}.

deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl

@@ -109,11 +109,6 @@ init(Ref) ->
 handle_call({info, InfoItems}, _From, State) ->
     {reply, infos(InfoItems, State), State, ?HIBERNATE_AFTER};
 
-handle_call({shutdown, Explanation} = Reason, _From, State = #state{conn_name = ConnName}) ->
-    %% rabbit_networking:close_all_user_connections -> rabbit_reader:shutdow
-    ?LOG_INFO("MQTT closing connection ~tp: ~p", [ConnName, Explanation]),
-    {stop, Reason, ok, State};
-
 handle_call(Msg, From, State) ->
     {stop, {mqtt_unexpected_call, Msg, From}, State}.
 
@@ -252,7 +247,7 @@ handle_info({'DOWN', _MRef, process, QPid, _Reason}, State) ->
     {noreply, State, ?HIBERNATE_AFTER};
 
 handle_info({shutdown, Explanation} = Reason, State = #state{conn_name = ConnName}) ->
-    %% rabbitmq_management plugin requests to close connection.
+    %% rabbitmq_management plugin or CLI command requests to close connection.
     ?LOG_INFO("MQTT closing connection ~tp: ~p", [ConnName, Explanation]),
     {stop, Reason, State};