Handle undefined msg ID from queue

Makes the following tests green:
```
make -C deps/amqp10_client/ ct-system
```

Classic queues send an 'undefined' message ID to the channel when
no ack is required. No ack is required when the send settle mode is
settled. In this case it should be perfectly valid to always send the
same (empty binary) delivery-tag from server to client.

Author: ansd

deps/amqp10_client/test/system_SUITE.erl

@@ -14,21 +14,10 @@
 
 -include("src/amqp10_client.hrl").
 
--compile(export_all).
+-compile([export_all, nowarn_export_all]).
 
--define(UNAUTHORIZED_USER, <<"test_user_no_perm">>).
-
-%% The latch constant defines how many processes are spawned in order
-%% to run certain functionality in parallel. It follows the standard
-%% countdown latch pattern.
--define(LATCH, 100).
-
-%% The wait constant defines how long a consumer waits before it
-%% unsubscribes
--define(WAIT, 200).
-
-%% How to long wait for a process to die after an expected failure
--define(PROCESS_EXIT_TIMEOUT, 5000).
+suite() ->
+    [{timetrap, {seconds, 120}}].
 
 all() ->
     [
@@ -344,7 +333,7 @@ roundtrip(OpenConf, Body) ->
                                                         <<"test1">>,
                                                         settled,
                                                         unsettled_state),
-    {ok, OutMsg} = amqp10_client:get_msg(Receiver, 60000 * 5),
+    {ok, OutMsg} = amqp10_client:get_msg(Receiver, 60_000 * 4),
     ok = amqp10_client:end_session(Session),
     ok = amqp10_client:close_connection(Connection),
     % ct:pal(?LOW_IMPORTANCE, "roundtrip message Out: ~tp~nIn: ~tp~n", [OutMsg, Msg]),
@@ -379,7 +368,7 @@ filtered_roundtrip(OpenConf, Body) ->
                                                         settled,
                                                         unsettled_state),
     ok = amqp10_client:send_msg(Sender, Msg1),
-    {ok, OutMsg1} = amqp10_client:get_msg(DefaultReceiver, 60000 * 5),
+    {ok, OutMsg1} = amqp10_client:get_msg(DefaultReceiver, 60_000 * 4),
     ?assertEqual(<<"msg-1-tag">>, amqp10_msg:delivery_tag(OutMsg1)),
 
     timer:sleep(5 * 1000),
@@ -398,10 +387,10 @@ filtered_roundtrip(OpenConf, Body) ->
                                                         unsettled_state,
                                                         #{<<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > ", Now2Binary/binary>>}),
 
-    {ok, OutMsg2} = amqp10_client:get_msg(DefaultReceiver, 60000 * 5),
+    {ok, OutMsg2} = amqp10_client:get_msg(DefaultReceiver, 60_000 * 4),
     ?assertEqual(<<"msg-2-tag">>, amqp10_msg:delivery_tag(OutMsg2)),
 
-    {ok, OutMsgFiltered} = amqp10_client:get_msg(FilteredReceiver, 60000 * 5),
+    {ok, OutMsgFiltered} = amqp10_client:get_msg(FilteredReceiver, 60_000 * 4),
     ?assertEqual(<<"msg-2-tag">>, amqp10_msg:delivery_tag(OutMsgFiltered)),
 
     ok = amqp10_client:end_session(Session),
@@ -676,11 +665,13 @@ incoming_heartbeat(Config) ->
               idle_time_out => 1000, notify => self()},
     {ok, Connection} = amqp10_client:open_connection(CConf),
     receive
-        {amqp10_event, {connection, Connection,
-         {closed, {resource_limit_exceeded, <<"remote idle-time-out">>}}}} ->
+        {amqp10_event,
+         {connection, Connection0,
+          {closed, {resource_limit_exceeded, <<"remote idle-time-out">>}}}}
+          when Connection0 =:= Connection ->
             ok
     after 5000 ->
-          exit(incoming_heartbeat_assert)
+              exit(incoming_heartbeat_assert)
     end,
     demonitor(MockRef).
 
@@ -704,25 +695,30 @@ publish_messages(Sender, Data, Num) ->
 
 receive_one(Receiver) ->
     receive
-        {amqp10_msg, Receiver, Msg} ->
+        {amqp10_msg, Receiver0, Msg}
+          when Receiver0 =:= Receiver ->
             amqp10_client:accept_msg(Receiver, Msg)
     after 2000 ->
           timeout
     end.
 
 await_disposition(DeliveryTag) ->
     receive
-        {amqp10_disposition, {accepted, DeliveryTag}} -> ok
+        {amqp10_disposition, {accepted, DeliveryTag0}}
+          when DeliveryTag0 =:= DeliveryTag -> ok
     after 3000 ->
               flush(),
               exit(dispostion_timeout)
     end.
 
 await_link(Who, What, Err) ->
     receive
-        {amqp10_event, {link, Who, What}} ->
+        {amqp10_event, {link, Who0, What0}}
+          when Who0 =:= Who andalso
+               What0 =:= What ->
             ok;
-        {amqp10_event, {link, Who, {detached, Why}}} ->
+        {amqp10_event, {link, Who0, {detached, Why}}}
+          when Who0 =:= Who ->
             exit(Why)
     after 5000 ->
               flush(),

deps/rabbit/src/rabbit_amqqueue.erl

@@ -96,7 +96,7 @@
 -type qfun(A) :: fun ((amqqueue:amqqueue()) -> A | no_return()).
 -type qmsg() :: {name(), pid() | {atom(), pid()}, msg_id(),
                  boolean(), mc:state()}.
--type msg_id() :: non_neg_integer().
+-type msg_id() :: rabbit_types:option(non_neg_integer()).
 -type ok_or_errors() ::
         'ok' | {'error', [{'error' | 'exit' | 'throw', any()}]}.
 -type absent_reason() :: 'nodedown' | 'crashed' | stopped | timeout.

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -952,11 +952,22 @@ handle_deliver(ConsumerTag, AckRequired,
             rabbit_log:warning("Delivery to non-existent consumer ~tp", [ConsumerTag]),
             State0;
         #outgoing_link{send_settled = SendSettled} ->
+            %% "The delivery-tag MUST be unique amongst all deliveries that could be
+            %% considered unsettled by either end of the link." [2.6.12]
+            Dtag = if is_integer(MsgId) ->
+                          %% delivery-tag must be unique only per link (not per session)
+                          <<MsgId:64>>;
+                      MsgId =:= undefined ->
+                          %% Both ends of the link will always consider this message settled because
+                          %% "the sender will send all deliveries settled to the receiver" [3.8.2].
+                          %% Hence, the delivery tag does not have to be unique on this link.
+                          %% However, the spec still mandates to send a delivery tag.
+                          <<>>
+                   end,
             Transfer = #'v1_0.transfer'{
                           handle = Handle,
                           delivery_id = {uint, DeliveryId},
-                          %% delivery-tag must be unique only per link (not per session)
-                          delivery_tag = {binary, <<MsgId:64>>},
+                          delivery_tag = {binary, Dtag},
                           %% [3.2.16]
                           message_format = {uint, 0},
                           settled = SendSettled,