Support predeclared feature in static shovels

Author: MarcialRosales

deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

@@ -41,7 +41,7 @@
 %% from and can break with the next upgrade. It should not be used by
 %% another one that the one who created it or survive a node restart.
 %% Thus, function references have been replace by the following MFA.
--export([decl_fun/3, publish_fun/4, props_fun_timestamp_header/4,
+-export([decl_fun/3, check_fun/3, publish_fun/4, props_fun_timestamp_header/4,
          props_fun_forward_header/5]).
 
 -define(MAX_CONNECTION_CLOSE_TIMEOUT, 10000).
@@ -56,7 +56,7 @@ parse(_Name, {source, Source}) ->
     CArgs = proplists:get_value(consumer_args, Source, []),
     #{module => ?MODULE,
       uris => proplists:get_value(uris, Source),
-      resource_decl => decl_fun(Source),
+      resource_decl => decl_fun({source, Source}),
       queue => Queue,
       delete_after => proplists:get_value(delete_after, Source, never),
       prefetch_count => Prefetch,
@@ -72,7 +72,7 @@ parse(Name, {destination, Dest}) ->
     PropsFun2 = add_timestamp_header_fun(ATH, PropsFun1),
     #{module => ?MODULE,
       uris => proplists:get_value(uris, Dest),
-      resource_decl  => decl_fun(Dest),
+      resource_decl  => decl_fun({destination, Dest}),
       props_fun => PropsFun2,
       fields_fun => PubFieldsFun,
       add_forward_headers => AFH,
@@ -606,16 +606,35 @@ parse_declaration({[{Method, Props} | _Rest], _Acc}) ->
 parse_declaration({[Method | Rest], Acc}) ->
     parse_declaration({[{Method, []} | Rest], Acc}).
 
-decl_fun(Endpoint) ->
-    Decl = parse_declaration({proplists:get_value(declarations, Endpoint, []),
-                              []}),
+decl_fun({source, Endpoint}) ->
+    case parse_declaration({proplists:get_value(declarations, Endpoint, []), []}) of 
+        [] -> 
+            case proplists:get_value(predeclared, application:get_env(?APP, topology, []), false) of
+                true -> case proplists:get_value(queue, Endpoint) of 
+                            <<>> -> fail({invalid_parameter_value, declarations, {require_non_empty}});
+                            Queue -> {?MODULE, check_fun, [Queue]}
+                        end;
+                false -> {?MODULE, decl_fun, []}
+            end;
+        Decl -> {?MODULE, decl_fun, [Decl]}
+    end;
+decl_fun({destination, Endpoint}) ->
+    Decl = parse_declaration({proplists:get_value(declarations, Endpoint, []), []}),
     {?MODULE, decl_fun, [Decl]}.
-
+    
 decl_fun(Decl, _Conn, Ch) ->
     [begin
          amqp_channel:call(Ch, M)
      end || M <- lists:reverse(Decl)].
 
+check_fun(Queue, _Conn, Ch) ->
+    try
+        amqp_channel:call(Ch, #'queue.declare'{queue   = Queue,
+                                               passive = true})
+    after
+        catch amqp_channel:close(Ch)
+    end.
+
 parse_parameter(Param, Fun, Value) ->
     try
         Fun(Value)

deps/rabbitmq_shovel/src/rabbit_shovel_config.erl

@@ -66,11 +66,13 @@ convert_from_legacy(Config) ->
      {reconnect_delay, RD}].
 
 parse(ShovelName, Config0) ->
+    rabbit_log:debug("rabbit_shovel_config:parse ~p ~p", [ShovelName, Config0]),
     try
         validate(Config0),
         case is_legacy(Config0) of
-            true ->
+            true ->            
                 Config = convert_from_legacy(Config0),
+                rabbit_log:debug("rabbit_shovel_config:parse is_legacy -> ~p", [Config]),
                 parse_current(ShovelName, Config);
             false ->
                 parse_current(ShovelName, Config0)
@@ -124,8 +126,9 @@ validate_uris0([Uri | Uris]) ->
 validate_uris0([]) -> ok.
 
 parse_current(ShovelName, Config) ->
+    rabbit_log:debug("rabbit_shovel_config:parse_current ~p", [ShovelName]),
     {source, Source} = proplists:lookup(source, Config),
-    validate(Source),
+    validate(Source),    
     SrcMod = resolve_module(proplists:get_value(protocol, Source, amqp091)),
     {destination, Destination} = proplists:lookup(destination, Config),
     validate(Destination),

deps/rabbitmq_shovel/src/rabbit_shovel_sup.erl

@@ -70,6 +70,7 @@ parse_configuration(_Defaults, [], Acc) ->
 parse_configuration(Defaults, [{ShovelName, ShovelConfig} | Env], Acc) when
     is_atom(ShovelName) andalso is_list(ShovelConfig)
 ->
+    rabbit_log:debug("rabbit_shovel:parse_configuration ~p ~p", [ShovelName, ShovelConfig]),
     case dict:is_key(ShovelName, Acc) of
         true ->
             {error, {duplicate_shovel_definition, ShovelName}};

deps/rabbitmq_shovel/test/configuration_SUITE.erl

@@ -12,6 +12,7 @@
 
 -compile(export_all).
 
+-define(QUEUE,    <<"test_queue">>).
 -define(EXCHANGE,    <<"test_exchange">>).
 -define(TO_SHOVEL,   <<"to_the_shovel">>).
 -define(FROM_SHOVEL, <<"from_the_shovel">>).
@@ -31,7 +32,10 @@ groups() ->
           invalid_legacy_configuration,
           valid_legacy_configuration,
           valid_configuration
-        ]}
+        ]},
+      {with_predefined_topology, [], [
+          valid_configuration_with_predefined_resources
+      ]}
     ].
 
 %% -------------------------------------------------------------------
@@ -53,9 +57,19 @@ end_per_suite(Config) ->
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_group(with_predefined_topology, Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, application, set_env,
+        [rabbitmq_shovel, topology, [{predeclared, true}]]),
+    Config;
+
 init_per_group(_, Config) ->
     Config.
 
+end_per_group(with_predefined_topology, Config) ->
+     ok = rabbit_ct_broker_helpers:rpc(Config, 0, application, unset_env,
+        [rabbitmq_shovel, topology]),
+    Config;
+
 end_per_group(_, Config) ->
     Config.
 
@@ -209,6 +223,11 @@ valid_configuration(Config) ->
     ok = setup_shovels(Config),
     run_valid_test(Config).
 
+valid_configuration_with_predefined_resources(Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, setup_shovels2, [Config]),
+    run_valid_test2(Config),
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE,  await_running_shovel, [test_shovel]).
+
 run_valid_test(Config) ->
     Chan = rabbit_ct_client_helpers:open_channel(Config, 0),
 
@@ -271,14 +290,20 @@ run_valid_test(Config) ->
 
     rabbit_ct_client_helpers:close_channel(Chan).
 
+run_valid_test2(Config) ->
+    Chan = rabbit_ct_client_helpers:open_channel(Config, 0),
+    amqp_channel:call(Chan, #'queue.declare'{queue   = ?QUEUE,
+                                             durable = true}),             
+    rabbit_ct_client_helpers:close_channel(Chan).
+
 setup_legacy_shovels(Config) ->
     ok = rabbit_ct_broker_helpers:rpc(Config, 0,
       ?MODULE, setup_legacy_shovels1, [Config]).
 
 setup_shovels(Config) ->
     ok = rabbit_ct_broker_helpers:rpc(Config, 0,
       ?MODULE, setup_shovels1, [Config]).
-
+    
 setup_legacy_shovels1(Config) ->
     _ = application:stop(rabbitmq_shovel),
     Hostname = ?config(rmq_hostname, Config),
@@ -349,6 +374,34 @@ setup_shovels1(Config) ->
     ok = application:start(rabbitmq_shovel),
     await_running_shovel(test_shovel).
 
+setup_shovels2(Config) ->
+    _ = application:stop(rabbitmq_shovel),
+    Hostname = ?config(rmq_hostname, Config),
+    TcpPort = rabbit_ct_broker_helpers:get_node_config(Config, 0,
+      tcp_port_amqp),
+    %% a working config
+    application:set_env(
+      rabbitmq_shovel,
+      shovels,
+      [{test_shovel,
+        [{source,
+          [{uris, [rabbit_misc:format("amqp://~ts:~b/%2f?heartbeat=5",
+                                      [Hostname, TcpPort])]},
+           {queue, ?QUEUE}]},
+         {destination,
+          [{uris, [rabbit_misc:format("amqp://~ts:~b/%2f",
+                                      [Hostname, TcpPort])]},
+           {publish_fields, [{exchange, ?EXCHANGE}, {routing_key, ?FROM_SHOVEL}]},
+           {publish_properties, [{delivery_mode, 2},
+                                 {cluster_id,    <<"my-cluster">>},
+                                 {content_type,  ?SHOVELLED}]},
+           {add_forward_headers, true},
+           {add_timestamp_header, true}]},
+         {ack_mode, on_confirm}]}],
+      infinity),
+
+    ok = application:start(rabbitmq_shovel).    
+
 await_running_shovel(Name) ->
     case [N || {N, _, {running, _}, _}
                       <- rabbit_shovel_status:status(),