issue-97: handle cache reconfiguration at startup

Exchange and Queue modules now make sure old caches are reconfigured
at startup.

Signed-off-by: Matteo Cafasso <[email protected]>

Author: noxdafox

lib/rabbitmq_message_deduplication/rabbit_message_deduplication_exchange.ex

@@ -43,6 +43,7 @@ defmodule RabbitMQMessageDeduplication.Exchange do
     :rabbit_boot_step,
     accumulate: true, persist: true)
 
+  @exchange_type <<"x-message-deduplication">>
   @rabbit_boot_step {__MODULE__,
                      [{:description, "exchange type x-message-deduplication"},
                       {:mfa, {__MODULE__, :register, []}},
@@ -63,21 +64,22 @@ defmodule RabbitMQMessageDeduplication.Exchange do
   """
   @spec register() :: :ok
   def register() do
-    RabbitRegistry.register(:exchange, <<"x-message-deduplication">>, __MODULE__)
+    RabbitRegistry.register(:exchange, @exchange_type, __MODULE__)
+    maybe_reconfigure_caches()
   end
 
   @doc """
   Unregister the exchange type from the Broker.
   """
   @spec unregister() :: :ok
   def unregister() do
-    RabbitRegistry.unregister(:exchange, <<"x-message-deduplication">>)
+    RabbitRegistry.unregister(:exchange, @exchange_type)
   end
 
   @impl :rabbit_exchange_type
   def description() do
     [
-      {:name, <<"x-message-deduplication">>},
+      {:name, @exchange_type},
       {:description, <<"Message Deduplication Exchange.">>}
     ]
   end
@@ -165,9 +167,7 @@ defmodule RabbitMQMessageDeduplication.Exchange do
     name |> Common.cache_name() |> CacheManager.destroy()
   end
 
-  def delete(:transaction, exchange, _bs), do: delete(:none, exchange)
-
-  def delete(:none, _ex, _bs), do: :ok
+  def delete(_tx, exchange, _bs), do: delete(:none, exchange)
 
   @impl :rabbit_exchange_type
   def policy_changed(_ex, exchange(name: name, arguments: args, policy: :undefined)) do
@@ -249,4 +249,15 @@ defmodule RabbitMQMessageDeduplication.Exchange do
       Cache.change_option(cache, key, value)
     end
   end
+
+  # Caches created prior to v0.6.0 need to be reconfigured.
+  defp maybe_reconfigure_caches() do
+    RabbitLog.debug("Deduplication Exchanges startup, reconfiguring old caches")
+
+    RabbitExchange.list()
+    |> Enum.filter(fn(exchange(name: type)) -> type == @exchange_type end)
+    |> Enum.map(fn(exchange) -> create(:none, exchange) end)
+
+    :ok
+  end
 end

lib/rabbitmq_message_deduplication/rabbit_message_deduplication_queue.ex

@@ -29,6 +29,7 @@ defmodule RabbitMQMessageDeduplication.Queue do
 
   alias :amqqueue, as: AMQQueue
   alias :rabbit_log, as: RabbitLog
+  alias :rabbit_amqqueue, as: RabbitQueue
   alias RabbitMQMessageDeduplication.Common, as: Common
   alias RabbitMQMessageDeduplication.Cache, as: Cache
   alias RabbitMQMessageDeduplication.CacheManager, as: CacheManager
@@ -116,6 +117,7 @@ defmodule RabbitMQMessageDeduplication.Queue do
           "Deduplication queues enabled, real BQ is ~s~n", [backing_queue])
         Application.put_env(__MODULE__, :backing_queue_module, backing_queue)
         Application.put_env(:rabbit, :backing_queue_module, __MODULE__)
+        maybe_reconfigure_caches()
     end
   end
 
@@ -474,10 +476,22 @@ defmodule RabbitMQMessageDeduplication.Queue do
       disable_dedup_queue?(state) ->
         :ok = delete_cache(queue)
         dqstate(queue: queue, queue_state: qs, dedup_enabled: false)
-      true -> dqstate(queue: queue, queue_state: qs, dedup_enabled: false)
+      true ->
+        dqstate(queue: queue, queue_state: qs, dedup_enabled: false)
     end
   end
 
+  # Caches created prior to v0.6.0 need to be reconfigured.
+  defp maybe_reconfigure_caches() do
+    RabbitLog.debug("Deduplication Queues startup, reconfiguring old caches")
+
+    RabbitQueue.list()
+    |> Enum.filter(&dedup_arg?/1)
+    |> Enum.map(&init_cache/1)
+
+    :ok
+  end
+
   # Initialize the deduplication cache
   defp init_cache(queue) do
     cache = queue |> AMQQueue.get_name() |> Common.cache_name()