Get all existing rabbitmq_mqtt tests green

Author: ansd

deps/rabbit/src/rabbit_core_ff.erl

@@ -30,7 +30,8 @@
    {stream_queue,
     #{desc          => "Support queues of type `stream`",
       doc_url       => "https://www.rabbitmq.com/stream.html",
-      stability     => stable,
+      %%TODO remove compatibility code
+      stability     => required,
       depends_on    => [quorum_queue]
      }}).
 

deps/rabbitmq_mqtt/include/rabbit_mqtt.hrl

@@ -11,6 +11,7 @@
 
 %% reader state
 -record(state, {socket,
+                proxy_socket,
                 conn_name,
                 await_recv,
                 deferred_recv,
@@ -42,7 +43,7 @@
          %% same IDs because client and server assign IDs independently of each other.)
          packet_id = 1 :: packet_id(),
          client_id,
-         clean_sess,
+         clean_sess :: boolean(),
          will_msg,
          exchange :: rabbit_exchange:name(),
          ssl_login_name,
@@ -72,37 +73,41 @@
                peer_port,
                proto_human}).
 
-%% does not include vhost: it is used in
-%% the table name
+%% does not include vhost because vhost is used in the (D)ETS table name
 -record(retained_message, {topic,
                            mqtt_msg}).
 
 -define(INFO_ITEMS,
-    [host,
-     port,
-     peer_host,
-     peer_port,
-     protocol,
-     frame_max,
-     client_properties,
-     ssl,
-     ssl_protocol,
-     ssl_key_exchange,
-     ssl_cipher,
-     ssl_hash,
-     conn_name,
-     connection_state,
-     connection,
-     unacked_client_pubs,
-     unacked_server_pubs,
-     packet_id,
-     client_id,
-     clean_sess,
-     will_msg,
-     exchange,
-     ssl_login_name,
-     retainer_pid,
-     user,
-     vhost]).
+        [protocol,
+         host,
+         port,
+         peer_host,
+         peer_port,
+         connection,
+         conn_name,
+         connection_state,
+         ssl,
+         ssl_protocol,
+         ssl_key_exchange,
+         ssl_cipher,
+         ssl_hash,
+         ssl_login_name,
+         client_id,
+         vhost,
+         user,
+         recv_cnt,
+         recv_oct,
+         send_cnt,
+         send_oct,
+         send_pend,
+         clean_sess,
+         will_msg,
+         retainer_pid,
+         exchange,
+         subscriptions,
+         prefetch,
+         messages_unconfirmed,
+         messages_unacknowledged
+        ]).
 
 -define(MQTT_GUIDE_URL, <<"https://rabbitmq.com/mqtt.html">>).

deps/rabbitmq_mqtt/src/rabbit_mqtt_confirms.erl

@@ -37,7 +37,7 @@ insert(PktId, _, State)
 insert(PktId, QNames, State)
   when is_integer(PktId) andalso PktId > 0 ->
     QMap = maps:from_keys(QNames, ok),
-    maps:put(PktId, QMap, State).
+    {ok, maps:put(PktId, QMap, State)}.
 
 -spec confirm([packet_id()], queue_name(), state()) ->
     {[packet_id()], state()}.

deps/rabbitmq_mqtt/src/rabbit_mqtt_connection_info.erl

@@ -489,45 +489,45 @@ hand_off_to_retainer(RetainerPid, Amqp2MqttFun, Topic0, Msg) ->
 
 maybe_send_retained_message(RPid, #mqtt_topic{name = Topic0, qos = SubscribeQos},
                             #proc_state{amqp2mqtt_fun = Amqp2MqttFun,
-                                        packet_id = PacketId} = PState0) ->
+                                        packet_id = PacketId0} = PState0) ->
     Topic1 = Amqp2MqttFun(Topic0),
     case rabbit_mqtt_retainer:fetch(RPid, Topic1) of
-        undefined -> PState0;
-        Msg       ->
-            %% calculate effective QoS as the lower value of SUBSCRIBE frame QoS
-            %% and retained message QoS. The spec isn't super clear on this, we
-            %% do what Mosquitto does, per user feedback.
-            Qos = erlang:min(SubscribeQos, Msg#mqtt_msg.qos),
-            {Id, PState} = case Qos of
-                               ?QOS_0 -> {undefined, PState0};
-                               ?QOS_1 -> {PacketId, PState0#proc_state{packet_id = increment_packet_id(PacketId)}}
-                           end,
+        undefined ->
+            PState0;
+        Msg ->
+            Qos = effective_qos(Msg#mqtt_msg.qos, SubscribeQos),
+            {PacketId, PState} = case Qos of
+                                     ?QOS_0 ->
+                                         {undefined, PState0};
+                                     ?QOS_1 ->
+                                         {PacketId0, PState0#proc_state{packet_id = increment_packet_id(PacketId0)}}
+                                 end,
             serialise_and_send_to_client(
               #mqtt_frame{fixed = #mqtt_frame_fixed{
                                      type = ?PUBLISH,
                                      qos  = Qos,
                                      dup  = false,
                                      retain = Msg#mqtt_msg.retain
                                     }, variable = #mqtt_frame_publish{
-                                                     message_id = Id,
+                                                     message_id = PacketId,
                                                      topic_name = Topic1
                                                     },
                           payload = Msg#mqtt_msg.payload},
               PState),
             PState
     end.
 
-make_will_msg(#mqtt_frame_connect{ will_flag   = false }) ->
+make_will_msg(#mqtt_frame_connect{will_flag   = false}) ->
     undefined;
-make_will_msg(#mqtt_frame_connect{ will_retain = Retain,
-                                   will_qos    = Qos,
-                                   will_topic  = Topic,
-                                   will_msg    = Msg }) ->
-    #mqtt_msg{ retain  = Retain,
-               qos     = Qos,
-               topic   = Topic,
-               dup     = false,
-               payload = Msg }.
+make_will_msg(#mqtt_frame_connect{will_retain = Retain,
+                                  will_qos    = Qos,
+                                  will_topic  = Topic,
+                                  will_msg    = Msg}) ->
+    #mqtt_msg{retain  = Retain,
+              qos     = Qos,
+              topic   = Topic,
+              dup     = false,
+              payload = Msg}.
 
 process_login(_UserBin, _PassBin, _ClientId,
               #proc_state{peer_addr  = Addr,
@@ -999,9 +999,8 @@ binding_action(
                        key = RoutingKey},
     BindingFun(Binding, Username).
 
-send_will(PState = #proc_state{will_msg = undefined}) ->
-    PState;
-
+send_will(#proc_state{will_msg = undefined}) ->
+    ok;
 send_will(PState = #proc_state{will_msg = WillMsg = #mqtt_msg{retain = Retain,
                                                               topic = Topic},
                                retainer_pid = RPid,
@@ -1010,25 +1009,14 @@ send_will(PState = #proc_state{will_msg = WillMsg = #mqtt_msg{retain = Retain,
         ok ->
             publish_to_queues(WillMsg, PState),
             case Retain of
-                false -> ok;
-                true  ->
+                false ->
+                    ok;
+                true ->
                     hand_off_to_retainer(RPid, Amqp2MqttFun, Topic, WillMsg)
             end;
-        Error  ->
-            rabbit_log:warning(
-                "Could not send last will: ~p",
-                [Error])
-    end,
-    %%TODO cancel queue client?
-    % case ChQos1 of
-    %     undefined -> ok;
-    %     _         -> amqp_channel:close(ChQos1)
-    % end,
-    % case ChQos0 of
-    %     undefined -> ok;
-    %     _         -> amqp_channel:close(ChQos0)
-    % end,
-    PState.
+        {error, access_refused = Reason}  ->
+            rabbit_log:error("failed to send will message: ~p", [Reason])
+    end.
 
 publish_to_queues(undefined, PState) ->
     {ok, PState};
@@ -1104,6 +1092,11 @@ deliver_to_queues(Delivery,
 process_routing_confirm(#delivery{confirm = false}, [], PState) ->
     rabbit_global_counters:messages_unroutable_dropped(mqtt, 1),
     PState;
+process_routing_confirm(#delivery{confirm = true,
+                                  msg_seq_no = undefined}, [], PState) ->
+    %% unroutable will message with QoS > 0
+    rabbit_global_counters:messages_unroutable_dropped(mqtt, 1),
+    PState;
 process_routing_confirm(#delivery{confirm = true,
                                   msg_seq_no = MsgId}, [], PState) ->
     rabbit_global_counters:messages_unroutable_returned(mqtt, 1),
@@ -1114,11 +1107,15 @@ process_routing_confirm(#delivery{confirm = true,
     PState;
 process_routing_confirm(#delivery{confirm = false}, _, PState) ->
     PState;
+process_routing_confirm(#delivery{confirm = true,
+                                  msg_seq_no = undefined}, [_|_], PState) ->
+    %% routable will message with QoS > 0
+    PState;
 process_routing_confirm(#delivery{confirm = true,
                                   msg_seq_no = MsgId},
                         Qs, PState = #proc_state{unacked_client_pubs = U0}) ->
     QNames = queue_names(Qs),
-    U = rabbit_mqtt_confirms:insert(MsgId, QNames, U0),
+    {ok, U} = rabbit_mqtt_confirms:insert(MsgId, QNames, U0),
     PState#proc_state{unacked_client_pubs = U}.
 
 send_puback(MsgIds, PState)
@@ -1250,10 +1247,7 @@ deliver_one_to_client(Msg = {QName, QPid, QMsgId, _Redelivered,
                         false ->
                             ?QOS_0
                     end,
-    %% "The QoS of Application Messages sent in response to a Subscription MUST be the minimum
-    %% of the QoS of the originally published message and the Maximum QoS granted by the Server
-    %% [MQTT-3.8.4-8]."
-    QoS = min(PublisherQoS, SubscriberQoS),
+    QoS = effective_qos(PublisherQoS, SubscriberQoS),
     PState1 = maybe_publish_to_client(Msg, QoS, PState0),
     PState = maybe_ack(AckRequired, QoS, QName, QMsgId, PState1),
     %%TODO GC
@@ -1264,6 +1258,13 @@ deliver_one_to_client(Msg = {QName, QPid, QMsgId, _Redelivered,
     ok = maybe_notify_sent(QName, QPid, PState),
     PState.
 
+-spec effective_qos(qos(), qos()) -> qos().
+effective_qos(PublisherQoS, SubscriberQoS) ->
+    %% "The QoS of Application Messages sent in response to a Subscription MUST be the minimum
+    %% of the QoS of the originally published message and the Maximum QoS granted by the Server
+    %% [MQTT-3.8.4-8]."
+    erlang:min(PublisherQoS, SubscriberQoS).
+
 maybe_publish_to_client({_, _, _, _Redelivered = true, _}, ?QOS_0, PState) ->
     %% Do no redeliver to MQTT subscriber who gets message at most once.
     PState;
@@ -1405,28 +1406,27 @@ check_topic_access(TopicName, Access,
             end
     end.
 
-info(unacked_client_pubs, #proc_state{unacked_client_pubs = Val}) -> Val;
+info(protocol, #proc_state{info = #info{proto_human = Val}}) -> Val;
+info(host, #proc_state{info = #info{host = Val}}) -> Val;
+info(port, #proc_state{info = #info{port = Val}}) -> Val;
+info(peer_host, #proc_state{info = #info{peer_host = Val}}) -> Val;
+info(peer_port, #proc_state{info = #info{peer_port = Val}}) -> Val;
+info(ssl_login_name, #proc_state{ssl_login_name = Val}) -> Val;
 info(client_id, #proc_state{client_id = Val}) ->
     rabbit_data_coercion:to_binary(Val);
+info(vhost, #proc_state{auth_state = #auth_state{vhost = Val}}) -> Val;
+info(user, #proc_state{auth_state = #auth_state{username = Val}}) -> Val;
 info(clean_sess, #proc_state{clean_sess = Val}) -> Val;
 info(will_msg, #proc_state{will_msg = Val}) -> Val;
-info(exchange, #proc_state{exchange = #resource{name = Val}}) -> Val;
-info(ssl_login_name, #proc_state{ssl_login_name = Val}) -> Val;
 info(retainer_pid, #proc_state{retainer_pid = Val}) -> Val;
-info(user, #proc_state{auth_state = #auth_state{username = Val}}) -> Val;
-info(vhost, #proc_state{auth_state = #auth_state{vhost = Val}}) -> Val;
-info(host, #proc_state{info = #info{host = Val}}) -> Val;
-info(port, #proc_state{info = #info{port = Val}}) -> Val;
-info(peer_host, #proc_state{info = #info{peer_host = Val}}) -> Val;
-info(peer_port, #proc_state{info = #info{peer_port = Val}}) -> Val;
-info(protocol, #proc_state{info = #info{proto_human = Val}}) -> Val;
-% info(frame_max, PState) -> additional_info(frame_max, PState);
-% info(client_properties, PState) -> additional_info(client_properties, PState);
-% info(ssl, PState) -> additional_info(ssl, PState);
-% info(ssl_protocol, PState) -> additional_info(ssl_protocol, PState);
-% info(ssl_key_exchange, PState) -> additional_info(ssl_key_exchange, PState);
-% info(ssl_cipher, PState) -> additional_info(ssl_cipher, PState);
-% info(ssl_hash, PState) -> additional_info(ssl_hash, PState);
+info(exchange, #proc_state{exchange = #resource{name = Val}}) -> Val;
+info(subscriptions, #proc_state{subscriptions = Val}) ->
+    maps:keys(Val);
+info(prefetch, #proc_state{info = #info{prefetch = Val}}) -> Val;
+info(messages_unconfirmed, #proc_state{unacked_client_pubs = Val}) ->
+    rabbit_mqtt_confirms:size(Val);
+info(messages_unacknowledged, #proc_state{unacked_server_pubs = Val}) ->
+    maps:size(Val);
 info(Other, _) -> throw({bad_argument, Other}).
 
 -spec ssl_login_name(rabbit_net:socket()) ->

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -71,6 +71,7 @@ init(Ref) ->
              rabbit_event:init_stats_timer(
               control_throttle(
                #state{socket                 = RealSocket,
+                      proxy_socket           = rabbit_net:maybe_get_proxy_socket(Sock),
                       conn_name              = ConnStr,
                       await_recv             = false,
                       connection_state       = running,
@@ -409,12 +410,17 @@ parse(Bytes, ParseState) ->
             {error, {cannot_parse, Reason, Stacktrace}}
     end.
 
+%% TODO Send will message in all abnormal shutdowns?
+%% => in terminate/2 depending on Reason
+%% "The Will Message MUST be published when the Network Connection is subsequently
+%% closed unless the Will Message has been deleted by the Server on receipt of a
+%% DISCONNECT Packet [MQTT-3.1.2-8]."
 send_will_and_terminate(PState, State) ->
     send_will_and_terminate(PState, {shutdown, conn_closed}, State).
 
 send_will_and_terminate(PState, Reason, State = #state{conn_name = ConnStr}) ->
-    rabbit_mqtt_processor:send_will(PState),
     rabbit_log_connection:debug("MQTT: about to send will message (if any) on connection ~p", [ConnStr]),
+    rabbit_mqtt_processor:send_will(PState),
     {stop, Reason, State}.
 
 network_error(closed,
@@ -527,5 +533,23 @@ info_internal(connection_state, #state{connection_state = Val}) ->
     Val;
 info_internal(connection, _State) ->
     self();
+info_internal(ssl, #state{socket = Sock, proxy_socket = ProxySock}) ->
+    rabbit_net:proxy_ssl_info(Sock, ProxySock) /= nossl;
+info_internal(ssl_protocol,       S) -> ssl_info(fun ({P,         _}) -> P end, S);
+info_internal(ssl_key_exchange,   S) -> ssl_info(fun ({_, {K, _, _}}) -> K end, S);
+info_internal(ssl_cipher,         S) -> ssl_info(fun ({_, {_, C, _}}) -> C end, S);
+info_internal(ssl_hash,           S) -> ssl_info(fun ({_, {_, _, H}}) -> H end, S);
 info_internal(Key, #state{proc_state = ProcState}) ->
     rabbit_mqtt_processor:info(Key, ProcState).
+
+ssl_info(F, #state{socket = Sock, proxy_socket = ProxySock}) ->
+    case rabbit_net:proxy_ssl_info(Sock, ProxySock) of
+        nossl       -> '';
+        {error, _}  -> '';
+        {ok, Items} ->
+            P = proplists:get_value(protocol, Items),
+            #{cipher := C,
+              key_exchange := K,
+              mac := H} = proplists:get_value(selected_cipher_suite, Items),
+            F({P, {K, C, H}})
+    end.