Support dynamic creation of queues

 ## What?
Support the `dynamic` field of sources and targets.

 ## Why?
1. This allows AMQP clients to dynamically create exclusive queues, which
   can be useful for RPC workloads.
2. Support creation of JMS temporary queues over AMQP using the Qpid JMS
   client. Exclusive queues map very nicely to JMS temporary queues
   because:

> Although sessions are used to create temporary destinations, this is only
for convenience. Their scope is actually the entire connection. Their
lifetime is that of their connection and any of the connection’s sessions
are allowed to create a consumer for them.

https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#creating-temporary-destinations

 ## How?
If the terminus contains the capability `temporary-queue` as defined in
[amqp-bindmap-jms-v1.0-wd10](https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=67638)
[5.2] and as sent by Qpid JMS client,
RabbitMQ will create an exclusive queue.
(This allows a future commit to take other actions if capability
`temporary-topic` will be used, such as the additional creation of bindings.)

No matter what the desired node properties are, RabbitMQ will set the
lifetime policy delete-on-close deleting the exclusive queue when the
link which caused its creation ceases to exist. This means the exclusive
queue will be deleted if either:
* the link gets detached, or
* the session ends, or
* the connection closes

Although the AMQP JMS Mapping and Qpid JMS create only a **sending** link
with `dynamic=true`, this commit also supports **receiving** links with
`dynamic=true` for non-JMS AMQP clients.

RabbitMQ is free to choose the generated queue name. As suggested by the
AMQP spec, the generated queue name will contain the container-id and link
name unless they are very long.

Co-authored-by: Arnaud Cogoluègnes <[email protected]>

Author: ansd

deps/amqp10_client/src/amqp10_client_session.erl

@@ -698,23 +698,39 @@ build_frames(Channel, Trf, Payload, MaxPayloadSize, Acc) ->
 
 make_source(#{role := {sender, _}}) ->
     #'v1_0.source'{};
-make_source(#{role := {receiver, #{address := Address} = Source, _Pid}, filter := Filter}) ->
+make_source(#{role := {receiver, Source, _Pid},
+              filter := Filter}) ->
     Durable = translate_terminus_durability(maps:get(durable, Source, none)),
+    Dynamic = maps:get(dynamic, Source, false),
     TranslatedFilter = translate_filters(Filter),
-    #'v1_0.source'{address = {utf8, Address},
+    #'v1_0.source'{address = make_address(Source),
                    durable = {uint, Durable},
-                   filter = TranslatedFilter}.
+                   dynamic = Dynamic,
+                   filter = TranslatedFilter,
+                   capabilities = make_capabilities(Source)}.
 
 make_target(#{role := {receiver, _Source, _Pid}}) ->
     #'v1_0.target'{};
-make_target(#{role := {sender, #{address := Address} = Target}}) ->
+make_target(#{role := {sender, Target}}) ->
     Durable = translate_terminus_durability(maps:get(durable, Target, none)),
-    TargetAddr = case is_binary(Address) of
-                     true -> {utf8, Address};
-                     false -> Address
-                 end,
-    #'v1_0.target'{address = TargetAddr,
-                   durable = {uint, Durable}}.
+    Dynamic = maps:get(dynamic, Target, false),
+    #'v1_0.target'{address = make_address(Target),
+                   durable = {uint, Durable},
+                   dynamic = Dynamic,
+                   capabilities = make_capabilities(Target)}.
+
+make_address(#{address := Addr}) ->
+    if is_binary(Addr) ->
+           {utf8, Addr};
+       is_atom(Addr) ->
+           Addr
+    end.
+
+make_capabilities(#{capabilities := Caps0}) ->
+    Caps = [{symbol, C} || C <- Caps0],
+    {array, symbol, Caps};
+make_capabilities(_) ->
+    undefined.
 
 max_message_size(#{max_message_size := Size})
   when is_integer(Size) andalso

deps/rabbit/src/rabbit_amqp_reader.erl

@@ -260,8 +260,8 @@ wait_for_shutdown_sessions(TimerRef, #v1{tracked_channels = Channels} = State0)
             State = untrack_channel(ChannelNum, SessionPid, State0),
             wait_for_shutdown_sessions(TimerRef, State);
         shutdown_sessions_timeout ->
-            ?LOG_INFO("sessions not shut down after ~b ms: ~p",
-                      [?SHUTDOWN_SESSIONS_TIMEOUT, Channels]),
+            ?LOG_INFO("sessions running ~b ms after requested to be shut down: ~p",
+                      [?SHUTDOWN_SESSIONS_TIMEOUT, maps:values(Channels)]),
             State0
     end.
 
@@ -792,6 +792,7 @@ send_to_new_session(
       connection = #v1_connection{outgoing_max_frame_size = MaxFrame,
                                   vhost = Vhost,
                                   user = User,
+                                  container_id = ContainerId,
                                   name = ConnName},
       writer = WriterPid} = State) ->
     %% Subtract fixed frame header size.
@@ -804,6 +805,7 @@ send_to_new_session(
                  OutgoingMaxFrameSize,
                  User,
                  Vhost,
+                 ContainerId,
                  ConnName,
                  BeginFrame],
     case rabbit_amqp_session_sup:start_session(SessionSup, ChildArgs) of

deps/rabbit/src/rabbit_amqp_session.erl

@@ -55,9 +55,12 @@ groups() ->
       [
        %% authz
        attach_source_queue,
+       attach_source_queue_dynamic,
        attach_target_exchange,
        attach_target_topic_exchange,
        attach_target_queue,
+       attach_target_queue_dynamic_exchange_write,
+       attach_target_queue_dynamic_queue_configure,
        target_per_message_exchange,
        target_per_message_internal_exchange,
        target_per_message_topic,
@@ -437,6 +440,39 @@ attach_source_queue(Config) ->
     end,
     ok = close_connection_sync(Conn).
 
+attach_source_queue_dynamic(Config) ->
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+
+    %% missing configure permission to queue
+    ok = set_permissions(Config, <<>>, <<".*">>, <<".*">>),
+
+    Source = #{address => undefined,
+               dynamic => true,
+               capabilities => [<<"temporary-queue">>],
+               durable => none},
+    AttachArgs = #{name => <<"my link">>,
+                   role => {receiver, Source, self()},
+                   snd_settle_mode => unsettled,
+                   rcv_settle_mode => first,
+                   filter => #{}},
+    {ok, _Recv} = amqp10_client:attach_link(Session, AttachArgs),
+    receive {amqp10_event,
+             {session, Session,
+              {ended, Error}}} ->
+                #'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_UNAUTHORIZED_ACCESS,
+                              description = {utf8, Description}} = Error,
+                ?assertEqual(
+                   match,
+                   re:run(Description,
+                          <<"^configure access to queue 'amq\.dyn-.*' in vhost "
+                            "'test vhost' refused for user 'test user'$">>,
+                          [{capture, none}]))
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
+    end,
+    ok = close_connection_sync(Connection).
+
 attach_target_exchange(Config) ->
     XName = <<"amq.fanout">>,
     Address1 = rabbitmq_amqp_address:exchange(XName),
@@ -485,6 +521,61 @@ attach_target_queue(Config) ->
     end,
     ok = amqp10_client:close_connection(Conn).
 
+attach_target_queue_dynamic_exchange_write(Config) ->
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+
+    %% missing write permission to default exchange
+    ok = set_permissions(Config, <<".*">>, <<>>, <<".*">>),
+
+    Target = #{address => undefined,
+               dynamic => true,
+               capabilities => [<<"temporary-queue">>]},
+    AttachArgs = #{name => <<"my link">>,
+                   role => {sender, Target},
+                   snd_settle_mode => mixed,
+                   rcv_settle_mode => first},
+    {ok, _Recv} = amqp10_client:attach_link(Session, AttachArgs),
+    ExpectedErr = error_unauthorized(
+                    <<"write access to exchange 'amq.default' ",
+                      "in vhost 'test vhost' refused for user 'test user'">>),
+    receive {amqp10_event, {session, Session, {ended, ExpectedErr}}} -> ok
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
+    end,
+    ok = close_connection_sync(Connection).
+
+attach_target_queue_dynamic_queue_configure(Config) ->
+    OpnConf = connection_config(Config),
+    {ok, Connection} = amqp10_client:open_connection(OpnConf),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+
+    %% missing configure permission to queue
+    ok = set_permissions(Config, <<>>, <<".*">>, <<".*">>),
+
+    Target = #{address => undefined,
+               dynamic => true,
+               capabilities => [<<"temporary-queue">>]},
+    AttachArgs = #{name => <<"my link">>,
+                   role => {sender, Target},
+                   snd_settle_mode => mixed,
+                   rcv_settle_mode => first},
+    {ok, _Recv} = amqp10_client:attach_link(Session, AttachArgs),
+    receive {amqp10_event,
+             {session, Session,
+              {ended, Error}}} ->
+                #'v1_0.error'{condition = ?V_1_0_AMQP_ERROR_UNAUTHORIZED_ACCESS,
+                              description = {utf8, Description}} = Error,
+                ?assertEqual(
+                   match,
+                   re:run(Description,
+                          <<"^configure access to queue 'amq\.dyn-.*' in vhost "
+                            "'test vhost' refused for user 'test user'$">>,
+                          [{capture, none}]))
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
+    end,
+    ok = close_connection_sync(Connection).
+
 target_per_message_exchange(Config) ->
     TargetAddress = null,
     To1 = rabbitmq_amqp_address:exchange(<<"amq.fanout">>),