Increase MQTT test coverage and fix edge cases

Author: ansd

deps/rabbitmq_ct_helpers/src/rabbit_ct_broker_helpers.erl

@@ -1461,11 +1461,10 @@ set_vhost_limit(Config, Node, VHost, Limit0, Value) ->
       max_queues      -> <<"max-queues">>;
       Other -> rabbit_data_coercion:to_binary(Other)
     end,
-    Definition = rabbit_json:encode(#{Limit => Value}),
+    Limits = [{Limit, Value}],
     rpc(Config, Node,
-        rabbit_vhost_limit,
-        set,
-        [VHost, Definition, <<"ct-tests">>]).
+        rabbit_vhost_limit, set,
+        [VHost, Limits, <<"ct-tests">>]).
 
 set_user_limits(Config, Username, Limits) ->
     set_user_limits(Config, 0, Username, Limits).

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -142,7 +142,6 @@ process_request(?PUBACK,
         {QMsgId, U} ->
             case rabbit_queue_type:settle(QName, complete, ?CONSUMER_TAG, [QMsgId], QStates0) of
                 {ok, QStates, Actions} ->
-                    %%TODO rabbit_channel:incr_queue_stats/3
                     message_acknowledged(QName, State),
                     {ok, handle_queue_actions(Actions, State#state{unacked_server_pubs = U,
                                                                    queue_states = QStates})};
@@ -349,14 +348,14 @@ process_connect(#mqtt_packet{
                               PacketConnect, State0) of
         {ok, SessionPresent0, State1} ->
             {?CONNACK_ACCEPT, SessionPresent0, State1};
-        {error, ReturnCode0, State1} ->
-            {ReturnCode0, false, State1}
+        {error, ConnectionRefusedReturnCode, State1} ->
+            {ConnectionRefusedReturnCode, false, State1}
     end,
-    ResponsePacket = #mqtt_packet{fixed    = #mqtt_packet_fixed{type = ?CONNACK},
-                                  variable = #mqtt_packet_connack{
-                                                session_present = SessionPresent,
-                                                return_code = ReturnCode}},
-    SendFun(ResponsePacket, State),
+    Response = #mqtt_packet{fixed = #mqtt_packet_fixed{type = ?CONNACK},
+                            variable = #mqtt_packet_connack{
+                                          session_present = SessionPresent,
+                                          return_code = ReturnCode}},
+    SendFun(Response, State),
     return_connack(ReturnCode, State).
 
 check_protocol_version(#mqtt_packet_connect{proto_ver = ProtoVersion}) ->
@@ -368,7 +367,7 @@ check_protocol_version(#mqtt_packet_connect{proto_ver = ProtoVersion}) ->
     end.
 
 check_client_id(#mqtt_packet_connect{clean_sess = false,
-                                     client_id = []}) ->
+                                     client_id = <<>>}) ->
     {error, ?CONNACK_ID_REJECTED};
 check_client_id(_) ->
     ok.
@@ -401,12 +400,10 @@ login({UserBin, PassBin,
       State0) ->
     ClientId = ensure_client_id(ClientId0),
     case process_login(UserBin, PassBin, ClientId, State0) of
-        already_connected ->
-            {ok, already_connected};
         {ok, State} ->
             {ok, Packet, State#state{clean_sess = CleanSess,
                                      client_id = ClientId}};
-        {error, _Reason, _State} = Err ->
+        {error, _ConnectionRefusedReturnCode, _State} = Err ->
             Err
     end.
 
@@ -419,8 +416,6 @@ ensure_client_id(ClientId)
   when is_binary(ClientId) ->
     ClientId.
 
-register_client(already_connected, _State) ->
-    ok;
 register_client(Packet = #mqtt_packet_connect{proto_ver = ProtoVersion},
                 State = #state{client_id = ClientId,
                                socket = Socket,
@@ -462,8 +457,6 @@ register_client(Packet = #mqtt_packet_connect{proto_ver = ProtoVersion},
             {ok, NewProcState(undefined)}
     end.
 
-notify_connection_created(already_connected) ->
-    ok;
 notify_connection_created(#mqtt_packet_connect{}) ->
     rabbit_networking:register_non_amqp_connection(self()),
     self() ! connection_created,
@@ -616,7 +609,7 @@ maybe_unbind(TopicName, TopicNames, QName, State0) ->
                 {ok, _Output, State} ->
                     {ok, State};
                 {error, Reason, _State} = Err ->
-                    rabbit_log:error("Failed to unbind ~s with topic ~s: ~p",
+                    rabbit_log:error("Failed to unbind ~s with topic '~s': ~p",
                                      [rabbit_misc:rs(QName), TopicName, Reason]),
                     Err
             end
@@ -674,17 +667,18 @@ make_will_msg(#mqtt_packet_connect{will_retain = Retain,
               dup     = false,
               payload = Msg}.
 
-process_login(_UserBin, _PassBin, _ClientId,
+process_login(_UserBin, _PassBin, ClientId,
               #state{peer_addr  = Addr,
                      auth_state = #auth_state{username = Username,
                                               user = User,
                                               vhost = VHost
-                                             }})
+                                             }} = State)
   when Username =/= undefined, User =/= undefined, VHost =/= underfined ->
     rabbit_core_metrics:auth_attempt_failed(list_to_binary(inet:ntoa(Addr)), Username, mqtt),
-    rabbit_log_connection:warning("MQTT detected duplicate connect/login attempt for user ~ts, vhost ~ts",
-                                  [Username, VHost]),
-    already_connected;
+    rabbit_log_connection:error(
+      "MQTT detected duplicate connect attempt for client ID '~ts', user '~ts', vhost '~ts'",
+      [ClientId, Username, VHost]),
+    {error, ?CONNACK_ID_REJECTED, State};
 process_login(UserBin, PassBin, ClientId,
               #state{socket = Sock,
                      ssl_login_name = SslLoginName,
@@ -713,7 +707,7 @@ process_login(UserBin, PassBin, ClientId,
         {ok, _Output, State} ->
             rabbit_core_metrics:auth_attempt_succeeded(RemoteIpAddressBin, UsernameBin, mqtt),
             {ok, State};
-        {error, _Reason, _State} = Err ->
+        {error, _ConnectionRefusedReturnCode, _State} = Err ->
             rabbit_core_metrics:auth_attempt_failed(RemoteIpAddressBin, UsernameBin, mqtt),
             Err
     end.
@@ -730,30 +724,30 @@ check_vhost_exists(#{vhost := VHost,
     end.
 
 check_vhost_connection_limit(#{vhost := VHost,
-                               username_bin := UsernameBin}) ->
+                               client_id := ClientId,
+                               username_bin := Username}) ->
     case rabbit_vhost_limit:is_over_connection_limit(VHost) of
         false ->
             ok;
         {true, Limit} ->
             rabbit_log_connection:error(
-              "Error on MQTT connection ~p~n"
-              "access to vhost '~s' refused for user '~s': "
-              "vhost connection limit (~p) is reached",
-              [self(), VHost, UsernameBin, Limit]),
+              "Failed to create MQTT connection because vhost connection limit is reached; "
+              "vhost: '~s'; connection limit: ~p; user: '~s'; client ID '~s'",
+              [VHost, Limit, Username, ClientId]),
             {error, ?CONNACK_NOT_AUTHORIZED}
     end.
 
 check_vhost_alive(#{vhost := VHost,
+                    client_id := ClientId,
                     username_bin := UsernameBin}) ->
     case rabbit_vhost_sup_sup:is_vhost_alive(VHost) of
         true  ->
             ok;
         false ->
             rabbit_log_connection:error(
-              "Error on MQTT connection ~p~n"
-              "access refused for user '~s': "
-              "vhost is down",
-              [self(), UsernameBin, VHost]),
+              "Failed to create MQTT connection because vhost is down; "
+              "vhost: ~s; user: ~s; client ID: ~s",
+              [VHost, UsernameBin, ClientId]),
             {error, ?CONNACK_NOT_AUTHORIZED}
     end.
 
@@ -798,16 +792,16 @@ notify_auth_result(AuthResult, Username, #state{conn_name = ConnName}) ->
        {connection_name, ConnName},
        {connection_type, network}]).
 
-check_user_connection_limit(#{user := #user{username = Username}}) ->
+check_user_connection_limit(#{user := #user{username = Username},
+                              client_id := ClientId}) ->
     case rabbit_auth_backend_internal:is_over_connection_limit(Username) of
         false ->
             ok;
         {true, Limit} ->
             rabbit_log_connection:error(
-              "Error on MQTT connection ~p~n"
-              "access refused for user '~s': "
-              "user connection limit (~p) is reached",
-              [self(), Username, Limit]),
+              "Failed to create MQTT connection because user connection limit is reached; "
+              "user: '~s'; connection limit: ~p; client ID '~s'",
+              [Username, Limit, ClientId]),
             {error, ?CONNACK_NOT_AUTHORIZED}
     end.
 
@@ -1230,8 +1224,6 @@ deliver_to_queues(Delivery,
                   RoutedToQNames,
                   State0 = #state{queue_states = QStates0,
                                   proto_ver = ProtoVer}) ->
-    %% TODO only lookup fields that are needed using ets:select / match?
-    %% TODO Use ETS continuations to be more space efficient
     Qs0 = rabbit_amqqueue:lookup(RoutedToQNames),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
     case rabbit_queue_type:deliver(Qs, Delivery, QStates0) of
@@ -1588,7 +1580,7 @@ maybe_publish_to_client(
       routing_keys = [RoutingKey | _CcRoutes],
       content = #content{payload_fragments_rev = FragmentsRev}}},
   QoS, State0 = #state{send_fun = SendFun}) ->
-    {PacketId, State} = queue_packet_id_to_packet_id(QMsgId, QoS, State0),
+    {PacketId, State} = msg_id_to_packet_id(QMsgId, QoS, State0),
     Packet =
     #mqtt_packet{
        fixed = #mqtt_packet_fixed{
@@ -1609,11 +1601,11 @@ maybe_publish_to_client(
     message_delivered(QNameOrType, Redelivered, QoS, State),
     State.
 
-queue_packet_id_to_packet_id(_, ?QOS_0, State) ->
+msg_id_to_packet_id(_, ?QOS_0, State) ->
     %% "A PUBLISH packet MUST NOT contain a Packet Identifier if its QoS value is set to 0 [MQTT-2.2.1-2]."
     {undefined, State};
-queue_packet_id_to_packet_id(QMsgId, ?QOS_1, #state{packet_id = PktId,
-                                                    unacked_server_pubs = U} = State) ->
+msg_id_to_packet_id(QMsgId, ?QOS_1, #state{packet_id = PktId,
+                                           unacked_server_pubs = U} = State) ->
     {PktId, State#state{packet_id = increment_packet_id(PktId),
                         unacked_server_pubs = maps:put(PktId, QMsgId, U)}}.
 
@@ -1626,14 +1618,8 @@ increment_packet_id(Id) ->
 
 maybe_auto_ack(_AckRequired = true, ?QOS_0, QName, QMsgId,
                State = #state{queue_states = QStates0}) ->
-    case rabbit_queue_type:settle(QName, complete, ?CONSUMER_TAG, [QMsgId], QStates0) of
-        {ok, QStates, Actions} ->
-            %%TODO rabbit_channel:incr_queue_stats/3
-            handle_queue_actions(Actions, State#state{queue_states = QStates});
-        {protocol_error, _ErrorType, _Reason, _ReasonArgs} = Err ->
-            %%TODO handle error
-            throw(Err)
-    end;
+    {ok, QStates, Actions} = rabbit_queue_type:settle(QName, complete, ?CONSUMER_TAG, [QMsgId], QStates0),
+    handle_queue_actions(Actions, State#state{queue_states = QStates});
 maybe_auto_ack(_, _, _, _, State) ->
     State.
 

deps/rabbitmq_mqtt/src/rabbit_mqtt_util.erl

@@ -37,8 +37,6 @@ queue_name_bin(ClientId, QoS) ->
     queue_name0(Prefix, QoS).
 
 queue_name0(Prefix, ?QOS_0) ->
-    %%TODO consider shortening the QoS0 queue name to save memory
-    %%(can't change QoS1 name to not break rolling updates)
     <<Prefix/binary, "0">>;
 queue_name0(Prefix, ?QOS_1) ->
     <<Prefix/binary, "1">>.