Web MQTT: Send CONNACK error code before closing connection

"If the Client supplies a zero-byte ClientId with CleanSession set to 0,
the Server MUST respond to the CONNECT Packet with a CONNACK return code 0x02
(Identifier rejected) and then close the Network Connection" [MQTT-3.1.3-8].

In Web MQTT, the CONNACK was not sent to the client because the Web MQTT
connection process terminated before being sending the CONNACK to the
client.

deps/rabbitmq_mqtt/src/rabbit_mqtt.erl

@@ -106,5 +106,5 @@ persist_static_configuration() ->
     rabbit_mqtt_util:init_sparkplug(),
 
     {ok, MailboxSoftLimit} = application:get_env(?APP_NAME, mailbox_soft_limit),
-    ?assert(is_number(MailboxSoftLimit)),
+    ?assert(is_integer(MailboxSoftLimit)),
     ok = persistent_term:put(?PERSISTENT_TERM_MAILBOX_SOFT_LIMIT, MailboxSoftLimit).

deps/rabbitmq_mqtt/test/reader_SUITE.erl

@@ -37,7 +37,6 @@ groups() ->
        quorum_clean_session_true,
        classic_clean_session_true,
        classic_clean_session_false,
-       non_clean_sess_empty_client_id,
        event_authentication_failure,
        rabbit_mqtt_qos0_queue_overflow
       ]}
@@ -218,22 +217,6 @@ classic_clean_session_true(Config) ->
 classic_clean_session_false(Config) ->
     validate_durable_queue_type(Config, <<"classicCleanSessionFalse">>, false, rabbit_classic_queue).
 
-%% "If the Client supplies a zero-byte ClientId with CleanSession set to 0,
-%% the Server MUST respond to the CONNECT Packet with a CONNACK return code 0x02
-%% (Identifier rejected) and then close the Network Connection" [MQTT-3.1.3-8].
-non_clean_sess_empty_client_id(Config) ->
-    {ok, C} = emqtt:start_link(
-                [{clientid, <<>>},
-                 {clean_start, false},
-                 {proto_ver, v4},
-                 {host, "localhost"},
-                 {port, rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt)}
-                ]),
-    process_flag(trap_exit, true),
-    ?assertMatch({error, {client_identifier_not_valid, _}},
-                 emqtt:connect(C)),
-    ok = await_exit(C).
-
 event_authentication_failure(Config) ->
     {ok, C} = emqtt:start_link(
                 [{username, <<"Trudy">>},

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -79,6 +79,7 @@ subgroups() ->
          ,non_clean_sess_reconnect_qos1
          ,non_clean_sess_reconnect_qos0
          ,non_clean_sess_reconnect_qos0_and_qos1
+         ,non_clean_sess_empty_client_id
          ,subscribe_same_topic_same_qos
          ,subscribe_same_topic_different_qos
          ,subscribe_multiple
@@ -852,6 +853,16 @@ non_clean_sess_reconnect_qos0_and_qos1(Config) ->
     C3 = connect(ClientId, Config, [{clean_start, true}]),
     ok = emqtt:disconnect(C3).
 
+%% "If the Client supplies a zero-byte ClientId with CleanSession set to 0,
+%% the Server MUST respond to the CONNECT Packet with a CONNACK return code 0x02
+%% (Identifier rejected) and then close the Network Connection" [MQTT-3.1.3-8].
+non_clean_sess_empty_client_id(Config) ->
+    {C, Connect} = util:start_client(<<>>, Config, 0, [{clean_start, false}]),
+    process_flag(trap_exit, true),
+    ?assertMatch({error, {client_identifier_not_valid, _}},
+                 Connect(C)),
+    ok = await_exit(C).
+
 subscribe_same_topic_same_qos(Config) ->
     C = connect(?FUNCTION_NAME, Config),
     Topic = <<"a/b">>,

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl

@@ -139,6 +139,8 @@ websocket_info({bump_credit, Msg}, State) ->
     handle_credits(State);
 websocket_info({reply, Data}, State) ->
     {[{binary, Data}], State, hibernate};
+websocket_info({stop, CloseCode, Error}, State) ->
+    stop(State, CloseCode, Error);
 websocket_info({'EXIT', _, _}, State) ->
     stop(State);
 websocket_info({'$gen_cast', QueueEvent = {queue_event, _, _}},
@@ -271,9 +273,8 @@ handle_data1(Data, State = #state{socket = Socket,
                                                 proc_state = ProcState1});
                         {error, Reason} ->
                             ?LOG_ERROR("Rejected Web MQTT connection ~ts: ~p", [ConnName, Reason]),
-                            stop_mqtt_protocol_error({_SendWill = false, State},
-                                                     connect_packet_rejected,
-                                                     ConnName)
+                            self() ! {stop, ?CLOSE_PROTOCOL_ERROR, connect_packet_rejected},
+                            {[], {_SendWill = false, State}}
                     end;
                 _ ->
                     case rabbit_mqtt_processor:process_packet(Packet, ProcState) of