Packages

event_bus

1.4.2
1.7.0 1.6.2 1.6.1 1.6.0 retired 1.5.2 retired 1.5.1 retired 1.5.0 retired 1.4.4 retired 1.4.3 retired 1.4.2 retired 1.4.1 retired 1.4.0 retired 1.3.8 retired 1.3.7 retired 1.3.6 retired 1.3.5 retired 1.3.4 retired 1.3.3 retired 1.3.2 retired 1.3.1 retired 1.3.0 retired 1.2.0 retired 1.1.3 retired 1.1.2 retired 1.1.1 retired 1.1.0 retired 1.0.0 retired 1.0.0-beta4 retired 1.0.0-beta3 retired 1.0.0-beta2 retired 1.0.0-beta1 retired 0.9.0 retired 0.8.0 retired 0.7.0 retired 0.6.1 retired 0.6.0 retired 0.5.0 retired 0.4.1 retired 0.4.0 retired 0.3.1 retired 0.3.0 retired 0.2.1 retired 0.2.0 retired 0.1.0 retired

Traceable, extendable and minimalist event bus implementation for Elixir with built-in event store and event watcher based on ETS

Retired package: Deprecated - Please prefer always latest minor
HexDocs

Current section

Files

Jump to
Readme
44 Versions
0 Dependencies
Dependants
Files
Activity
event_bus lib event_bus services observation.ex
Raw

lib/event_bus/services/observation.ex 

defmodule EventBus.Service.Observation do

  @moduledoc false



  alias EventBus.Manager.Store, as: StoreManager

  alias :ets, as: Ets



  @ets_opts [

    :set,

    :public,

    :named_table,

    {:write_concurrency, true},

    {:read_concurrency, true}

  ]

  @prefix "eb_ew_"



  @doc false

  @spec exist?(atom()) :: boolean()

  def exist?(topic) do

    table_name = table_name(topic)

    all_tables = Ets.all()

    Enum.any?(all_tables, fn table -> table == table_name end)

  end



  @doc false

  @spec register_topic(atom()) :: no_return()

  def register_topic(topic) do

    unless exist?(topic), do: Ets.new(table_name(topic), @ets_opts)

  end



  @doc false

  @spec unregister_topic(atom()) :: no_return()

  def unregister_topic(topic) do

    if exist?(topic), do: Ets.delete(table_name(topic))

  end



  @doc false

  @spec mark_as_completed(tuple()) :: no_return()

  def mark_as_completed({listener, event_shadow}) do

    {listeners, completers, skippers} = fetch(event_shadow)

    save_or_delete(event_shadow, {listeners, [listener | completers], skippers})

  end



  @doc false

  @spec mark_as_skipped(tuple()) :: no_return()

  def mark_as_skipped({listener, event_shadow}) do

    {listeners, completers, skippers} = fetch(event_shadow)

    save_or_delete(event_shadow, {listeners, completers, [listener | skippers]})

  end



  @doc false

  @spec fetch(tuple()) :: any()

  def fetch({topic, id}) do

    case Ets.lookup(table_name(topic), id) do

      [{_, data}] -> data

      _ -> nil

    end

  end



  @doc false

  @spec save(tuple(), tuple()) :: no_return()

  def save({topic, id}, watcher) do

    save_or_delete({topic, id}, watcher)

    :ok

  end



  @spec complete?(tuple()) :: boolean()

  defp complete?({listeners, completers, skippers}) do

    length(listeners) == length(completers) + length(skippers)

  end



  @spec save_or_delete(tuple(), tuple()) :: no_return()

  defp save_or_delete({topic, id}, watcher) do

    if complete?(watcher) do

      delete_with_relations({topic, id})

    else

      Ets.insert(table_name(topic), {id, watcher})

    end

  end



  @spec delete_with_relations(tuple()) :: no_return()

  defp delete_with_relations({topic, id}) do

    StoreManager.delete({topic, id})

    Ets.delete(table_name(topic), id)

  end



  @spec table_name(atom()) :: atom()

  defp table_name(name) do

    :"#{@prefix}#{name}"

  end

end