Code cleanup

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

Author: noxdafox

lib/cache.ex

@@ -13,8 +13,6 @@ defmodule RabbitMQ.Cache do
 
   use GenServer
 
-  require Record
-
   alias :os, as: Os
   alias :timer, as: Timer
   alias :mnesia, as: Mnesia
@@ -49,60 +47,49 @@ defmodule RabbitMQ.Cache do
     GenServer.call(cache, {:drop, cache})
   end
 
-  @doc """
-  Return the PID of the given cache, nil if non existing.
-  """
-  def process(cache) do
-    GenServer.whereis(cache)
-  end
-
   ## Server Callbacks
 
+  # Creates the Mnesia table and starts the janitor process.
   def init({cache, options}) do
     Mnesia.start()
 
     :ok = cache_create(cache, options)
 
-    Process.send_after(cache, {:cache, cache}, 3000)
+    Process.send_after(cache, {:cache, cache}, Timer.seconds(3))
 
     {:ok, %{}}
   end
 
+  # The janitor process deletes expired cache entries.
   def handle_info({:cache, cache}, state) do
     {_, result} = cache_delete_expired(cache)
     if (result == :ok) do
-      Process.send_after(cache, {:cache, cache}, 3000)
+      Process.send_after(cache, {:cache, cache}, Timer.seconds(3))
     end
 
     {:noreply, state}
   end
 
+  # Puts a new entry in the cache.
+  # If the cache is full, remove an element to make space.
   def handle_call({:put, cache, value, ttl}, _from, state) do
-    {:default_ttl, default_ttl} = cache_default_ttl(cache)
-    expiration = cond do
-      ttl != nil -> Os.system_time(:seconds) + ttl
-      default_ttl != nil -> Os.system_time(:seconds) + default_ttl
-      true -> nil
-    end
-
-    # Remove first element if cache is full
-    size = Mnesia.table_info(cache, :size)
-    {:limit, limit} = cache_limit(cache)
-    if size >= limit do
+    if cache_full?(cache) do
       cache_delete_first(cache)
     end
 
     Mnesia.transaction(fn ->
-      Mnesia.write({cache, value, expiration})
+      Mnesia.write({cache, value, entry_expiration(cache, ttl)})
     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
 
+  # Drop the Mnesia cache table.
   def handle_call({:drop, cache}, _from, state) do
     case Mnesia.delete_table(cache) do
       {:atomic, :ok} -> {:reply, :ok, state}
@@ -112,6 +99,7 @@ defmodule RabbitMQ.Cache do
 
   ## Utility functions
 
+  # Mnesia cache table creation.
   defp cache_create(cache, options) do
     persistence = case Keyword.get(options, :persistence) do
                     :disk -> :disc_copies
@@ -128,6 +116,7 @@ defmodule RabbitMQ.Cache do
     Mnesia.wait_for_tables([cache], Timer.seconds(30))
   end
 
+  # Mnesia cache lookup. The entry is not returned if expired.
   defp cache_member?(cache, value) do
     {:atomic, entries} = Mnesia.transaction(fn -> Mnesia.read(cache, value) end)
 
@@ -137,6 +126,7 @@ defmodule RabbitMQ.Cache do
     end
   end
 
+  # Remove all expired entries from the Mnesia cache.
   defp cache_delete_expired(cache) do
     select = fn ->
       Mnesia.select(cache, [{{cache, :"$1", :_, :"$3"},
@@ -154,20 +144,33 @@ defmodule RabbitMQ.Cache do
     end
   end
 
+  # Delete the first element from the cache.
+  # As the Mnesia Set is not ordered, the first element is random.
   defp cache_delete_first(cache) do
-    Mnesia.transaction(
-      fn ->
-        Mnesia.delete({cache, Mnesia.first(cache)})
-      end)
+    Mnesia.transaction(fn -> Mnesia.delete({cache, Mnesia.first(cache)}) end)
+  end
+
+  defp cache_full?(cache) do
+    Mnesia.table_info(cache, :size) >= cache_property(cache, :limit)
   end
 
-  defp cache_limit(cache) do
-    Enum.find(Mnesia.table_info(cache, :user_properties),
-      fn(element) -> match?({:limit, _}, element) end)
+  # Calculate the expiration given a TTL or the cache default TTL
+  defp entry_expiration(cache, ttl) do
+    default = cache_property(cache, :default_ttl)
+
+    cond do
+      ttl != nil -> Os.system_time(:seconds) + ttl
+      default != nil -> Os.system_time(:seconds) + default
+      true -> nil
+    end
   end
 
-  defp cache_default_ttl(cache) do
-    Enum.find(Mnesia.table_info(cache, :user_properties),
-      fn(element) -> match?({:default_ttl, _}, element) end)
+  defp cache_property(cache, property) do
+    {^property, value} =
+      cache
+      |> Mnesia.table_info(:user_properties)
+      |> Enum.find(fn(element) -> match?({^property, _}, element) end)
+
+    value
   end
 end

lib/rabbit_message_deduplication_exchange.ex

@@ -160,7 +160,7 @@ defmodule RabbitMQ.ExchangeTypeMessageDeduplication do
   # Whether to route the message or not.
   defp route?(cache, message) do
     case message_headers(message) do
-      headers when is_list(headers) -> not cached_message?(cache, headers)
+      headers when is_list(headers) -> not cached?(cache, headers)
       :undefined -> true
     end
   end
@@ -171,7 +171,7 @@ defmodule RabbitMQ.ExchangeTypeMessageDeduplication do
   # false otherwise.
   #
   # If `x-deduplication-header` value is not present in the cache, it is added.
-  defp cached_message?(cache, headers) do
+  defp cached?(cache, headers) do
     case rabbitmq_keyfind(headers, "x-deduplication-header") do
       nil -> false
       key ->
@@ -189,23 +189,10 @@ defmodule RabbitMQ.ExchangeTypeMessageDeduplication do
     end
   end
 
-  # Unpacks the message headers
-  defp message_headers(message) do
-    message |> elem(2) |> elem(3)
-  end
-
   # Returns a sanitized atom composed by the resource and exchange name
   defp cache_name({:resource, resource, :exchange, exchange}) do
-    resource =
-      resource
-      |> String.replace(~r/[-\. ]/, "_")
-      |> String.replace("/", "")
-      |> String.downcase()
-    exchange =
-      exchange
-      |> String.replace(~r/[-\. ]/, "_")
-      |> String.replace("/", "")
-      |> String.downcase()
+    resource = sanitize_string(resource)
+    exchange = sanitize_string(exchange)
 
     String.to_atom("cache_#{resource}_#{exchange}")
   end
@@ -217,4 +204,16 @@ defmodule RabbitMQ.ExchangeTypeMessageDeduplication do
       _ -> default
     end
   end
+
+  # Unpacks the message headers
+  defp message_headers(message) do
+    message |> elem(2) |> elem(3)
+  end
+
+  defp sanitize_string(string) do
+    string
+    |> String.replace(~r/[-\. ]/, "_")
+    |> String.replace("/", "")
+    |> String.downcase()
+  end
 end

lib/supervisor.ex

@@ -1,15 +1,23 @@
 defmodule RabbitMQ.CacheSupervisor do
-  use DynamicSupervisor
+  @moduledoc """
+  The Cache Supervisor supervisions the Cache GenServer Processes.
+  """
 
-  require RabbitMQ.Cache
+  use DynamicSupervisor
 
+  @doc """
+  Start the Supervisor process.
+  """
   def start_link() do
     case DynamicSupervisor.start_link(__MODULE__, [], name: __MODULE__) do
       {:ok, _pid} -> :ok
       _ -> :error
     end
   end
 
+  @doc """
+  Starts a new cache with the given name and options.
+  """
   def start_cache(cache, options) do
     specifications = %{id: cache,
                        start: {RabbitMQ.Cache, :start_link, [cache, options]}}
@@ -21,10 +29,16 @@ defmodule RabbitMQ.CacheSupervisor do
     end
   end
 
+  @doc """
+  Stops the given cache.
+  """
   def stop_cache(cache) do
-    DynamicSupervisor.terminate_child(__MODULE__, RabbitMQ.Cache.process(cache))
+    DynamicSupervisor.terminate_child(__MODULE__, Process.whereis(cache))
   end
 
+  @doc """
+  Supervisor initialization callback.
+  """
   def init(_args) do
     DynamicSupervisor.init(strategy: :one_for_one)
   end