Web mqtt: close connection when receive invalid data

- from https://www.rfc-editor.org/rfc/rfc6455#section-7.4.1,
code `1003` is for "an endpoint is terminating the connection
because it has received a type of data it cannot accept"

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl

@@ -39,6 +39,7 @@
 %% Close frame status codes as defined in https://www.rfc-editor.org/rfc/rfc6455#section-7.4.1
 -define(CLOSE_NORMAL, 1000).
 -define(CLOSE_PROTOCOL_ERROR, 1002).
+-define(CLOSE_INVALID_DATA, 1003).
 -define(CLOSE_INCONSISTENT_MSG_TYPE, 1007).
 
 %% cowboy_sub_protcol
@@ -135,14 +136,11 @@ websocket_handle({Ping, _}, State)
 websocket_handle(Ping, State)
   when Ping =:= ping orelse Ping =:= pong ->
     {[], State, hibernate};
-%% Log any other unexpected frames.
+%% Log and close connection when receiving any other unexpected frames.
 websocket_handle(Frame, State) ->
     rabbit_log_connection:info("Web MQTT: unexpected WebSocket frame ~tp",
                     [Frame]),
-    %%TODO close connection instead?
+    stop(State, ?CLOSE_INVALID_DATA, "unexpected WebSocket frame").
-    %%"MQTT Control Packets MUST be sent in WebSocket binary data frames.
-    %% If any other type of data frame is received the recipient MUST close the Network Connection"
-    {[], State, hibernate}.
 
 -spec websocket_info(any(), State) ->
     {cowboy_websocket:commands(), State} |

deps/rabbitmq_web_mqtt/test/system_SUITE.erl

@@ -32,6 +32,7 @@ groups() ->
         , maintenance
         , client_no_supported_protocol
         , client_not_support_mqtt
+        , invalid_data
         ]}
     ].
 
@@ -193,7 +194,14 @@ client_protocol_test(Config, Protocol) ->
     {_, [{http_response, Res}]} = rfc6455_client:open(WS),
     {'HTTP/1.1', 400, <<"Bad Request">>, _} = cow_http:parse_status_line(rabbit_data_coercion:to_binary(Res)),
     rfc6455_client:send_binary(WS, rabbit_ws_test_util:mqtt_3_1_1_connect_packet()),
-    {close, _P} = rfc6455_client:recv(WS).
+    {close, _} = rfc6455_client:recv(WS, timer:seconds(1)).
+
+invalid_data(Config) ->
+    PortStr = rabbit_ws_test_util:get_web_mqtt_port_str(Config),
+    WS = rfc6455_client:new("ws://localhost:" ++ PortStr ++ "/ws", self(), undefined, ["mqtt"]),
+    {ok, _} = rfc6455_client:open(WS),
+    rfc6455_client:send(WS, "not-binary-data"),
+    {close, {1003, _}} = rfc6455_client:recv(WS, timer:seconds(1)).
 
 %% Web mqtt connections are tracked together with mqtt connections
 num_mqtt_connections(Config, Node) ->