defmodule Enki do @moduledoc """ Enki is a simple queue that provides Mnesia persistence across nodes and `ttf` (time-to-flight) capability. Time-to-flight means that, when dequeuing a message, if the dequeue is not ack'd within a given period of time, the message is automatically added back to the queue. This ensures that no messages are lost. ## Examples Enki.enq(%{a: 1, b: 2}) %{id: id, payload: %{a: 1, b: 2}} = Enki.deq() :ok = Enki.ack(id) """ @moduledoc since: "0.1.0" alias Enki.Message defmodule Counter do @moduledoc false use Agent def start_link(_opts) do Agent.start_link(fn -> 0 end, name: __MODULE__) end def next_value() do Agent.get(__MODULE__, &(&1 + 1)) end end @sup Enki.SupervisedClients @doc """ Enki application start method. Gets called automatically when included in your `mix` `applications` list. """ def start(_type, _args) do import Supervisor.Spec, warn: false if file_persist() do Memento.stop() Memento.Schema.create(nodes()) Memento.start() maybe_create_table(disc_copies: nodes()) else maybe_create_table() end children = [ worker(Counter, [[]]), {DynamicSupervisor, strategy: :one_for_one, name: @sup} ] Supervisor.start_link(children, strategy: :one_for_one, name: Enki.Supervisor, max_restarts: 10_000 ) end @doc """ Adds a message to the queue. Returns a Message instance containing the `id` of the message on the queue and the message itself as a `payload`. """ @spec enq(any()) :: Message.t() def enq(message) do Memento.transaction!(fn -> id = Counter.next_value() Memento.Query.write(%Message{id: "#{id}_#{UUID.uuid4(:hex)}", payload: message}) end) end @doc """ Dequeues a message from the queue. Returns the message in a `Message` struct as its `payload` parameter. The message is typically the oldest in the queue. ## Parameters | name | description | | ---- | ----------- | | `ttf` | The time-to-flight for the message. If provided, verrides the message in the config (optional). | """ def deq(ttf \\ nil) do with %Enki.Message{id: id} = message <- deq_(), {:ok, _pid} <- DynamicSupervisor.start_child( @sup, Supervisor.child_spec( {Enki.InFlight, message: message, ttf: ttf || time_to_flight(), id: child_name(id)}, id: child_name(id), restart: :transient ) ) do message end end @doc """ Acknowledges a dequeued message. If the message is in-flight, it will not be re-added to the queue after the alotted `ttf`. ## Parameters | name | description | | ---- | ----------- | | `id` | The `id` of the message to acknowledge (required). | """ def ack(id) do child_exit(id) end @doc """ Retrieves a message by `id` without dequeuing. Recalling a message directly does NOT put it in flight. ## Parameters | name | description | | ---- | ----------- | | `id` | The `id` of the message to retrieve (required). | """ def get(id) do Memento.transaction!(fn -> Memento.Query.read(Message, id) end) end @doc """ Deletes a message by `id`. Directly deletes a message in the queue. ## Parameters | name | description | | ---- | ----------- | | `id` | The `id` of the message to delete (required). | """ def delete(id) do Memento.transaction!(fn -> case get(id) do %Message{id: id} -> child_exit(id) Memento.Query.delete(Message, id) _ -> :ok end end) end @doc """ Delete all messages in the queue. Any in-flight messages are cancelled, so messages are not added back to the queue. """ def delete_all() do Memento.transaction!(fn -> Memento.Query.all(Message) |> Enum.each(fn rec -> child_exit(rec.id) Memento.Query.delete_record(rec) end) end) end @doc false def child_exists?(id), do: child_name(id) |> Process.whereis() |> is_alive?() @doc false def monitor(id), do: child_name(id) |> Process.whereis() |> monitor_() defp maybe_create_table(opts \\ []) do try do Memento.Table.info(Message) catch :exit, _ -> Memento.Table.create!(Message, opts) end end defp deq_() do Memento.transaction!(fn -> with [%Message{} = msg] <- Memento.Query.select(Message, [], limit: 1), _ <- Memento.Query.delete_record(msg) do msg else [] -> nil end end) end defp child_exit(id) do if child_name(id) |> Process.whereis() |> is_alive?() do GenServer.stop(child_name(id)) end :ok end defp child_name(id), do: "enki_#{inspect(id)}" |> String.to_atom() defp is_alive?(pid) when is_pid(pid), do: Process.alive?(pid) defp is_alive?(_), do: false defp monitor_(pid) when is_pid(pid), do: {Process.monitor(pid), pid} defp monitor_(_), do: nil defp nodes(), do: [node() | Node.list()] defp time_to_flight(), do: Application.get_env(:enki, :ttf, 5000) defp file_persist(), do: Application.get_env(:enki, :file_persist, false) end