AMQP 1.0: Support dynamic creation of queues (backport #13231)

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/include/rabbit_amqp_reader.hrl

@@ -3,6 +3,8 @@
 -define(CLOSING_TIMEOUT, 30_000).
 -define(SILENT_CLOSE_DELAY, 3_000).
 
+-define(SHUTDOWN_SESSIONS_TIMEOUT, 10_000).
+
 %% Allow for potentially large sets of tokens during the SASL exchange.
 %% https://docs.oasis-open.org/amqp/amqp-cbs/v1.0/csd01/amqp-cbs-v1.0-csd01.html#_Toc67999915
 -define(INITIAL_MAX_FRAME_SIZE, 8192).

deps/rabbit/src/rabbit_amqp_reader.erl

@@ -220,10 +220,17 @@ terminate(_, _) ->
 %%--------------------------------------------------------------------------
 %% error handling / termination
 
-close(Error, State = #v1{connection = #v1_connection{timeout = Timeout}}) ->
+close(Error, State0 = #v1{connection = #v1_connection{timeout = Timeout}}) ->
     %% Client properties will be emitted in the connection_closed event by rabbit_reader.
-    ClientProperties = i(client_properties, State),
+    ClientProperties = i(client_properties, State0),
     put(client_properties, ClientProperties),
+
+    %% "It is illegal to send any more frames (or bytes of any other kind)
+    %% after sending a close frame." [2.7.9]
+    %% Sessions might send frames via the writer proc.
+    %% Therefore, let's first try to orderly shutdown our sessions.
+    State = shutdown_sessions(State0),
+
     Time = case Timeout > 0 andalso
                 Timeout < ?CLOSING_TIMEOUT of
                true -> Timeout;
@@ -233,6 +240,31 @@ close(Error, State = #v1{connection = #v1_connection{timeout = Timeout}}) ->
     ok = send_on_channel0(State, #'v1_0.close'{error = Error}, amqp10_framing),
     State#v1{connection_state = closed}.
 
+shutdown_sessions(#v1{tracked_channels = Channels} = State) ->
+    maps:foreach(fun(_ChannelNum, Pid) ->
+                         gen_server:cast(Pid, shutdown)
+                 end, Channels),
+    TimerRef = erlang:send_after(?SHUTDOWN_SESSIONS_TIMEOUT,
+                                 self(),
+                                 shutdown_sessions_timeout),
+    wait_for_shutdown_sessions(TimerRef, State).
+
+wait_for_shutdown_sessions(TimerRef, #v1{tracked_channels = Channels} = State)
+  when map_size(Channels) =:= 0 ->
+    ok = erlang:cancel_timer(TimerRef, [{async, false},
+                                        {info, false}]),
+    State;
+wait_for_shutdown_sessions(TimerRef, #v1{tracked_channels = Channels} = State0) ->
+    receive
+        {{'DOWN', ChannelNum}, _MRef, process, SessionPid, _Reason} ->
+            State = untrack_channel(ChannelNum, SessionPid, State0),
+            wait_for_shutdown_sessions(TimerRef, State);
+        shutdown_sessions_timeout ->
+            ?LOG_INFO("sessions running ~b ms after requested to be shut down: ~p",
+                      [?SHUTDOWN_SESSIONS_TIMEOUT, maps:values(Channels)]),
+            State0
+    end.
+
 handle_session_exit(ChannelNum, SessionPid, Reason, State0) ->
     State = untrack_channel(ChannelNum, SessionPid, State0),
     S = case terminated_normally(Reason) of
@@ -760,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.
@@ -772,6 +805,7 @@ send_to_new_session(
                  OutgoingMaxFrameSize,
                  User,
                  Vhost,
+                 ContainerId,
                  ConnName,
                  BeginFrame],
     case rabbit_amqp_session_sup:start_session(SessionSup, ChildArgs) of