Merge pull request #8356 from rabbitmq/rabbitmq-server-8323

Fix issues with multiple src and dest queue args

Author: michaelklishin

deps/rabbitmq_shovel/src/rabbit_shovel_util.erl

@@ -224,12 +224,23 @@ validate_params_user(#amqp_params_network{}, _User) ->
     ok.
 
 validate_queue_args(Name, Term0) ->
-    Term = rabbit_data_coercion:to_proplist(Term0),
-    rabbit_parameter_validation:proplist(Name, rabbit_amqqueue:declare_args(), Term).
+    validate_amqp_table_args(Name, Term0, rabbit_amqqueue:declare_args()).
 
 validate_consumer_args(Name, Term0) ->
-    Term = rabbit_data_coercion:to_proplist(Term0),
-    rabbit_parameter_validation:proplist(Name, rabbit_amqqueue:consume_args(), Term).
+    validate_amqp_table_args(Name, Term0, rabbit_amqqueue:consume_args()).
+
+validate_amqp_table_args(Name, Term0, Args) ->
+    Table = rabbit_misc:to_amqp_table(rabbit_data_coercion:to_map(Term0)),
+    [case rabbit_misc:table_lookup(Table, Key) of
+         undefined -> ok;
+         TypeVal ->
+             case Fun(TypeVal, Table) of
+                 ok -> ok;
+                 {error, Error} ->
+                     {error, "invalid arg '~ts' in ~ts: ~255p",
+                      [Key, Name, Error]}
+             end
+     end || {Key, Fun} <- Args].
 
 validate_delete_after(_Name, <<"never">>)          -> ok;
 validate_delete_after(_Name, <<"queue-length">>)   -> ok;

deps/rabbitmq_shovel/test/amqp091_dynamic_SUITE.erl

@@ -48,7 +48,8 @@ groups() ->
     ]},
 
     {stream_queue_tests, [], [
-          stream_queues
+          stream_queues,
+          stream_queues_with_dest_args
     ]}
     ].
 
@@ -109,7 +110,9 @@ quorum_queues(Config) ->
                              {<<"src-queue">>,       <<"src">>},
                              {<<"dest-queue">>,      <<"dest">>},
                              {<<"src-queue-args">>,  #{<<"x-queue-type">> => <<"quorum">>}},
-                             {<<"dest-queue-args">>, #{<<"x-queue-type">> => <<"quorum">>}}
+                             {<<"dest-queue-args">>, #{<<"x-queue-type">> => <<"quorum">>,
+                                                       <<"x-dead-letter-exchange">> => <<"dlx1">>,
+                                                       <<"x-dead-letter-routing-key">> => <<"rk1">>}}
                             ]),
               amqp091_publish_expect(Ch, <<>>, <<"src">>, <<"dest">>, <<"hello">>)
       end).
@@ -128,6 +131,21 @@ stream_queues(Config) ->
               amqp091_publish_expect(Ch, <<>>, <<"src">>, <<"dest">>, <<"hello">>)
       end).
 
+stream_queues_with_dest_args(Config) ->
+    with_amqp091_ch(Config,
+      fun (Ch) ->
+              shovel_test_utils:set_param(
+                Config,
+                <<"test">>, [
+                             {<<"src-queue">>,       <<"src">>},
+                             {<<"dest-queue">>,      <<"dest">>},
+                             {<<"dest-queue-args">>, #{<<"x-queue-type">> => <<"stream">>,
+                                                       <<"x-max-age">>    => <<"12h">>}}
+                            ]),
+              amqp091_publish(Ch, <<>>, <<"src">>, <<"hello">>),
+              shovel_test_utils:amqp091_await_empty(Ch, <<"src">>)
+      end).
+
 set_properties_using_map(Config) ->
     with_amqp091_ch(Config,
       fun (Ch) ->