cache: entry TTL in milliseconds

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

Author: noxdafox

README.md

@@ -48,7 +48,7 @@ Extra arguments:
 
   * `x-cache-size`: maximum size for the deduplication cache.
     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 messages 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.
@@ -58,7 +58,7 @@ Message deduplication
 
 Each message containing the `x-deduplication-header` header will not be routed if its value has been already submitted previously and has not expired.
 
-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.
+The optional header `x-cache-ttl` will override the default one if provided during the exchange declaration. This parameter controls for how many milliseconds to deduplicate the message. After the TTL expires, a new message with the same `x-deduplication-header` header will be routed again.
 
 Running the tests
 -----------------

lib/cache.ex

@@ -35,7 +35,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.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})
@@ -175,16 +175,16 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.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
 
@@ -213,8 +213,8 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.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

test/cache_test.exs

@@ -15,7 +15,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Cache.Test do
     end
 
     cache_options = [size: 1, ttl: nil, persistence: :memory]
-    cache_ttl_options = [size: 1, ttl: 1, persistence: :memory]
+    cache_ttl_options = [size: 1, ttl: Timer.seconds(1), persistence: :memory]
 
     start_supervised!(%{id: cache,
                         start: {MessageCache,
@@ -39,7 +39,7 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Cache.Test do
   test "TTL at insertion", %{cache: cache, cache_ttl: _} do
     assert MessageCache.member?(cache, "foo") == false
 
-    MessageCache.put(cache, "foo", 1)
+    MessageCache.put(cache, "foo", Timer.seconds(1))
     assert MessageCache.member?(cache, "foo") == true
 
     1 |> Timer.seconds() |> Timer.sleep()