Behaviour for the queue storage.

Every event received is dispatched to operators (EventStreamex.Operators.Operator), and executed sequentially to keep ordenancing.

Every time an event is received, a task is added to the queue of events with each operator listening for this event. And everytime a an operator task is finished, its completion status is updated in the task. When all operators have finished their task for this event, it is removed from the queue.

The adapters keeps a copy of this queue, mostly in case of crash (Except for EventStreamex.Operators.Queue.NoAdapter which does not save the queue and should not be used in production)

There are currently 2 queue adapters:

• EventStreamex.Operators.Queue.DbAdapter: A queue that uses the database tu store its items (This is the default queue adapter)
• EventStreamex.Operators.Queue.NoAdapter: A queue that does not store its items

You can create your own adapter and set the config to use it.

Here is the full code of a possible memory adapter:

defmodule EventStreamex.Operators.Queue.MemAdapter do
  use GenServer
  @behaviour EventStreamex.Operators.Queue.QueueStorageAdapter

  @impl true
  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl EventStreamex.Operators.Queue.QueueStorageAdapter
  def add_item(item) do
    GenServer.call(__MODULE__, {:save, item})
  end

  @impl EventStreamex.Operators.Queue.QueueStorageAdapter
  def delete_item(item) do
    GenServer.call(__MODULE__, {:delete, item})
  end

  @impl EventStreamex.Operators.Queue.QueueStorageAdapter
  def update_processors_status(item) do
    GenServer.call(__MODULE__, {:update, item})
  end

  @impl EventStreamex.Operators.Queue.QueueStorageAdapter
  def load_queue() do
    GenServer.call(__MODULE__, :load)
  end

  @impl EventStreamex.Operators.Queue.QueueStorageAdapter
  def reset_queue() do
    GenServer.call(__MODULE__, :reset)
  end

  # Callbacks

  @impl true
  def init(_opts) do
    {:ok, []}
  end

  @impl true
  def handle_call({:save, item}, _from, queue) do
    new_queue = queue ++ [item]
    {:reply, {:ok, new_queue}, new_queue}
  end

  @impl true
  def handle_call({:delete, {id, _item}}, _from, queue) do
    new_queue = queue |> Enum.reject(fn {i, _t} -> i == id end)
    {:reply, {:ok, new_queue}, new_queue}
  end

  @impl true
  def handle_call({:update, {id, item}}, _from, queue) do
    new_queue = queue |> Enum.map(fn {i, t} ->
      if i == id do
        item
      else
        t
      end
    end)
    {:reply, {:ok, new_queue}, new_queue}
  end

  @impl true
  def handle_call({_action, nil}, _from, queue) do
    {:reply, {:ok, queue}, queue}
  end

  @impl true
  def handle_call(:load, _from, queue) do
    {:reply, {:ok, queue}, queue}
  end

  @impl true
  def handle_call(:reset, _from, _queue) do
    {:reply, {:ok, []}, []}
  end
end

The adapter must be a GenServer and declare the function start_link/1.

Then, it must provide these functions:

• add_item/1: Add a new item to the queue
• delete_item/1: Delete an item from the queue
• update_processors_status/1: Updates the specified item
• load_queue/0: Return the entire queue
• reset_queue/0: Remove all items from the queue

An item is a tuple containing:

• The ID of the queue item
• The operator modules that process the event. It's a map with the module of the operator as key and a boolean as value. The boolean indicates the completion status for this operator
• The WalEx event (more on this structure in EventStreamex.EventListener)