issue #28, #30: use new cache APIs in exchange and queue behaviours

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

Author: noxdafox

lib/rabbit_message_deduplication_exchange.ex

@@ -26,15 +26,14 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Exchange do
 
   require RabbitMQ.MessageDeduplicationPlugin.Cache
   require RabbitMQ.MessageDeduplicationPlugin.Common
-  require RabbitMQ.MessageDeduplicationPlugin.Supervisor
 
   alias :rabbit_log, as: RabbitLog
   alias :rabbit_misc, as: RabbitMisc
   alias :rabbit_router, as: RabbitRouter
   alias :rabbit_exchange, as: RabbitExchange
-  alias RabbitMQ.MessageDeduplicationPlugin.Cache, as: MessageCache
   alias RabbitMQ.MessageDeduplicationPlugin.Common, as: Common
-  alias RabbitMQ.MessageDeduplicationPlugin.Supervisor, as: CacheSupervisor
+  alias RabbitMQ.MessageDeduplicationPlugin.Cache, as: Cache
+  alias RabbitMQ.MessageDeduplicationPlugin.CacheManager, as: CacheManager
 
   @behaviour :rabbit_exchange_type
 
@@ -141,19 +140,15 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Exchange do
       "Starting exchange deduplication cache ~s with options ~p~n",
       [cache, options])
 
-    CacheSupervisor.start_cache(cache, options)
+    CacheManager.create(cache, options)
   end
 
-  def create(:none, _ex) do
+  def create(_tx, _ex) do
     :ok
   end
 
   def delete(:transaction, exchange(name: name), _bs) do
-    cache = Common.cache_name(name)
-
-    :ok = MessageCache.drop(cache)
-
-    CacheSupervisor.stop_cache(cache)
+    name |> Common.cache_name() |> CacheManager.destroy()
   end
 
   def delete(:none, _ex, _bs) do
@@ -181,7 +176,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Exchange do
   end
 
   def info(exchange(name: name), [:cache_info]) do
-    [cache_info: name |> Common.cache_name() |> MessageCache.info()]
+    [cache_info: name |> Common.cache_name() |> Cache.info()]
   end
 
   def info(_ex, _it) do

lib/rabbit_message_deduplication_queue.ex

@@ -2,7 +2,7 @@
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 #
-# Copyright (c) 2017-2018, Matteo Cafasso.
+# Copyright (c) 2017-2019, Matteo Cafasso.
 # All rights reserved.
 
 defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
@@ -26,12 +26,11 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
 
   require RabbitMQ.MessageDeduplicationPlugin.Cache
   require RabbitMQ.MessageDeduplicationPlugin.Common
-  require RabbitMQ.MessageDeduplicationPlugin.Supervisor
 
   alias :rabbit_log, as: RabbitLog
-  alias RabbitMQ.MessageDeduplicationPlugin.Cache, as: MessageCache
   alias RabbitMQ.MessageDeduplicationPlugin.Common, as: Common
-  alias RabbitMQ.MessageDeduplicationPlugin.Supervisor, as: CacheSupervisor
+  alias RabbitMQ.MessageDeduplicationPlugin.Cache, as: Cache
+  alias RabbitMQ.MessageDeduplicationPlugin.CacheManager, as: CacheManager
 
   @behaviour :rabbit_backing_queue
 
@@ -40,8 +39,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
     accumulate: true, persist: true
 
   @rabbit_boot_step {__MODULE__,
-                     [{:description,
-                       "message deduplication queue: supervisor"},
+                     [{:description, "message deduplication queue"},
                       {:mfa, {__MODULE__, :enable_deduplication_queues, []}},
                       {:requires, :database},
                       {:enables, :external_infrastructure}]}
@@ -134,7 +132,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
         "Starting queue deduplication cache ~s with options ~p~n",
         [cache, options])
 
-      CacheSupervisor.start_cache(cache, options)
+      CacheManager.create(cache, options)
     end
 
     passthrough1(dqstate(queue: queue)) do
@@ -150,10 +148,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
     dqstate(queue: queue, queue_state: qs) = state
 
     if duplicate?(queue) do
-      cache = queue |> amqqueue(:name) |> Common.cache_name()
-
-      :ok = MessageCache.drop(cache)
-      :ok = CacheSupervisor.stop_cache(cache)
+      queue |> amqqueue(:name) |> Common.cache_name() |> CacheManager.destroy()
     end
 
     passthrough1(state) do
@@ -169,7 +164,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
     if duplicate?(queue) do
       cache = queue |> amqqueue(:name) |> Common.cache_name()
 
-      :ok = MessageCache.flush(cache)
+      :ok = Cache.flush(cache)
     end
 
     passthrough2(state, do: purge(qs))
@@ -437,7 +432,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
     queue
     |> amqqueue(:name)
     |> Common.cache_name()
-    |> MessageCache.delete(header)
+    |> Cache.delete(header)
   end
 
   defp maybe_delete_cache_entry(_queue, header) when is_nil(header) do end
@@ -447,7 +442,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Queue do
     cache = Common.cache_name(name)
 
     try do
-      MessageCache.info(cache)
+      Cache.info(cache)
     catch
       :exit, {:noproc, {GenServer, :call, [^cache | _]}} -> []
     end