yggdrasil

defmodule Yggdrasil do
  @moduledoc """
  > *Yggdrasil* is an immense mythical tree that connects the nine worlds in
  > Norse cosmology.

  `Yggdrasil` manages subscriptions to channels/queues in several brokers with
  the possibility to add more. Simple Redis, RabbitMQ and PostgreSQL adapters
  are implemented. Message passing is done through `Phoenix.PubSub` adapters.

  `Yggdrasil` also manages publishing pools to Redis, RabbitMQ and PostgreSQL
  using `:poolboy` library.

  This library provides three functions i.e:

    + To subscribe:
      ```elixir
      Yggdrasil.subscribe(Yggdrasil.Channel.t()) :: :ok
      ```
    + To unsubscribe:
      ```elixir
      Yggdrasil.unsubscribe(Yggdrasil.Channel.t()) :: :ok
      ```
    + To publish:
      ```elixir
      Yggdrasil.publish(Yggdrasil.Channel.t(), message()) :: :ok
        when message: term
      ```

  On subscription, the client receives a message with the following structure:

  ```elixir
  {:Y_CONNECTED, Yggdrasil.Channel.t()}
  ```

  and when the server publishes a message to the channel the message received
  by the clients has the following structure:

  ```elixir
  {:Y_EVENT, Yggdrasil.Channel.t(), message()} when message: term()
  ```

  The channels can be defined with the structure `Yggdrasil.Channel.t()`, e.g:

  ```elixir
  channel = %Yggdrasil.Channel{
    name: "redis_channel",
    transformer: Yggdrasil.Transformer.Default,
    adapter: :redis,
    namespace: TestRedis
  }
  ```

  where:

    + `name` - It's the name of the channel. For `Redis` adapter, the name of
    the channel is a string.
    + `transformer` - Module that contains the functions `decode/2` and
    `encode/2` to transform the messages. The default `transformer`
    `Yggdrasil.Default.Transformer` module does nothing to the messages.
    + `adapter` - It's the adapter to be used for subscriptions and/or
    publishing. In this case `:redis` is the adapter used both for
    subscriptions and publishing.
    + `namespace`: The namespace for the adapter configuration to allow several
    adapter configurations run at once depending on the needs of the system. By
    default is `Yggdrasil` e.g:
      ```elixir
      use Mix.Config

      config :yggdrasil,
        redis: [host: "localhost"]
      ```
    but for `TestRedis` namespace would be like this:
      ```elixir
      use Mix.Config

      config: :yggdrasil, TestRedis,
        redis: [host: "localhost"]
      ```

  # Small Example

  The following example uses the Elixir distribution to send the messages. It
  is equivalent to the Redis, RabbitMQ and Postgres distributions when a
  connection configuration is supplied:

  ```elixir
  iex(1)> Yggdrasil.subscribe(%Yggdrasil.Channel{name: "channel"})
  iex(2)> flush()
  {:Y_CONNECTED, %Yggdrasil.Channel{name: "channel", (...)}}
  ```

  and to publish a for the subscribers:

  ```elixir
  iex(3)> Yggdrasil.publish(%Yggdrasil.Channel{name: "channel"}, "message")
  iex(4)> flush()
  {:Y_EVENT, %Yggdrasil.Channel{name: "channel", (...)}, "message"}
  ```

  # Adapters

  The provided subscription adapters are the following (all are equivalents):

    * For `Elixir`: `Yggdrasil.Subscriber.Adapter.Elixir` and
    `Yggdrasil.Distributor.Adapter.Elixir`.
    * For `Redis`: `Yggdrasil.Subscriber.Adapter.Redis` and
    `Yggdrasil.Distributor.Adapter.Redis`.
    * For `RabbitMQ`: `Yggdrasil.Subscriber.Adapter.RabbitMQ` and
    `Yggdrasil.Distributor.Adapter.RabbitMQ`.
    * For `Postgres`: `Yggdrasil.Subscriber.Adapter.Postgres` and
    `Yggdrasil.Distributor.Adapter.Postgres`.

  The provided publisher adapters are the following:

    * For `Elixir`: `Yggdrasil.Publisher.Adapter.Elixir`.
    * For `Redis`: `Yggdrasil.Publisher.Adapter.Redis`.
    * For `RabbitMQ`: `Yggdrasil.Publisher.Adapter.RabbitMQ`.
    * For `Postgres`: `Yggdrasil.Publisher.Adapter.Postgres`.

  Also there is possible to use hibrid adapters. They work both subscriptions
  and publishing (recommended):

    * For `Elixir`: `:elixir` and `Yggdrasil.Elixir`.
    * For `Redis`: `:redis` and `Yggdrasil.Redis`.
    * For `RabbitMQ`: `:rabbitmq` and `Yggdrasil.RabbitMQ`.
    * For `Postgres`: `:postgres` and `Yggdrasil.Postgres`.

  The format for the channels for every adapter is the following:

    * For `Elixir` adapter: any valid term.
    * For `Redis` adapter: a string.
    * For `RabbitMQ` adapter: a tuple with exchange and routing key.
    * For `Postgres` adapter: a string.

  # Custom Transformers

  The only available transformer module is `Yggdrasil.Transformer.Default` and
  does nothing to the messages. It's possible to define custom transformers to
  _decode_ and _encode_ messages _from_ and _to_ the brokers respectively e.g:

  ```elixir
  defmodule QuoteTransformer do
    use Yggdrasil.Transformer

    alias Yggdrasil.Channel

    def decode(%Channel{} = _channel, message) do
      with {:ok, quoted} <- Code.string_to_quoted(message),
           {encoded, _} <- Code.eval_quoted(quoted),
           do: {:ok, encoded}
    end

    def encode(%Channel{} = _channel, data) do
      encoded = inspect(data)
      {:ok, encoded}
    end
  end
  ```

  and using the RabbitMQ adapter the channel would be:

  ```elixir
  iex(1)> channel = %Yggdrasil.Channel{
  ...(1)>   name: {"amq.topic", "quotes"},
  ...(1)>   adapter: :rabbitmq,
  ...(1)>   transformer: QuoteTransformer
  ...(1)> }
  ```

  where the `name` of the channel for this adapter is a tuple with the
  exchange name and the routing key. Then:

  ```elixir
  iex(2)> Yggdrasil.subscribe(channel)
  :ok
  iex(3)> flush()
  {:Y_CONNECTED, %Yggdrasil.Channel{} = _channel}
  iex(4)> Yggdrasil.publish(channel, %{"answer" => 42})
  :ok
  iex(5)> flush()
  {:Y_EVENT, %Yggdrasil.Channel{} = _channel, %{"answer" => 42}}
  ```

  # Configuration

  There are several configuration options and all of them should be in the
  general configuration for `Yggdrasil`:

    * `pubsub_adapter` - `Phoenix.PubSub` adapter. By default
    `Phoenix.PubSub.PG2` is used.
    * `pubsub_name` - Name of the `Phoenix.PubSub` adapter. By default is
    `Yggdrasil.PubSub`.
    * `pubsub_options` - Options of the `Phoenix.PubSub` adapter. By default
    are `[pool_size: 1]`.
    * `publisher_options` - `Poolboy` options for publishing. By default are
    `[size: 5, max_overflow: 10]`.
    * `registry` - Process name registry. By default is used `ExReg`.

  Also it can be specified the connection configurations with or without
  namespace:

    * `redis` - List of options of `Redix`:
      + `host` - Redis hostname.
      + `port` - Redis port.
      + `password` - Redis password.
    * `rabbitmq` - List of options of `AMQP`:
      + `host` - RabbitMQ hostname.
      + `port` - RabbitMQ port.
      + `username` - Username.
      + `password` - Password.
      + `virtual_host` - Virtual host.
      + `subscriber_options` - `poolboy` options for RabbitMQ subscriber. By
      default are `[size: 5, max_overflow: 10]`.
    * `postgres` - List of options of `Postgrex`:
      + `hostname` - Postgres hostname.
      + `port` - Postgres port.
      + `username` - Postgres username.
      + `password` - Postgres password.
      + `database` - Postgres database name.
  """
  use Application
  use VersionCheck, application: :yggdrasil

  alias Yggdrasil.Channel
  alias Yggdrasil.Distributor.Backend
  alias Yggdrasil.Publisher

  @publisher_gen Yggdrasil.Publisher.Generator
  @distributor_gen Yggdrasil.Distributor.Generator
  @rabbitmq_gen Yggdrasil.Subscriber.Adapter.RabbitMQ.Generator
  @broker Yggdrasil.Broker

  ######################
  # Subscriber functions

  @doc """
  Subscribes to a `channel`.
  """
  @spec subscribe(Channel.t()) :: :ok | {:error, term()}
  def subscribe(%Channel{} = channel) do
    channel = transform_channel(:client, channel)
    with :ok <- Backend.subscribe(channel),
         do: @broker.subscribe(@broker, channel, self())
  end

  @doc """
  Unsubscribes from a `channel`.
  """
  @spec unsubscribe(Channel.t()) :: :ok | {:error, term()}
  def unsubscribe(%Channel{} = channel) do
    channel = transform_channel(:client, channel)
    with :ok <- Backend.unsubscribe(channel),
         do: @broker.unsubscribe(@broker, channel, self())
  end

  #####################
  # Publisher functions

  @doc """
  Publishes a `message` in a `channel`.
  """
  @spec publish(Channel.t(), term()) :: :ok | {:error, term()}
  def publish(%Channel{} = channel, message) do
    channel = transform_channel(:server, channel)
    with {:ok, _} <- @publisher_gen.start_publisher(@publisher_gen, channel),
         do: Publisher.publish(channel, message)
  end

  #########
  # Helpers

  @doc false
  def transform_channel(:server, %Channel{adapter: :elixir} = channel) do
    %Channel{channel | adapter: Yggdrasil.Publisher.Adapter.Elixir}
  end
  def transform_channel(:server, %Channel{adapter: :redis} = channel) do
    %Channel{channel | adapter: Yggdrasil.Publisher.Adapter.Redis}
  end
  def transform_channel(:server, %Channel{adapter: :rabbitmq} = channel) do
    %Channel{channel | adapter: Yggdrasil.Publisher.Adapter.RabbitMQ}
  end
  def transform_channel(:server, %Channel{adapter: :postgres} = channel) do
    %Channel{channel | adapter: Yggdrasil.Publisher.Adapter.Postgres}
  end
  def transform_channel(:client, %Channel{adapter: :elixir} = channel) do
    %Channel{channel | adapter: Yggdrasil.Subscriber.Adapter.Elixir}
  end
  def transform_channel(:client, %Channel{adapter: :redis} = channel) do
    %Channel{channel | adapter: Yggdrasil.Subscriber.Adapter.Redis}
  end
  def transform_channel(:client, %Channel{adapter: :rabbitmq} = channel) do
    %Channel{channel | adapter: Yggdrasil.Subscriber.Adapter.RabbitMQ}
  end
  def transform_channel(:client, %Channel{adapter: :postgres} = channel) do
    %Channel{channel | adapter: Yggdrasil.Subscriber.Adapter.Postgres}
  end
  def transform_channel(type, %Channel{adapter: Yggdrasil.Elixir} = channel) do
    transform_channel(type, %Channel{channel | adapter: :elixir})
  end
  def transform_channel(type, %Channel{adapter: Yggdrasil.Redis} = channel) do
    transform_channel(type, %Channel{channel | adapter: :redis})
  end
  def transform_channel(type, %Channel{adapter: Yggdrasil.RabbitMQ} = channel) do
    transform_channel(type, %Channel{channel | adapter: :rabbitmq})
  end
  def transform_channel(type, %Channel{adapter: Yggdrasil.Postgres} = channel) do
    transform_channel(type, %Channel{channel | adapter: :postgres})
  end
  def transform_channel(
    :client,
    %Channel{adapter: Yggdrasil.Distributor.Adapter.Elixir} = channel
  ) do
    transform_channel(:client, %Channel{channel | adapter: :elixir})
  end
  def transform_channel(
    :client,
    %Channel{adapter: Yggdrasil.Distributor.Adapter.Redis} = channel
  ) do
    transform_channel(:client, %Channel{channel | adapter: :redis})
  end
  def transform_channel(
    :client,
    %Channel{adapter: Yggdrasil.Distributor.Adapter.RabbitMQ} = channel
  ) do
    transform_channel(:client, %Channel{channel | adapter: :rabbitmq})
  end
  def transform_channel(
    :client,
    %Channel{adapter: Yggdrasil.Distributor.Adapter.Postgres} = channel
  ) do
    transform_channel(:client, %Channel{channel | adapter: :postgres})
  end
  def transform_channel(_, %Channel{} = channel) do
    channel
  end

  ###################
  # Application start

  @adapter Application.get_env(:yggdrasil, :pubsub_adapter, Phoenix.PubSub.PG2)
  @name Application.get_env(:yggdrasil, :pubsub_name, Yggdrasil.PubSub)
  @options Application.get_env(:yggdrasil, :pubsub_options, [pool_size: 1])

  @doc false
  def start(_type, _args) do
    import Supervisor.Spec, warn: false

    check_version()

    monitors = :ets.new(:monitors, [:set, :public, write_concurrency: false,
                                    read_concurrency: true])

    children = [
      supervisor(@adapter, [@name, @options]),
      supervisor(@publisher_gen, [[name: @publisher_gen]]),
      supervisor(@distributor_gen, [[name: @distributor_gen]]),
      worker(@broker, [@distributor_gen, monitors, [name: @broker]]),
      supervisor(@rabbitmq_gen, [[name: @rabbitmq_gen]])
    ]

    opts = [strategy: :rest_for_one, name: Yggdrasil.Supervisor]
    Supervisor.start_link(children, opts)
  end
end