Packages

yggdrasil

3.0.2
6.0.3 6.0.2 6.0.1 6.0.0 5.0.2 5.0.1 5.0.0 4.1.6 4.1.5 4.1.4 4.1.3 4.1.2 4.1.1 4.1.0 4.0.0 3.3.4 3.3.3 3.3.2 3.3.1 3.3.0 3.2.3 3.2.2 3.2.1 3.2.0 3.1.0 3.0.3 3.0.2 3.0.1 3.0.0 2.1.0 2.0.8 2.0.7 2.0.6 2.0.5 2.0.4 2.0.3 2.0.2 2.0.1 2.0.0 1.2.3 1.2.2 1.2.1 1.1.1 1.1.0 1.0.0

Agnostic pub/sub with Redis, RabbitMQ and Postgres support

HexDocs

Current section

Files

Jump to
Readme
45 Versions
8 Dependencies
Dependants
Files
Activity
yggdrasil lib yggdrasil distributor publisher.ex
Raw

lib/yggdrasil/distributor/publisher.ex 

defmodule Yggdrasil.Distributor.Publisher do
  @moduledoc """
  A server to distribute the messages.
  """
  use GenServer
  alias Yggdrasil.Channel
  alias Yggdrasil.Distributor.Backend
  require Logger
  #############################################################################
  # Client API.
  @doc """
  Starts a server to distribute messages in a `channel`. Additionally can
  receive `GenServer` `options`.
  """
  def start_link(%Channel{} = channel, options \\ []) do
    GenServer.start_link(__MODULE__, channel, options)
  end
  @doc """
  Stops a `publisher`.
  """
  def stop(publisher) do
    GenServer.stop(publisher)
  end
  @doc """
  Notifies synchronously of a new `message` coming from a `channel_name` to a
  `publisher`.
  """
  def notify(publisher, channel_name, message) do
    GenServer.call(publisher, {:notify, channel_name, message})
  end
  #############################################################################
  # GenServer callbacks.
  @doc false
  def init(%Channel{} = channel) do
    Logger.debug("Started publisher for #{inspect channel}")
    {:ok, channel}
  end
  @doc false
  def handle_call(
    {:notify, channel_name, message},
    _from,
    %Channel{transformer: transformer_module} = channel
  ) do
    real_channel = %Channel{channel | name: channel_name}
    result =
      with {:ok, decoded} <- transformer_module.decode(real_channel, message),
           do: Backend.publish(channel, decoded)
    {:reply, result, channel}
  end
  def handle_call(_msg, _from, %Channel{} = channel) do
    {:noreply, channel}
  end
end