Implemented durable routes

2 files changed, 33 insertions, 6 deletions

diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 2eaf1ad99f..bcd5499a6e 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -85,9 +85,22 @@
 %%----------------------------------------------------------------------------
 
 recover() ->
+    % These two functions share commonalities - maybe a refactoring target
     ok = recover_durable_exchanges(),
+    ok = recover_durable_routes(),
     ok.
 
+recover_durable_routes() ->
+    rabbit_misc:execute_mnesia_transaction(
+      fun () ->
+              mnesia:foldl(fun (Route, Acc) ->
+                                {_, ReverseRoute} = route_with_reverse(Route),
+                                ok = mnesia:write(Route),
+                                ok = mnesia:write(ReverseRoute),
+                                Acc
+                            end, ok, durable_routes)
+      end).
+
 recover_durable_exchanges() ->
     rabbit_misc:execute_mnesia_transaction(
       fun () ->
@@ -264,10 +277,10 @@ delete_binding(QueueName, ExchangeName, RoutingKey, _Arguments) ->
 add_binding(Binding) ->
     call_with_exchange_and_queue(
         Binding,
-        fun (X,Q) -> if Q#amqqueue.durable and not(X#exchange.durable) ->
+        fun (X, Q) -> if Q#amqqueue.durable and not(X#exchange.durable) ->
                             {error, durability_settings_incompatible};
                          true ->
-                             internal_add_binding(Binding)
+                             internal_add_binding(Binding, Q#amqqueue.durable)
                      end
         end).
 
@@ -275,7 +288,7 @@ add_binding(Binding) ->
 delete_binding(Binding) ->
     call_with_exchange_and_queue(
          Binding,
-         fun (X, _Q) -> ok = internal_delete_binding(Binding),
+         fun (X, Q) -> ok = internal_delete_binding(Binding, Q#amqqueue.durable),
                       maybe_auto_delete(X)
          end).
 
@@ -295,16 +308,26 @@ reverse_binding(#binding{exchange_name = Exchange, key = Key, queue_name = Queue
     #reverse_binding{exchange_name = Exchange, key = Key, queue_name = Queue}.
 
 %% Must run within a transaction.
-internal_add_binding(Binding) ->
+internal_add_binding(Binding, Durable) ->
+    synchronize_durable_binding(Binding, write, Durable),
     [ok, ok] = [mnesia:write(R) || R <- tuple_to_list(route_with_reverse(Binding))],
     ok.
 
 %% Must run within a transaction.
-internal_delete_binding(Binding) ->
+internal_delete_binding(Binding, Durable) ->
+    synchronize_durable_binding(Binding, delete, Durable),
     [ok,ok] = [mnesia:delete_object(R) || R <- tuple_to_list(route_with_reverse(Binding))],
     ok.
 
-route_with_reverse(Binding) ->
+synchronize_durable_binding(_, _, false) ->
+    ok;
+%% Must run within a transaction.
+synchronize_durable_binding(Binding, Function, true) ->
+    ok = mnesia:Function(durable_routes, #route{binding = Binding}, write).
+
+route_with_reverse(#route{binding = Binding}) ->
+    route_with_reverse(Binding);
+route_with_reverse(Binding = #binding{}) ->
     Route = #route{ binding = Binding },
     ReverseRoute = #reverse_route{ reverse_binding = reverse_binding(Binding) },
     {Route, ReverseRoute}.
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 3e031c3b31..958c380b2c 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -105,6 +105,10 @@ table_definitions() ->
      {rabbit_config, [{disc_copies, [node()]}]},
      {listener, [{type, bag},
                  {attributes, record_info(fields, listener)}]},
+     {durable_routes, [{disc_copies, [node()]},
+                       {type,ordered_set},
+                       {record_name, route},
+                       {attributes, record_info(fields, route)}]},
      {route, [{type,ordered_set},{attributes, record_info(fields, route)}]},
      {reverse_route, [{type,ordered_set},{attributes, record_info(fields, reverse_route)}]},
      {durable_exchanges, [{disc_copies, [node()]},