Make Erlang AMQP library safe by default

By default we want AMQP libraries to send messages as durable. The app
can always choose to opt in to send the message as non-durable.

Author: ansd

deps/amqp10_client/src/amqp10_msg.erl

@@ -266,24 +266,25 @@ body_bin(#amqp10_msg{body = #'v1_0.amqp_value'{} = Body}) ->
 %% following stucture:
 %% {amqp10_disposition, {accepted | rejected, DeliveryTag}}
 -spec new(delivery_tag(), amqp10_body() | binary(), boolean()) -> amqp10_msg().
-new(DeliveryTag, Body, Settled) when is_binary(Body) ->
-    #amqp10_msg{transfer = #'v1_0.transfer'{delivery_tag = {binary, DeliveryTag},
-                                            settled = Settled,
-                                            message_format = {uint, ?MESSAGE_FORMAT}},
-                body = [#'v1_0.data'{content = Body}]};
+new(DeliveryTag, Bin, Settled) when is_binary(Bin) ->
+    Body = [#'v1_0.data'{content = Bin}],
+    new(DeliveryTag, Body, Settled);
 new(DeliveryTag, Body, Settled) -> % TODO: constrain to amqp types
-    #amqp10_msg{transfer = #'v1_0.transfer'{delivery_tag = {binary, DeliveryTag},
-                                            settled = Settled,
-                                            message_format = {uint, ?MESSAGE_FORMAT}},
-                body = Body}.
+    #amqp10_msg{
+       transfer = #'v1_0.transfer'{
+                     delivery_tag = {binary, DeliveryTag},
+                     settled = Settled,
+                     message_format = {uint, ?MESSAGE_FORMAT}},
+       %% This lib is safe by default.
+       header = #'v1_0.header'{durable = true},
+       body = Body}.
 
 %% @doc Create a new settled amqp10 message using the specified delivery tag
 %% and body.
 -spec new(delivery_tag(), amqp10_body() | binary()) -> amqp10_msg().
 new(DeliveryTag, Body) ->
     new(DeliveryTag, Body, false).
 
-
 % First 3 octets are the format
 % the last 1 octet is the version
 % See 2.8.11 in the spec

deps/rabbit/test/amqp_client_SUITE.erl

@@ -58,6 +58,9 @@ groups() ->
        sender_settle_mode_unsettled,
        sender_settle_mode_unsettled_fanout,
        sender_settle_mode_mixed,
+       durable_field_classic_queue,
+       durable_field_quorum_queue,
+       durable_field_stream,
        invalid_transfer_settled_flag,
        quorum_queue_rejects,
        receiver_settle_mode_first,
@@ -916,6 +919,71 @@ sender_settle_mode_mixed(Config) ->
                  rabbitmq_amqp_client:delete_queue(LinkPair, QName)),
     ok = close(Init).
 
+durable_field_classic_queue(Config) ->
+    QName  = atom_to_binary(?FUNCTION_NAME),
+    durable_field(Config, <<"classic">>, QName).
+
+durable_field_quorum_queue(Config) ->
+    QName  = atom_to_binary(?FUNCTION_NAME),
+    durable_field(Config, <<"quorum">>, QName).
+
+durable_field_stream(Config) ->
+    QName  = atom_to_binary(?FUNCTION_NAME),
+    durable_field(Config, <<"stream">>, QName).
+
+durable_field(Config, QueueType, QName)
+  when is_binary(QueueType) ->
+    Address = rabbitmq_amqp_address:queue(QName),
+    {_Connection, Session, LinkPair} = Init = init(Config),
+    QProps = #{arguments => #{<<"x-queue-type">> => {utf8, QueueType}}},
+    {ok, #{type := QueueType}} = rabbitmq_amqp_client:declare_queue(LinkPair, QName, QProps),
+    {ok, Sender} = amqp10_client:attach_sender_link(
+                     Session, <<"test-sender">>, Address, unsettled),
+    wait_for_credit(Sender),
+
+    ok = amqp10_client:send_msg(
+           Sender,
+           amqp10_msg:set_headers(
+             #{durable => true},
+             amqp10_msg:new(<<"t1">>, <<"durable">>))),
+    ok = amqp10_client:send_msg(
+           Sender,
+           amqp10_msg:set_headers(
+             #{durable => false},
+             amqp10_msg:new(<<"t2">>, <<"non-durable">>))),
+    %% Even though the AMQP spec defines durable=false as default
+    %% (i.e. durable is false if the field is omitted on the wire),
+    %% we expect our AMQP Erlang library to be safe by default,
+    %% and therefore send the message as durable=true on behalf of us.
+    ok = amqp10_client:send_msg(
+           Sender, amqp10_msg:new(<<"t3">>, <<"lib sends as durable by default">>)),
+
+    ok = wait_for_accepts(3),
+    ok = detach_link_sync(Sender),
+    flush(sent),
+
+    Filter = consume_from_first(QueueType),
+    {ok, Receiver} = amqp10_client:attach_receiver_link(
+                       Session, <<"test-receiver">>, Address, unsettled,
+                       none, Filter),
+
+    ok = amqp10_client:flow_link_credit(Receiver, 3, never),
+    [M1, M2, M3] = receive_messages(Receiver, 3),
+    ?assertEqual(<<"durable">>, amqp10_msg:body_bin(M1)),
+    ?assertMatch(#{durable := true}, amqp10_msg:headers(M1)),
+    ?assertEqual(<<"non-durable">>, amqp10_msg:body_bin(M2)),
+    ?assertMatch(#{durable := false}, amqp10_msg:headers(M2)),
+    ?assertEqual(<<"lib sends as durable by default">>, amqp10_msg:body_bin(M3)),
+    ?assertMatch(#{durable := true}, amqp10_msg:headers(M3)),
+
+    ok = amqp10_client:accept_msg(Receiver, M1),
+    ok = amqp10_client:accept_msg(Receiver, M2),
+    ok = amqp10_client:accept_msg(Receiver, M3),
+
+    ok = detach_link_sync(Receiver),
+    {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, QName),
+    close(Init).
+
 invalid_transfer_settled_flag(Config) ->
     OpnConf = connection_config(Config),
     {ok, Connection} = amqp10_client:open_connection(OpnConf),
@@ -3278,7 +3346,7 @@ max_message_size_client_to_server(Config) ->
     {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"sender">>, Address, mixed),
     ok = wait_for_credit(Sender),
 
-    PayloadSmallEnough = binary:copy(<<0>>, MaxMessageSize - 10),
+    PayloadSmallEnough = binary:copy(<<0>>, MaxMessageSize - 20),
     ?assertEqual(ok,
                  amqp10_client:send_msg(Sender, amqp10_msg:new(<<"t1">>, PayloadSmallEnough, false))),
     ok = wait_for_accepted(<<"t1">>),