issue #76: revert to broker boot steps to initialize the plugin

As per commit
rabbitmq/rabbitmq-server@a0cd2e5,
the original issue within the broker was fixed.

Hence, the plugin is reverted to the original boot sequence as it is
more lightweight compared to the previous one.

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

Author: noxdafox

lib/rabbitmq_message_deduplication.ex

@@ -9,25 +9,21 @@ defmodule RabbitMQMessageDeduplication do
 
   use Application
 
+  # Start a dummy supervisor to enable the Application behaviour.
+  # http://erlang.org/pipermail/erlang-questions/2010-April/050508.html
+  @impl true
   def start(_, _) do
-    children = [
-      %{
-        id: RabbitMQMessageDeduplication.CacheManager,
-        start: {RabbitMQMessageDeduplication.CacheManager, :start_link, []}
-      }
-    ]
-
-    supervisor = Supervisor.start_link(children, [strategy: :one_for_one])
-
-    RabbitMQMessageDeduplication.Exchange.register()
-    RabbitMQMessageDeduplication.Queue.enable()
-    RabbitMQMessageDeduplication.Queue.restart_queues()
-
-    supervisor
+    Supervisor.start_link(__MODULE__, [], [])
   end
 
+  @impl true
   def stop(_) do
     RabbitMQMessageDeduplication.Exchange.unregister()
     RabbitMQMessageDeduplication.Queue.disable()
   end
+
+  @impl true
+  def init([]) do
+    Supervisor.init([], strategy: :one_for_one)
+  end
 end

lib/rabbitmq_message_deduplication/cache_manager.ex

@@ -18,6 +18,17 @@ defmodule RabbitMQMessageDeduplication.CacheManager do
   alias :mnesia, as: Mnesia
   alias RabbitMQMessageDeduplication.Cache, as: Cache
 
+  Module.register_attribute(__MODULE__,
+    :rabbit_boot_step,
+    accumulate: true, persist: true)
+
+  @rabbit_boot_step {
+    __MODULE__,
+    [description: "message deduplication plugin cache maintenance process",
+     mfa: {:rabbit_sup, :start_child, [__MODULE__]},
+     requires: :database,
+     enables: :external_infrastructure]}
+
   @caches :message_deduplication_caches
   @cleanup_period Timer.seconds(3)
   @table_wait_time Timer.seconds(30)

lib/rabbitmq_message_deduplication/rabbit_message_deduplication_exchange.ex

@@ -38,6 +38,16 @@ defmodule RabbitMQMessageDeduplication.Exchange do
 
   @behaviour :rabbit_exchange_type
 
+  Module.register_attribute(__MODULE__,
+    :rabbit_boot_step,
+    accumulate: true, persist: true)
+
+  @rabbit_boot_step {__MODULE__,
+                     [{:description, "exchange type x-message-deduplication"},
+                      {:mfa, {__MODULE__, :register, []}},
+                      {:requires, :rabbit_registry},
+                      {:enables, :kernel_ready}]}
+
   defrecord :exchange, extract(
     :exchange, from_lib: "rabbit_common/include/rabbit.hrl")
 

lib/rabbitmq_message_deduplication/rabbit_message_deduplication_queue.ex

@@ -29,15 +29,22 @@ defmodule RabbitMQMessageDeduplication.Queue do
 
   alias :amqqueue, as: AMQQueue
   alias :rabbit_log, as: RabbitLog
-  alias :supervisor2, as: Supervisor2
-  alias :rabbit_amqqueue, as: RabbitAMQQueue
-  alias :rabbit_amqqueue_sup_sup, as: RabbitAMQQueueSupervisor
   alias RabbitMQMessageDeduplication.Common, as: Common
   alias RabbitMQMessageDeduplication.Cache, as: Cache
   alias RabbitMQMessageDeduplication.CacheManager, as: CacheManager
 
   @behaviour :rabbit_backing_queue
 
+  Module.register_attribute(__MODULE__,
+    :rabbit_boot_step,
+    accumulate: true, persist: true)
+
+  @rabbit_boot_step {__MODULE__,
+                     [{:description, "message deduplication queue"},
+                      {:mfa, {__MODULE__, :enable, []}},
+                      {:requires, :database},
+                      {:enables, :external_infrastructure}]}
+
   defrecord :content, extract(
     :content, from_lib: "rabbit_common/include/rabbit.hrl")
 
@@ -128,27 +135,6 @@ defmodule RabbitMQMessageDeduplication.Queue do
     end
   end
 
-  @doc """
-  Restart deduplication queues.
-
-  When the broker starts, it initializes all queues before the Plugins.
-  Therefore, we need to restart all queues marked as deduplication
-  to ensure the correct implementation module is loaded.
-
-  The names of the restarted queues is returned.
-
-  """
-  @spec restart_queues() :: list
-  def restart_queues() do
-    RabbitLog.info(
-      "Restarting deduplication queues to load the correct module. " <>
-      "Ignore subsequent error messages.")
-
-    RabbitAMQQueue.list()
-    |> Enum.filter(fn(q) -> duplicate?(q) and local?(q) end)
-    |> Enum.map(fn(q) -> restart_queue_process(q) end)
-  end
-
   @impl :rabbit_backing_queue
   def start(vhost, queues) do
     passthrough do: start(vhost, queues)
@@ -506,39 +492,6 @@ defmodule RabbitMQMessageDeduplication.Queue do
     end
   end
 
-  # Returns true if the queue is local to this node
-  defp local?(queue) do
-    node(AMQQueue.get_pid(queue)) == node()
-  end
-
-  # Restarts the AMQQueue associated process
-  defp restart_queue_process(queue) do
-    qpid = AMQQueue.get_pid(queue)
-    {:resource, vhost, :queue, name} = AMQQueue.get_name(queue)
-    supervisor = find_queue_supervisor(vhost, qpid)
-
-    RabbitLog.debug("Restarting deduplication queue ~s.~n", [name])
-
-    :ok = Supervisor2.terminate_child(supervisor, :rabbit_amqqueue)
-    {:ok, _} = Supervisor2.restart_child(supervisor, :rabbit_amqqueue)
-
-    name
-  end
-
-  defp find_queue_supervisor(vhost, qpid) do
-    {:ok, parent} = RabbitAMQQueueSupervisor.find_for_vhost(vhost)
-
-    Supervisor2.which_children(parent)
-    |> Enum.map(&child_pid/1)
-    |> Enum.map(fn(s) ->
-         {s, Supervisor2.which_children(s) |> Enum.map(&child_pid/1)}
-       end)
-    |> Enum.find_value(fn({s, c}) -> if Enum.member?(c, qpid), do: s end)
-  end
-
-  defp child_pid({:rabbit_amqqueue, pid, _, _}), do: pid
-  defp child_pid({_, pid, _, [:rabbit_amqqueue_sup]}), do: pid
-
   # Returns the expiration property of the given message
   defp message_expiration(message) do
     basic_message(content: content(properties: properties)) = message