Retry if we could not start links because the plugin wasn't yet running

Per discussion with @dcorbacho.

3 files changed, 20 insertions, 12 deletions

diff --git a/deps/rabbitmq_federation/src/rabbit_federation_app.erl b/deps/rabbitmq_federation/src/rabbit_federation_app.erl
index 15d54eb5bd..b205ee4185 100644
--- a/deps/rabbitmq_federation/src/rabbit_federation_app.erl
+++ b/deps/rabbitmq_federation/src/rabbit_federation_app.erl
@@ -10,7 +10,7 @@
 -include("rabbit_federation.hrl").
 
 -behaviour(application).
--export([start/2, stop/1]).
+-export([start/2, stop/1, start_federation_links/0]).
 
 %% Dummy supervisor - see Ulf Wiger's comment at
 %% http://erlang.2086793.n4.nabble.com/initializing-library-applications-without-processes-td2094473.html
@@ -29,13 +29,17 @@
 -export([init/1]).
 
 start(_Type, _StartArgs) ->
-    rabbit_federation_exchange_link:go(),
-    rabbit_federation_queue_link:go(),
+    _ = timer:apply_after(500, ?MODULE, start_federation_links, []),
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
 stop(_State) ->
     rabbit_federation_pg:stop_scope(),
     ok.
+
+start_federation_links() ->
+    rabbit_federation_exchange_link:go(),
+    rabbit_federation_queue_link:go().
+
 %%----------------------------------------------------------------------------
 
 init([]) ->
diff --git a/deps/rabbitmq_federation/src/rabbit_federation_exchange_link.erl b/deps/rabbitmq_federation/src/rabbit_federation_exchange_link.erl
index 2520a972cb..a43202892d 100644
--- a/deps/rabbitmq_federation/src/rabbit_federation_exchange_link.erl
+++ b/deps/rabbitmq_federation/src/rabbit_federation_exchange_link.erl
@@ -62,6 +62,9 @@ list_routing_keys(XN) -> call(XN, list_routing_keys).
 start_link(Args) ->
     gen_server2:start_link(?MODULE, Args, [{timeout, infinity}]).
 
+federation_up() ->
+    is_pid(whereis(rabbit_federation_app)).
+
 init({Upstream, XName}) ->
     %% If we are starting up due to a policy change then it's possible
     %% for the exchange to have been deleted before we got here, in which
@@ -88,10 +91,12 @@ handle_call(list_routing_keys, _From, State = #state{bindings = Bindings}) ->
 handle_call(Msg, _From, State) ->
     {stop, {unexpected_call, Msg}, State}.
 
-handle_cast(maybe_go, S0 = {not_started, _Args}) ->
+handle_cast(maybe_go, State = {not_started, _Args}) ->
     case federation_up() of
-        true  -> go(S0);
-        false -> {noreply, S0}
+        true  -> go(State);
+        false ->
+            _ = timer:apply_after(1000, ?MODULE, go, []),
+            {noreply, State}
     end;
 
 handle_cast(go, S0 = {not_started, _Args}) ->
@@ -256,9 +261,6 @@ x(XName) ->
 
 %%----------------------------------------------------------------------------
 
-federation_up() ->
-    is_pid(whereis(rabbit_federation_app)).
-
 handle_command({add_binding, Binding}, State) ->
     add_binding(Binding, State);
 
diff --git a/deps/rabbitmq_federation/src/rabbit_federation_queue_link.erl b/deps/rabbitmq_federation/src/rabbit_federation_queue_link.erl
index 17a2c0a178..d723d86aec 100644
--- a/deps/rabbitmq_federation/src/rabbit_federation_queue_link.erl
+++ b/deps/rabbitmq_federation/src/rabbit_federation_queue_link.erl
@@ -53,11 +53,11 @@ q(QName) ->
             Members
     end.
 
+%%----------------------------------------------------------------------------
+
 federation_up() ->
     is_pid(whereis(rabbit_federation_app)).
 
-%%----------------------------------------------------------------------------
-
 init({Upstream, Queue}) when ?is_amqqueue(Queue) ->
     QName = amqqueue:get_name(Queue),
     case rabbit_amqqueue:lookup(QName) of
@@ -85,7 +85,9 @@ handle_call(Msg, _From, State) ->
 handle_cast(maybe_go, State) ->
     case federation_up() of
         true  -> go(State);
-        false -> {noreply, State}
+        false ->
+            _ = timer:apply_after(1000, ?MODULE, go, []),
+            {noreply, State}
     end;
 
 handle_cast(go, State = #not_started{}) ->