Storage adapter implementation: Implemented Quantum.Storage.Adapter behavior via PersistentETS.

Author: pyatkov

config/.credo.exs

@@ -9,6 +9,7 @@
       checks: [
         # For others you can also set parameters
         {Credo.Check.Readability.MaxLineLength, priority: :low, max_length: 120},
+        {Credo.Check.Design.TagTODO, false}
       ]
     }
   ]

lib/quantum_storage_ets.ex

@@ -1,5 +1,301 @@
 defmodule QuantumStorageEts do
   @moduledoc """
-  Documentation for QuantumStorageEts.
+  persistent_ets based implementation of a `Quantum.Storage.Adapter`.
+  See https://hexdocs.pm/persistent_ets
   """
+  require Logger
+  use GenServer
+  defstruct [:schedulers]
+
+  def start_link do
+    GenServer.start_link(__MODULE__, nil, name: __MODULE__)
+  end
+
+  # Callbacks
+
+  defp __server__, do: __MODULE__
+
+  def init(_) do
+    {:ok, %__MODULE__{schedulers: %{}}}
+  end
+
+  def handle_call(
+        {:add_job, scheduler_module, job},
+        _from,
+        %__MODULE__{schedulers: schedulers} = state
+      ) do
+    {
+      :reply,
+      do_add_job(scheduler_module, job),
+      %{
+        state
+        | schedulers:
+            schedulers
+            |> Map.put_new_lazy(scheduler_module, fn ->
+              create_scheduler_module_atom(scheduler_module)
+            end)
+      }
+    }
+  end
+
+  def handle_call({:jobs, scheduler_module}, _from, %__MODULE__{schedulers: schedulers} = state) do
+    {
+      :reply,
+      do_get_jobs(scheduler_module),
+      %{
+        state
+        | schedulers:
+            schedulers
+            |> Map.put_new_lazy(scheduler_module, fn ->
+              create_scheduler_module_atom(scheduler_module)
+            end)
+      }
+    }
+  end
+
+  def handle_call(
+        {:delete_job, scheduler_module, job},
+        _from,
+        %__MODULE__{schedulers: schedulers} = state
+      ) do
+    {
+      :reply,
+      do_delete_job(scheduler_module, job),
+      %{
+        state
+        | schedulers:
+            schedulers
+            |> Map.put_new_lazy(scheduler_module, fn ->
+              create_scheduler_module_atom(scheduler_module)
+            end)
+      }
+    }
+  end
+
+  def handle_call(
+        {:update_job_state, scheduler_module, job_name, job_state},
+        _from,
+        %__MODULE__{schedulers: schedulers} = state
+      ) do
+    {
+      :reply,
+      do_update_job_state(scheduler_module, job_name, job_state),
+      %{
+        state
+        | schedulers:
+            schedulers
+            |> Map.put_new_lazy(scheduler_module, fn ->
+              create_scheduler_module_atom(scheduler_module)
+            end)
+      }
+    }
+  end
+
+  def handle_call(
+        {:last_execution_date, scheduler_module},
+        _from,
+        %__MODULE__{schedulers: schedulers} = state
+      ) do
+    {
+      :reply,
+      do_get_last_execution_date(scheduler_module),
+      %{
+        state
+        | schedulers:
+            schedulers
+            |> Map.put_new_lazy(scheduler_module, fn ->
+              create_scheduler_module_atom(scheduler_module)
+            end)
+      }
+    }
+  end
+
+  def handle_call(
+        {:update_last_execution_date, scheduler_module, last_execution_date},
+        _from,
+        %__MODULE__{schedulers: schedulers} = state
+      ) do
+    {
+      :reply,
+      do_update_last_execution_date(scheduler_module, last_execution_date),
+      %{
+        state
+        | schedulers:
+            schedulers
+            |> Map.put_new_lazy(scheduler_module, fn ->
+              create_scheduler_module_atom(scheduler_module)
+            end)
+      }
+    }
+  end
+
+  def handle_call(
+        {:purge, scheduler_module},
+        _from,
+        %__MODULE__{schedulers: schedulers} = state
+      ) do
+    {
+      :reply,
+      do_purge(scheduler_module),
+      %{
+        state
+        | schedulers:
+            schedulers
+            |> Map.put_new_lazy(scheduler_module, fn ->
+              create_scheduler_module_atom(scheduler_module)
+            end)
+      }
+    }
+  end
+
+  # Helpers
+  defp create_scheduler_module_atom(scheduler_module) do
+    scheduler_module
+  end
+
+  defp job_key(job_name) do
+    {:job, job_name}
+  end
+
+  defp get_ets_by_scheduler(scheduler_module) do
+    scheduler_module_atom = create_scheduler_module_atom(scheduler_module)
+
+    if ets_exist?(scheduler_module_atom) do
+      scheduler_module_atom
+    else
+      PersistentEts.new(scheduler_module_atom, "#{scheduler_module_atom}.tab", [
+        :named_table,
+        :set
+      ])
+    end
+  end
+
+  defp ets_exist?(ets_name) do
+    Logger.debug(fn ->
+      "[#{inspect(Node.self())}][#{__MODULE__}] Determining whether ETS table with name [#{
+        inspect(ets_name)
+      }] exists"
+    end)
+
+    result =
+      case :ets.info(ets_name) do
+        :undefined -> false
+        _ -> true
+      end
+
+    Logger.debug(fn ->
+      "[#{inspect(Node.self())}][#{__MODULE__}] ETS table with name [#{inspect(ets_name)}] #{
+        if result, do: ~S|exists|, else: ~S|does not exist|
+      }"
+    end)
+
+    result
+  end
+
+  # Private functions
+  defp do_add_job(scheduler_module, job) do
+    table = get_ets_by_scheduler(scheduler_module)
+    :ets.insert(table, entry = {job_key(job.name), job})
+
+    Logger.debug(fn ->
+      "[#{inspect(Node.self())}][#{__MODULE__}] inserting [#{inspect(entry)}] into Persistent ETS table [#{
+        table
+      }]"
+    end)
+
+    :ok
+  end
+
+  defp do_get_jobs(scheduler_module) do
+    table = get_ets_by_scheduler(scheduler_module)
+
+    result =
+      case :ets.match(table, {{:job, :_}, :"$1"}) do
+        [] -> :not_applicable
+        [_h | _t] = jobs -> jobs |> List.flatten()
+      end
+
+    Logger.debug(fn ->
+      "[#{inspect(Node.self())}][#{__MODULE__}] jobs are: #{inspect(result)}"
+    end)
+
+    result
+  end
+
+  defp do_delete_job(scheduler_module, job_name) do
+    table = get_ets_by_scheduler(scheduler_module)
+    :ets.delete(table, job_key(job_name))
+    :ok
+  end
+
+  defp do_update_job_state(scheduler_module, job_name, state) do
+    table = get_ets_by_scheduler(scheduler_module)
+
+    job =
+      case :ets.lookup(table, {:job, job_name}) do
+        # TODO: should we raise here or should we handle the situation with a return value of a special kind?
+        [] ->
+          raise "Job #{job_name} does not exist in the storage"
+
+        [j | _t] ->
+          j
+      end
+
+    upd_job = %{job | state: state}
+    :ets.update_element(table, job_key(job_name), {1, upd_job})
+    :ok
+  end
+
+  defp do_get_last_execution_date(scheduler_module) do
+    table = get_ets_by_scheduler(scheduler_module)
+
+    case :ets.lookup(table, :last_execution_date) do
+      [] -> :unknown
+      [{:last_execution_date, date} | _t] -> date
+      {:last_execution_date, d} -> d
+    end
+  end
+
+  defp do_update_last_execution_date(scheduler_module, last_execution_date) do
+    table = get_ets_by_scheduler(scheduler_module)
+    :ets.insert(table, {:last_execution_date, last_execution_date})
+    :ok
+  end
+
+  defp do_purge(scheduler_module) do
+    table = get_ets_by_scheduler(scheduler_module)
+    :ets.delete_all_objects(table)
+    :ok
+  end
+
+  @behaviour Quantum.Storage.Adapter
+
+  def jobs(scheduler_module) do
+    __server__ |> GenServer.call({:jobs, scheduler_module})
+  end
+
+  def add_job(scheduler_module, job) do
+    __server__ |> GenServer.call({:add_job, scheduler_module, job})
+  end
+
+  def delete_job(scheduler_module, job_name) do
+    __server__ |> GenServer.call({:delete_job, scheduler_module, job_name})
+  end
+
+  def update_job_state(scheduler_module, job_name, state) do
+    __server__ |> GenServer.call({:update_job_state, scheduler_module, job_name, state})
+  end
+
+  def last_execution_date(scheduler_module) do
+    __server__ |> GenServer.call({:last_execution_date, scheduler_module})
+  end
+
+  def update_last_execution_date(scheduler_module, last_execution_date) do
+    __server__
+    |> GenServer.call({:update_last_execution_date, scheduler_module, last_execution_date})
+  end
+
+  def purge(scheduler_module) do
+    __server__ |> GenServer.call({:purge, scheduler_module})
+  end
 end

mix.exs

@@ -60,6 +60,8 @@ defmodule QuantumStorageEts.MixProject do
   # Run "mix help deps" to learn about dependencies.
   defp deps do
     [
+      {:persistent_ets, "~> 0.1.0"},
+      {:quantum, github: "c-rack/quantum-elixir", branch: "storage"},
       {:earmark, "~> 1.0", only: [:dev, :docs], runtime: false},
       {:ex_doc, "~> 0.13", only: [:dev, :docs], runtime: false},
       {:excoveralls, "~> 0.5", only: [:dev, :test], runtime: false},

mix.lock

@@ -2,18 +2,22 @@
   "bunt": {:hex, :bunt, "0.2.0", "951c6e801e8b1d2cbe58ebbd3e616a869061ddadcc4863d0a2182541acae9a38", [:mix], [], "hexpm"},
   "certifi": {:hex, :certifi, "2.0.0", "a0c0e475107135f76b8c1d5bc7efb33cd3815cb3cf3dea7aefdd174dabead064", [:rebar3], [], "hexpm"},
   "credo": {:hex, :credo, "0.8.10", "261862bb7363247762e1063713bb85df2bbd84af8d8610d1272cd9c1943bba63", [:mix], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}], "hexpm"},
+  "crontab": {:hex, :crontab, "1.1.2", "4784a50987b4a19af07a908f98e8a308b00f9c93efc5a7892155dc10cd8fc7d9", [:mix], [{:ecto, "~> 1.0 or ~> 2.0 or ~> 2.1", [hex: :ecto, repo: "hexpm", optional: true]}], "hexpm"},
   "dialyxir": {:hex, :dialyxir, "0.5.1", "b331b091720fd93e878137add264bac4f644e1ddae07a70bf7062c7862c4b952", [:mix], [], "hexpm"},
   "earmark": {:hex, :earmark, "1.2.4", "99b637c62a4d65a20a9fb674b8cffb8baa771c04605a80c911c4418c69b75439", [:mix], [], "hexpm"},
   "ex_doc": {:hex, :ex_doc, "0.18.3", "f4b0e4a2ec6f333dccf761838a4b253d75e11f714b85ae271c9ae361367897b7", [:mix], [{:earmark, "~> 1.1", [hex: :earmark, repo: "hexpm", optional: false]}], "hexpm"},
   "excoveralls": {:hex, :excoveralls, "0.8.1", "0bbf67f22c7dbf7503981d21a5eef5db8bbc3cb86e70d3798e8c802c74fa5e27", [:mix], [{:exjsx, ">= 3.0.0", [hex: :exjsx, repo: "hexpm", optional: false]}, {:hackney, ">= 0.12.0", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm"},
   "exjsx": {:hex, :exjsx, "4.0.0", "60548841e0212df401e38e63c0078ec57b33e7ea49b032c796ccad8cde794b5c", [:mix], [{:jsx, "~> 2.8.0", [hex: :jsx, repo: "hexpm", optional: false]}], "hexpm"},
+  "gen_stage": {:hex, :gen_stage, "0.13.1", "edff5bca9cab22c5d03a834062515e6a1aeeb7665fb44eddae086252e39c4378", [:mix], [], "hexpm"},
   "hackney": {:hex, :hackney, "1.11.0", "4951ee019df102492dabba66a09e305f61919a8a183a7860236c0fde586134b6", [:rebar3], [{:certifi, "2.0.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "5.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "1.0.1", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "1.0.2", [hex: :mimerl, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "1.1.1", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm"},
   "idna": {:hex, :idna, "5.1.0", "d72b4effeb324ad5da3cab1767cb16b17939004e789d8c0ad5b70f3cea20c89a", [:rebar3], [{:unicode_util_compat, "0.3.1", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm"},
   "inch_ex": {:hex, :inch_ex, "0.5.6", "418357418a553baa6d04eccd1b44171936817db61f4c0840112b420b8e378e67", [:mix], [{:poison, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},
   "jsx": {:hex, :jsx, "2.8.3", "a05252d381885240744d955fbe3cf810504eb2567164824e19303ea59eef62cf", [:mix, :rebar3], [], "hexpm"},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm"},
   "mimerl": {:hex, :mimerl, "1.0.2", "993f9b0e084083405ed8252b99460c4f0563e41729ab42d9074fd5e52439be88", [:rebar3], [], "hexpm"},
+  "persistent_ets": {:hex, :persistent_ets, "0.1.0", "a9ea1d6e41094441bb7fd3ea6a5717b66de82343bc6a2899e678a5bad46e495b", [:mix], [], "hexpm"},
   "poison": {:hex, :poison, "3.1.0", "d9eb636610e096f86f25d9a46f35a9facac35609a7591b3be3326e99a0484665", [:mix], [], "hexpm"},
+  "quantum": {:git, "https://github.com/c-rack/quantum-elixir.git", "d66854ffad75dcd8c0cbed978eee25fcf89e0ea6", [branch: "storage"]},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.1", "28a4d65b7f59893bc2c7de786dec1e1555bd742d336043fe644ae956c3497fbe", [:make, :rebar], [], "hexpm"},
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.3.1", "a1f612a7b512638634a603c8f401892afbf99b8ce93a45041f8aaca99cadb85e", [:rebar3], [], "hexpm"},
 }