Merge pull request #9271 from rabbitmq/mqtt-send-failed

Close MQTT connection if sending fails

Author: michaelklishin

deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl

@@ -336,13 +336,14 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
                 connect_packet_unprocessed ->
                     Send = fun(Data) ->
                                    try rabbit_net:port_command(Socket, Data)
-                                   catch error:Error ->
+                                   catch error:Reason ->
                                              ?LOG_ERROR("writing to MQTT socket ~p failed: ~p",
-                                                        [Socket, Error])
+                                                        [Socket, Reason]),
+                                             exit({send_failed, Reason})
                                    end,
                                    ok
                            end,
-                    case rabbit_mqtt_processor:init(Packet, Socket, ConnName, Send) of
+                    try rabbit_mqtt_processor:init(Packet, Socket, ConnName, Send) of
                         {ok, ProcState1} ->
                             ?LOG_INFO("Accepted MQTT connection ~ts for client ID ~ts",
                                       [ConnName, rabbit_mqtt_processor:info(client_id, ProcState1)]),
@@ -358,9 +359,11 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
                             ?LOG_ERROR("Rejected MQTT connection ~ts with Connect Reason Code ~p",
                                        [ConnName, ConnectReasonCode]),
                             {stop, shutdown, {_SendWill = false, State}}
+                    catch exit:{send_failed, Reason} ->
+                              network_error(Reason, State)
                     end;
                 _ ->
-                    case rabbit_mqtt_processor:process_packet(Packet, ProcState) of
+                    try rabbit_mqtt_processor:process_packet(Packet, ProcState) of
                         {ok, ProcState1} ->
                             process_received_bytes(
                               Rest,
@@ -377,6 +380,8 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
                             {stop, {shutdown, Reason}, pstate(State, ProcState1)};
                         {stop, {disconnect, {client_initiated, SendWill}}, ProcState1} ->
                             {stop, normal, {SendWill, pstate(State, ProcState1)}}
+                    catch exit:{send_failed, Reason} ->
+                              network_error(Reason, State)
                     end
             end;
         {error, {disconnect_reason_code, ReasonCode} = Reason} ->

deps/rabbitmq_mqtt/test/java_SUITE_data/src/test/java/com/rabbitmq/mqtt/test/MqttTest.java

@@ -520,6 +520,7 @@ public void sessionRedelivery(TestInfo info) throws MqttException, InterruptedEx
     // Message has been delivered but connection has failed.
     waitAtMost(() -> receivedMessagesSize() == 1);
     assertThat(firstMessage().getPayload()).isEqualTo(payload);
+    Thread.sleep(100);
     assertThat(client.isConnected()).isFalse();
 
     receivedMessages.clear();