Packages

yggdrasil

1.2.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
3 Dependencies
7 Dependants
Files
Activity
yggdrasil lib yggdrasil broker postgres.ex
Raw

lib/yggdrasil/broker/postgres.ex 

defmodule Yggdrasil.Broker.Postgres do

  use Yggdrasil.Broker



  def subscribe({repo, channel}, _callback) when is_atom(repo) do

    config = Application.get_env(:postgrex, repo)

    _subscribe(channel, config)

  end



  def subscribe(channel, _callback) when is_binary(channel) do

    config = Application.get_env(:postgrex, Yggdrasil.Repo)

    _subscribe(channel, config)

  end



  defp _subscribe(channel, config) do

    {:ok, conn} = Postgrex.Notifications.start_link(config)

    {:ok, ref} = Postgrex.Notifications.listen(conn, channel)

    _notify(channel, ref, config)

    {:ok, %{connection: conn, ref: ref}}

  end



  defp _notify(channel, message, config) when is_binary(message) do

    {:ok, conn} = Postgrex.start_link(config)

    Postgrex.query(conn, "NOTIFY #{channel}, '#{message}'", [])

    GenServer.stop(conn)

  end



  defp _notify(channel, message, config), do:

    _notify(channel, inspect(message), config)



  def unsubscribe(%{connection: conn, ref: ref}, _channel) do

    Postgrex.Notifications.unlisten(conn, ref)

    GenServer.stop(conn)

  end



  def handle_message(_, _, {:notification, _, ref, _, message}), do:

    _handle_message(message, inspect(ref))

  def handle_message(_, _, _), do:

    :whatever



  defp _handle_message(ref, ref), do:

    :subscribed

  defp _handle_message(message, _), do:

    {:message, message}

end