Delete AMQP 0.9.1 header x-mqtt-dup

AMQP 0.9.1 header x-mqtt-dup was determined by the incoming MQTT PUBLISH
packet's DUP flag. Its only use was to determine the outgoing MQTT
PUBLISH packet's DUP flag. However, that's wrong behaviour because
the MQTT 3.1.1 protocol spec mandates:
"The value of the DUP flag from an incoming PUBLISH packet is not
propagated when the PUBLISH Packet is sent to subscribers by the Server.
The DUP flag in the outgoing PUBLISH packet is set independently to the
incoming PUBLISH packet, its value MUST be determined solely by whether
the outgoing PUBLISH packet is a retransmission."
[MQTT-3.3.1-3]

Native MQTT fixes this wrong behaviour. Therefore, we can delete this
AMQP 0.9.1 header.

Author: ansd

deps/rabbitmq_mqtt/include/rabbit_mqtt_packet.hrl

@@ -59,10 +59,10 @@
 %% Packet identifier is a non zero two byte integer.
 -type packet_id() :: 1..16#ffff.
 
--record(mqtt_packet_fixed,    {type   = 0,
-                               dup    = 0,
-                               qos    = 0,
-                               retain = 0
+-record(mqtt_packet_fixed,    {type = 0 :: packet_type(),
+                               dup = false :: boolean(),
+                               qos = 0 :: qos(),
+                               retain = false :: boolean()
                               }).
 
 -record(mqtt_packet,          {fixed :: #mqtt_packet_fixed{},
@@ -74,7 +74,7 @@
 
 -record(mqtt_packet_connect,  {proto_ver :: 3 | 4,
                                will_retain :: boolean(),
-                               will_qos :: 0..2,
+                               will_qos :: qos(),
                                will_flag :: boolean(),
                                clean_sess :: boolean(),
                                keep_alive :: non_neg_integer(),
@@ -103,7 +103,7 @@
 -record(mqtt_msg,             {retain :: boolean(),
                                qos :: qos(),
                                topic :: binary(),
-                               dup :: option(boolean()),
+                               dup :: boolean(),
                                packet_id :: option(packet_id()),
                                payload :: binary()}).
 

deps/rabbitmq_mqtt/src/rabbit_mqtt_packet.erl

@@ -37,9 +37,9 @@ parse(<<>>, authenticated) ->
     {more, fun(Bin) -> parse(Bin, authenticated) end};
 parse(<<MessageType:4, Dup:1, QoS:2, Retain:1, Rest/binary>>, authenticated) ->
     parse_remaining_len(Rest, #mqtt_packet_fixed{ type   = MessageType,
-                                                  dup    = bool(Dup),
+                                                  dup    = int_to_bool(Dup),
                                                   qos    = QoS,
-                                                  retain = bool(Retain) });
+                                                  retain = int_to_bool(Retain) });
 parse(<<?CONNECT:4, 0:4, Rest/binary>>, unauthenticated) ->
     parse_remaining_len(Rest, #mqtt_packet_fixed{type = ?CONNECT});
 parse(Bin, Cont)
@@ -131,10 +131,10 @@ parse_connect(Bin, Fixed, Length) ->
                             wrap(Fixed,
                                  #mqtt_packet_connect{
                                     proto_ver   = ProtoVersion,
-                                    will_retain = bool(WillRetain),
+                                    will_retain = int_to_bool(WillRetain),
                                     will_qos    = WillQos,
-                                    will_flag   = bool(WillFlag),
-                                    clean_sess  = bool(CleanSession),
+                                    will_flag   = int_to_bool(WillFlag),
+                                    clean_sess  = int_to_bool(CleanSession),
                                     keep_alive  = KeepAlive,
                                     client_id   = ClientId,
                                     will_topic  = WillTopic,
@@ -184,9 +184,6 @@ parse_msg(Bin, 0) ->
 parse_msg(<<Len:16/big, Msg:Len/binary, Rest/binary>>, _) ->
     {Msg, Rest}.
 
-bool(0) -> false;
-bool(1) -> true.
-
 %% serialisation
 
 -spec serialise(#mqtt_packet{}, ?MQTT_PROTO_V3 | ?MQTT_PROTO_V4) ->
@@ -206,7 +203,7 @@ serialise_variable(#mqtt_packet_fixed   { type        = ?CONNACK } = Fixed,
                    #mqtt_packet_connack { session_present = SessionPresent,
                                           return_code = ReturnCode },
                    <<>> = PayloadBin, _Vsn) ->
-    VariableBin = <<?RESERVED:7, (opt(SessionPresent)):1, ReturnCode:8>>,
+    VariableBin = <<?RESERVED:7, (bool_to_int(SessionPresent)):1, ReturnCode:8>>,
     serialise_fixed(Fixed, VariableBin, PayloadBin);
 
 serialise_variable(#mqtt_packet_fixed  { type       = SubAck } = Fixed,
@@ -247,15 +244,16 @@ serialise_variable(#mqtt_packet_fixed {} = Fixed,
                    <<>> = _PayloadBin, _Vsn) ->
     serialise_fixed(Fixed, <<>>, <<>>).
 
-serialise_fixed(#mqtt_packet_fixed{ type   = Type,
-                                    dup    = Dup,
-                                    qos    = Qos,
-                                    retain = Retain }, VariableBin, Payload)
-  when is_integer(Type) andalso ?CONNECT =< Type andalso Type =< ?DISCONNECT ->
+serialise_fixed(#mqtt_packet_fixed{type   = Type,
+                                   dup    = Dup,
+                                   qos    = Qos,
+                                   retain = Retain}, VariableBin, Payload)
+  when is_integer(Type) andalso ?CONNECT =< Type andalso Type =< ?DISCONNECT andalso
+       is_integer(Qos) andalso 0 =< Qos andalso Qos =< 2 ->
     Len = size(VariableBin) + iolist_size(Payload),
     true = (Len =< ?MAX_LEN),
     LenBin = serialise_len(Len),
-    [<<Type:4, (opt(Dup)):1, (opt(Qos)):2, (opt(Retain)):1,
+    [<<Type:4, (bool_to_int(Dup)):1, Qos:2, (bool_to_int(Retain)):1,
        LenBin/binary, VariableBin/binary>>, Payload].
 
 serialise_utf(String) ->
@@ -269,10 +267,13 @@ serialise_len(N) when N =< ?LOWBITS ->
 serialise_len(N) ->
     <<1:1, (N rem ?HIGHBIT):7, (serialise_len(N div ?HIGHBIT))/binary>>.
 
-opt(undefined)            -> ?RESERVED;
-opt(false)                -> 0;
-opt(true)                 -> 1;
-opt(X) when is_integer(X) -> X.
-
 protocol_name_approved(Ver, Name) ->
     lists:member({Ver, Name}, ?PROTOCOL_NAMES).
+
+-spec int_to_bool(0 | 1) -> boolean().
+int_to_bool(0) -> false;
+int_to_bool(1) -> true.
+
+-spec bool_to_int(boolean()) -> 0 | 1.
+bool_to_int(false) -> 0;
+bool_to_int(true) -> 1.

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -1084,7 +1084,6 @@ binding_action(ExchangeName, TopicName, QName, BindingFun, #auth_state{user = #u
 publish_to_queues(
   #mqtt_msg{qos        = Qos,
             topic      = Topic,
-            dup        = Dup,
             packet_id  = PacketId,
             payload    = Payload},
   #state{cfg = #cfg{exchange = ExchangeName,
@@ -1095,10 +1094,8 @@ publish_to_queues(
         } = State) ->
     RoutingKey = mqtt_to_amqp(Topic),
     Confirm = Qos > ?QOS_0,
-    Headers = [{<<"x-mqtt-publish-qos">>, byte, Qos},
-               {<<"x-mqtt-dup">>, bool, Dup}],
     Props = #'P_basic'{
-               headers = Headers,
+               headers = [{<<"x-mqtt-publish-qos">>, byte, Qos}],
                delivery_mode = delivery_mode(Qos)},
     {ClassId, _MethodId} = rabbit_framing_amqp_0_9_1:method_id('basic.publish'),
     Content = #content{
@@ -1491,12 +1488,6 @@ maybe_publish_to_client(
        fixed = #mqtt_packet_fixed{
                   type = ?PUBLISH,
                   qos = QoS,
-                  %% "The value of the DUP flag from an incoming PUBLISH packet is not
-                  %% propagated when the PUBLISH Packet is sent to subscribers by the Server.
-                  %% The DUP flag in the outgoing PUBLISH packet is set independently to the
-                  %% incoming PUBLISH packet, its value MUST be determined solely by whether
-                  %% the outgoing PUBLISH packet is a retransmission [MQTT-3.3.1-3]."
-                  %% Therefore, we do not consider header value <<"x-mqtt-dup">> here.
                   dup = Redelivered},
        variable = #mqtt_packet_publish{
                      packet_id = PacketId,

deps/rabbitmq_mqtt/test/shared_SUITE.erl

@@ -1307,8 +1307,7 @@ trace(Config) ->
                    <<"channel">> := 0,
                    <<"user">> := <<"guest">>,
                    <<"properties">> := #{<<"delivery_mode">> := 2,
-                                         <<"headers">> := #{<<"x-mqtt-publish-qos">> := 1,
-                                                            <<"x-mqtt-dup">> := false}},
+                                         <<"headers">> := #{<<"x-mqtt-publish-qos">> := 1}},
                    <<"routed_queues">> := [<<"mqtt-subscription-trace_subscriberqos0">>]},
                  rabbit_misc:amqp_table(PublishHeaders)),
 
@@ -1324,8 +1323,7 @@ trace(Config) ->
                    <<"channel">> := 0,
                    <<"user">> := <<"guest">>,
                    <<"properties">> := #{<<"delivery_mode">> := 2,
-                                         <<"headers">> := #{<<"x-mqtt-publish-qos">> := 1,
-                                                            <<"x-mqtt-dup">> := false}},
+                                         <<"headers">> := #{<<"x-mqtt-publish-qos">> := 1}},
                    <<"redelivered">> := 0},
                  rabbit_misc:amqp_table(DeliverHeaders)),