Merge pull request #4 from noxdafox/queue

Queue-level de duplication support

Author: noxdafox

README.md

@@ -1,12 +1,10 @@
-RabbitMQ Message Deduplication Plugin
-=====================================
+# RabbitMQ Message Deduplication Plugin
 
 A plugin for filtering duplicate messages.
 
-Exchange Type: `x-message-deduplication`
+Messages can be deduplicated when published into an exchange or enqueued to a queue.
 
-Installing
-----------
+## Installing
 
 Supported RabbitMQ versions:
 
@@ -20,8 +18,7 @@ Enable the plugin:
     [sudo] rabbitmq-plugins enable rabbitmq_message_deduplication_exchange
 ```
 
-Building from Source
---------------------
+## Building from Source
 
 Please see RabbitMQ Plugin Development guide.
 
@@ -39,35 +36,55 @@ Then copy all the *.ez files inside the plugins folder to the [RabbitMQ plugins
     [sudo] rabbitmq-plugins enable rabbitmq_message_deduplication_exchange
 ```
 
-Declare an exchange
--------------------
+## Exchange level deduplication
+
+The exchange type `x-message-deduplication` allows to filter message duplicates before any routing rule is applied.
+
+Each message containing the `x-deduplication-header` header will not be routed if its value has been submitted previously. The amount of time a given message will be guaranteed to be unique can be controlled via the `x-cache-ttl` exchange argument or message header.
+
+### Declare an exchange
 
 To create a message deduplication exchange, just declare it providing the type `x-message-deduplication`.
 
 Extra arguments:
 
-  * `x-cache-size`: maximum size for the deduplication cache.
+  * `x-cache-size`: maximum size for the deduplication cache. If the deduplication cache fills up, older entries will be removed to give space to new ones.
     This parameter is mandatory.
-  * `x-cache-ttl`: amount of time in seconds messages are kept in cache.
+  * `x-cache-ttl`: amount of time in milliseconds duplicate headers are kept in cache.
     This parameter is optional.
-  * `x-cache-persistence`: whether the cache will persist on disk or in memory.
-    This parameter is optional. Default persistence is memory.
+  * `x-cache-persistence`: whether the duplicates cache will persist on disk or in memory.
+    This parameter is optional. Default persistence type is `memory`.
+
+### Message headers
+
+  * `x-deduplication-header`: messages will be deduplicated based on the content of this header. If the header is not provided, the message will not be checked against duplicates.
+  * `x-cache-ttl`: this header is optional and will override the default value provided during the exchange declaration. This header controls for how many milliseconds to deduplicate the message. After the TTL expires, a new message with the same header will be routed again.
+
+## Queue level deduplication
+
+A queue declared with the `x-message-deduplication` parameter enabled will filter message duplicates before they are published within.
+
+Each message containing the `x-deduplication-header` header will not be enqueued if another message with the same header is already present within the queue.
+
+### Declare a queue
+
+When declaring a queue, it is possible to enable message deduplication via the `x-message-deduplication` boolean argument.
+
+Extra arguments:
 
-Message deduplication
----------------------
+  * `x-cache-persistence`: whether the duplicates cache will persist on disk or in memory.
+    This parameter is optional. Default persistence type is `memory`.
 
-Each message containing the `x-deduplication-header` header will not be routed if its value has been already submitted previously and has not expired.
+### Message headers
 
-The optional header `x-cache-ttl` will override the default one if provided during the exchange declaration. This parameter controls for how many seconds to deduplicate the message. After the TTL expires, a new message with the same `x-deduplication-header` header will be routed again.
+  * `x-deduplication-header`: messages will be deduplicated based on the content of this header. If the header is not provided, the message will not be checked against duplicates.
 
-Running the tests
------------------
+## Running the tests
 
 ```bash
     make tests
 ```
 
-License
--------
+## License
 
 See the LICENSE file.

lib/cache.ex

@@ -6,7 +6,7 @@
 # All rights reserved.
 
 
-defmodule RabbitMQ.Cache do
+defmodule RabbitMQ.MessageDeduplicationPlugin.Cache do
   @moduledoc """
   Simple cache implemented on top of Mnesia.
 
@@ -35,12 +35,19 @@ defmodule RabbitMQ.Cache do
   end
 
   @doc """
-  Put the given value into the cache.
+  Put the given entry into the cache.
   """
   def put(cache, value, ttl \\ nil) do
     GenServer.call(cache, {:put, cache, value, ttl})
   end
 
+  @doc """
+  Delete the given value from the cache.
+  """
+  def delete(cache, value) do
+    GenServer.call(cache, {:delete, cache, value})
+  end
+
   @doc """
   True if the value is contained within the cache.
   """
@@ -49,10 +56,10 @@ defmodule RabbitMQ.Cache do
   end
 
   @doc """
-  Return information related to the given cache.
+  Flush the cache content.
   """
-  def info(cache) do
-    GenServer.call(cache, {:info, cache})
+  def flush(cache) do
+    GenServer.call(cache, {:flush, cache})
   end
 
   @doc """
@@ -62,6 +69,13 @@ defmodule RabbitMQ.Cache do
     GenServer.call(cache, {:drop, cache})
   end
 
+  @doc """
+  Return information related to the given cache.
+  """
+  def info(cache) do
+    GenServer.call(cache, {:info, cache})
+  end
+
   ## Server Callbacks
 
   # Creates the Mnesia table and starts the janitor process.
@@ -99,17 +113,26 @@ defmodule RabbitMQ.Cache do
     {:reply, :ok, state}
   end
 
+  # Removes the given entry from the cache.
+  def handle_call({:delete, cache, value}, _from, state) do
+    Mnesia.transaction(fn ->
+      Mnesia.delete({cache, value})
+    end)
+
+    {:reply, :ok, state}
+  end
+
   # True if the value is in the cache.
   def handle_call({:member?, cache, value}, _from, state) do
     {:reply, cache_member?(cache, value), state}
   end
 
-  # Return cache information: number of elements and max size
-  def handle_call({:info, cache}, _from, state) do
-    info = [size: cache_property(cache, :limit),
-            entries: Mnesia.table_info(cache, :size)]
-
-    {:reply, info, state}
+  # Flush the Mnesia cache table.
+  def handle_call({:flush, cache}, _from, state) do
+    case Mnesia.clear_table(cache) do
+      {:atomic, :ok} -> {:reply, :ok, state}
+      _ -> {:reply, :error, state}
+    end
   end
 
   # Drop the Mnesia cache table.
@@ -120,6 +143,14 @@ defmodule RabbitMQ.Cache do
     end
   end
 
+  # Return cache information: number of elements and max size
+  def handle_call({:info, cache}, _from, state) do
+    info = [size: cache_property(cache, :limit),
+            entries: Mnesia.table_info(cache, :size)]
+
+    {:reply, info, state}
+  end
+
   ## Utility functions
 
   # Mnesia cache table creation.
@@ -144,16 +175,16 @@ defmodule RabbitMQ.Cache do
     {:atomic, entries} = Mnesia.transaction(fn -> Mnesia.read(cache, value) end)
 
     case List.keyfind(entries, value, 1) do
-      {_, _, expiration} -> expiration > Os.system_time(:seconds)
+      {_, _, expiration} -> expiration > Os.system_time(:millisecond)
       nil -> false
     end
   end
 
   # Remove all expired entries from the Mnesia cache.
   defp cache_delete_expired(cache) do
     select = fn ->
-      Mnesia.select(cache, [{{cache, :"$1", :_, :"$3"},
-                             [{:>, Os.system_time(:seconds), :"$3"}],
+      Mnesia.select(cache, [{{cache, :"$1", :"$2"},
+                             [{:>, Os.system_time(:millisecond), :"$2"}],
                              [:"$1"]}])
     end
 
@@ -182,8 +213,8 @@ defmodule RabbitMQ.Cache do
     default = cache_property(cache, :default_ttl)
 
     cond do
-      ttl != nil -> Os.system_time(:seconds) + ttl
-      default != nil -> Os.system_time(:seconds) + default
+      ttl != nil -> Os.system_time(:millisecond) + ttl
+      default != nil -> Os.system_time(:millisecond) + default
       true -> nil
     end
   end

lib/rabbit_message_deduplication_exchange.ex

@@ -6,15 +6,18 @@
 # All rights reserved.
 
 
-defmodule RabbitMQ.MessageDeduplicationExchangeType do
+defmodule RabbitMQ.MessageDeduplicationPlugin.Exchange do
   import Record, only: [defrecord: 2, extract: 2]
 
-  require RabbitMQ.Cache
-  require RabbitMQ.CacheSupervisor
+  require RabbitMQ.MessageDeduplicationPlugin.Cache
+  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.Supervisor, as: CacheSupervisor
 
   @behaviour :rabbit_exchange_type
 
@@ -30,13 +33,6 @@ defmodule RabbitMQ.MessageDeduplicationExchangeType do
                       {:requires, :rabbit_registry},
                       {:enables, :kernel_ready}]}
 
-  @rabbit_boot_step {:rabbit_exchange_type_caches_supervisor,
-                     [{:description,
-                       "message deduplication exchange type: supervisor"},
-                      {:mfa, {__MODULE__, :start_caches_supervisor, []}},
-                      {:requires, :database},
-                      {:enables, :external_infrastructure}]}
-
   defrecord :exchange, extract(
     :exchange, from_lib: "rabbit_common/include/rabbit.hrl")
 
@@ -49,10 +45,6 @@ defmodule RabbitMQ.MessageDeduplicationExchangeType do
   defrecord :basic_message, extract(
     :basic_message, from_lib: "rabbit_common/include/rabbit.hrl")
 
-  def start_caches_supervisor() do
-    RabbitMQ.CacheSupervisor.start_link()
-  end
-
   def description() do
     [
       {:name, <<"x-message-deduplication">>},
@@ -119,7 +111,9 @@ defmodule RabbitMQ.MessageDeduplicationExchangeType do
       |> String.to_atom()
     options = [size: size, ttl: ttl, persistence: persistence]
 
-    RabbitMQ.CacheSupervisor.start_cache(cache, options)
+    RabbitLog.debug("Starting exchange deduplication cache ~s~n", [cache])
+
+    CacheSupervisor.start_cache(cache, options)
   end
 
   def create(:none, _ex) do
@@ -129,9 +123,9 @@ defmodule RabbitMQ.MessageDeduplicationExchangeType do
   def delete(:transaction, exchange(name: name), _bs) do
     cache = cache_name(name)
 
-    :ok = RabbitMQ.Cache.drop(cache)
+    :ok = MessageCache.drop(cache)
 
-    RabbitMQ.CacheSupervisor.stop_cache(cache)
+    CacheSupervisor.stop_cache(cache)
   end
 
   def delete(:none, _ex, _bs) do
@@ -159,7 +153,7 @@ defmodule RabbitMQ.MessageDeduplicationExchangeType do
   end
 
   def info(exchange(name: name), [:cache_info]) do
-    [cache_info: name |> cache_name() |> RabbitMQ.Cache.info()]
+    [cache_info: name |> cache_name() |> MessageCache.info()]
   end
 
   def info(_ex, _it) do
@@ -185,7 +179,7 @@ defmodule RabbitMQ.MessageDeduplicationExchangeType do
   defp cached?(cache, headers) do
     case rabbitmq_keyfind(headers, "x-deduplication-header") do
       nil -> false
-      key -> case RabbitMQ.Cache.member?(cache, key) do
+      key -> case MessageCache.member?(cache, key) do
                true -> true
                false -> cache_put(cache, key, headers)
                         false
@@ -196,8 +190,8 @@ defmodule RabbitMQ.MessageDeduplicationExchangeType do
   # Puts the key and related headers in the cache
   defp cache_put(cache, key, headers) do
     case rabbitmq_keyfind(headers, "x-cache-ttl") do
-      nil -> RabbitMQ.Cache.put(cache, key)
-      ttl -> RabbitMQ.Cache.put(cache, key, ttl)
+      nil -> MessageCache.put(cache, key)
+      ttl -> MessageCache.put(cache, key, ttl)
     end
   end
 
@@ -206,7 +200,7 @@ defmodule RabbitMQ.MessageDeduplicationExchangeType do
     resource = sanitize_string(resource)
     exchange = sanitize_string(exchange)
 
-    String.to_atom("cache_#{resource}_#{exchange}")
+    String.to_atom("cache_exchange_#{resource}_#{exchange}")
   end
 
   # Returns the value given a key from a RabbitMQ list [{"key", :type, value}]