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
  [`YProcess`](https://github.com/gmtprime/y_process). `YProcess` allows to use
  `Phoenix.PubPub` as a pub/sub to distribute messages between processes.

  ## Example using Redis

  ```elixir
  iex(1)> channel = %Yggdrasil.Channel{channel: "redis_channel", decoder: Yggdrasil.Decoder.Default.Redis}
  iex(2)> Yggdrasil.subscribe(channel)
  ```

  > By default, the Redis adapter connects to `"redis://localhost:6379"`.

  Then in redis:

  ```
  127.0.0.1:6379> PUBLISH "redis_channel" "hello"
  (integer) (1)
  ```

  And finally if you flush in your `iex` you'll see the message received
  by the Elixir shell:

  ```elixir
  iex(3> flush()
  {:Y_CAST_EVENT, "redis_channel", "hello"}
  :ok
  ```

  Things to note:

    * Every message coming from a broker (Redis, RabbitMQ, PostgreSQL) will be
    like:

    ```elixir
    {:Y_CAST_EVENT, channel, message}
    ```

    * The process calling `Yggdrasil.subscribe/1` will be the one that subscribes
    to the channel

  ## Example using RabbitMQ

  First you must have the RabbitMQ exchange created. Otherwise the client won't
  connect. Channels for the RabbitMQ adapter use a tupple instead of a string:

  ```elixir
  {"exchange_name", "routing_key"}
  ```

  where the exchange should be of type `:topic`. This would allow you to connect
  to a channel using a routing key like `"*.error"` where messages with routing
  keys like `"miami.error"` and `"barcelona.error"` would match the routing key
  of the `Yggdrasil` channel.

  Let's say you want to connect to the exchange `"amq.topic"` (created by default)
  with the previous routing key (`"*.error"`) where you'll receive errors from all
  the servers:

  ```elixir
  iex(1)> channel = %Yggdrasil.Channel{channel: {"amq.topic", "*.error"}, decoder: Yggdrasil.Decoder.Default.RabbitMQ}
  iex(2)> Yggdrasil.subscribe(channel)
  ```

  > By default, the RabbitMQ adapter connects to
  > `"amqp://guest:guest@localhost:5672/"`

  Then using `AMQP` library, publish some messages in RabbitMQ:

  ```elixir
  iex(3)> options = Application.get_env(:yggdrasil, :rabbitmq, [])
  iex(4)> {:ok, conn} = AMQP.Connection.open(options)
  iex(5)> {:ok, chan} = AMQP.Channel.open(conn)
  iex(6)> AMQP.Basic.publish(chan, "amq.topic", "miami.error", "Error from Miami")
  iex(7)> AMQP.Basic.publish(chan, "amq.topic", "barcelona.error", "Error from Barcelona")
  ```

  And finally if you flush in your `iex` you'll see the message received
  by the Elixir shell:

  ```elixir
  iex(8> flush()
  {:Y_CAST_EVENT, {"amq.topic", "*.error"}, "Error from Miami"}
  {:Y_CAST_EVENT, {"amq.topic", "*.error"}, "Error from Barcelona"}
  :ok
  ```

  ## Example using PostgreSQL

  For this example, it's necessary to provide a valid configuration for the
  PostgreSQL adapter i.e:

  ```elixir
  use Mix.Config

  config :yggdrasil,
    postgres: [hostname: "localhost",
              port: 5432,
              username: "yggdrasil_test",
              password: "yggdrasil_test",
              database: "yggdrasil_test"]
  ```

  This will connect the adapter to the database `yggdrasil_test` with the user
  `yggdrasil_test` and the password `yggdrasil_test` on `localhost:5432`.

  ```elixir
  iex(1)> channel = %Yggdrasil.Channel{channel: "postgres_channel", decoder: Yggdrasil.Decoder.Default.Postgres}
  iex(2)> Yggdrasil.subscribe(channel)
  ```

  Then in PostgreSQL:

  ```
  yggdrasil_test=> NOTIFY postgres_channel, 'hello'
  NOTIFY
  ```

  And finally if you flush in `iex` you'll see the message received by the Elixir
  shell:

  ```elixir
  iex(8> flush()
  {:Y_CAST_EVENT, "postgres_channel", "hello"}
  :ok
  ```

  ## Example using GenServer

  Any of the previous examples can be wrapped inside a `GenServer`, in this case
  it is Redis:

  ```elixir
  defmodule Subscriber do
    use GenServer

    ###################
    # Client functions.

    def start_link(channel, opts \\ []) do
      GenServer.start_link(__MODULE__, channel, opts)
    end

    def stop(subscriber, reason \\ :normal) do
      GenServer.stop(subscriber, reason)
    end

    ######################
    # GenServer callbacks.

    def init(channel) do
      Yggdrasil.subscribe(channel)
      {:ok, channel}
    end

    def handle_info({:Y_CAST_EVENT, channel, message}, state) do
      IO.inspect %{channel: channel, message: message}
      {:noreply, state}
    end

    def terminate(_reason, channel) do
      Yggdrasil.unsubscribe(channel)
      :ok
    end
  end
  ```

  So in `iex`:

  ```elixir
  iex(1)> channel = %Yggdrasil.Channel{channel: "redis_channel", decoder: Yggdrasil.Decoder.Default.Redis}
  iex(2)> {:ok, subscriber} = Subscriber.start_link(channel)
  iex(3)>
  ```

  Again in Redis:

  ```
  127.0.0.1:6379> PUBLISH "redis_channel" "hello"
  (integer) (1)
  ```

  And finally you'll see in your `iex` the following:

  ```elixir
  %{channel: "redis_channel", message: "hello"}
  iex(3)>
  ```

  ## Example using YProcess

  `YProcess` is a `GenServer` wrapper with pubsub capabilities and it has
  great sinergy with `Yggdrasil`. The previous example implemented with
  `YProcess` would be:

  ```elixir
  defmodule YSubscriber do
    use YProcess, backend: Yggdrasil.Backend

    ###################
    # Client functions.

    def start_link(channel, opts \\ []) do
      YProcess.start_link(__MODULE__, channel, opts)
    end

    def stop(subscriber, reason \\ :normal) do
      YProcess.stop(subscriber, reason)
    end

    #####################
    # YProcess callbacks.

    def init(channel) do
      {:join, [channel], channel}
    end

    def handle_event(channel, message, state) do
      IO.inspect %{channel: channel, message: message}
      {:noreply, state}
    end
  end
  ```

  So in `iex`:

  ```elixir
  iex(1)> channel = %Yggdrasil.Channel{channel: "redis_channel", decoder: Yggdrasil.Decoder.Default.Redis}
  iex(2)> {:ok, y_subscriber} = YSubscriber.start_link(channel)
  iex(3)>
  ```

  Again in Redis:

  ```
  127.0.0.1:6379> PUBLISH "redis_channel" "hello"
  (integer) (1)
  ```

  And finally you'll see in your `iex` the following:

  ```elixir
  %{channel: "redis_channel", message: "hello"}
  iex(3)>
  ```

  ## Yggdrasil Channels

  `Yggdrasil` channels have the name of the channel in the broker and the
  name of the module of the message decoder. A decoder module also defines
  which adapter should be used to connect to the channel.

  ```elixir
  %Yggdrasil.Channel{channel: "channel", decoder: Yggdrasil.Decoder.Default.Redis}
  ```
  The previous example will tell Yggdrasil to subscribe to the channel `"channel"`.
  The decoder module `Yggdrasil.Decoder.Default.Redis` defined Redis as the broker
  and does not change the message coming from Redis before sending it to the
  subscribers.

  ### Decoders

  The current `Yggdrasil` version has the following decoder modules:

    * `Yggdrasil.Decoder.Default`: Does nothing to the message and uses the
    `Yggdrasil.Adapter.Elixir`.
    * `Yggdrasil.Decoder.Default.Redis`: Does nothing to the message and uses the
    `Yggdrasil.Adapter.Redis`.
    * `Yggdrasil.Decoder.Default.RebbitMQ`: Does nothing to the message and uses the
    `Yggdrasil.Adapter.RabbitMQ`.
    * `Yggdrasil.Decoder.Default.Postgres`: Does nothing to the message and uses the
    `Yggdrasil.Adapter.Postgres`.

  > For more information about adapters, see the Adapters section.

  To implement a decoder is necessary to implement the `decode/2` callback for
  `Yggdrasil.Decoder` behaviour, i.e. subscribe to a Redis channel `"test"` that
  publishes JSON. The subscribers must receive a map instead of a string with the
  JSON.

  ```elixir
  defmodule CustomDecoder do
    use Yggdrasil.Decoder, adapter: Yggdrasil.Adapter.Redis

    def decode(_channel, message) do
      Poison.decode!(message)
    end
  end
  ```

  > **Important**: The channel received by the `decode/2` function might be
  > different than the channel the client is subscribed. For example, with the
  > RabbitMQ adapter you can subscribe to the channel `{"amq.topic", "*.error"}`,
  > but if the routing key of the received message is `"barcelona.error"`, then
  > the channel received by this function will be `{"amq.topic", "barcelona.error"}`
  > instead of `{"amq.topic", "*.error"}`. It is a good idea to include this
  > channel to the decoded message in order to know its real procedence.

  To subscribe to this channel, clients must use the following `Yggdrasil` channel:

  ```elixir
  %Yggdrasil.Channel{channel: "test", decoder: CustomDecoder}
  ```

  ### Adapters

  The current `Yggdrasil` version has the following adapters:

    * `Yggdrasil.Adapter.Elixir`: Message distribution using Elixir messages.
    * `Yggdrasil.Adapter.Redis`: Messages come from a Redis channel.
    * `Yggdrasil.Adapter.RabbitMQ`: Messages come from a RabbitMQ queue. A
    `channel` is a tuple that contains the exchange and the routing key:
    `{exchange, routing_key}`.
    * `Yggdrasil.Adapter.Postgres`: Messages come from the notifies of a
    PostgreSQL database.

  Also the function `Yggdrasil.publish/2` is used to simulate published messages
  by any of the brokers.

  To implement a new adapter is necessary to use a `GenServer` or any wrapper over
  `GenServer`. For more information, see the source code of any of the implemented
  adapters.

  ## Installation

  `Yggdrasil` is available as a Hex package. To install, add it to your
  dependencies in your `mix.exs` file:

  ```elixir
  def deps do
    [{:amqp_client, git: "https://github.com/jbrisbin/amqp_client.git", override: true},
    {:yggdrasil, "~> 2.0.6"}]
  end
  ```

  > Overriding `:amqp_client` dependency is necessary in order to use `Yggdrasil`
  > with Erlang 19.

  and ensure `Yggdrasil` is started before your application:

  ```elixir
  def application do
    [applications: [:yggdrasil]]
  end
  ```

  ## Configuration

  `Yggdrasil` uses `YProcess` as a means to distribute the messages. So it is
  necessary to provide a configuration for `YProcess` if you want to use, for
  example, `Phoenix.PubSub` as your pubsub, i.e:

  ```elixir
  use Mix.Config

  config :y_process,
    backend: YProcess.Backend.PhoenixPubSub,
    name: Yggdrasil.PubSub,
    adapter: Phoenix.PubSub.PG2,
    options: [pool_size: 10]
  ```
  by default, `YProcess` will use `YProcess.Backend.PG2` as default backend.

  For `Yggdrasil`, there's only one general configuration parameter which is
  the process name registry. By default, it uses `ExReg`, which is a simple but
  rich process name registry. It is possible to use another one like `:gproc`.

  ```elixir
  use Mix.Config

  config :yggdrasil,
    registry: :gproc
  ```

  Specific configuration parameters are as follows:

    * To configure `Yggdrasil` with the provided Redis adapter
    (`Yggdrasil.Adapter.Redis`):

    ```elixir
    use Mix.Config

    config :yggdrasil,
      redis: [host: "localhost",
              port: 6379,
              password: "my password"]
    ```
    The default Redis adapter uses `Redix`, so the configuration parameters have
    the same name as the ones in `Redix`. By default connects to `redis://localhost`.

    * To configure `Yggdrasil` with the provided RabbitMQ adapter
    (`Yggdrasil.Adapter.RabbitMQ`):

    ```elixir
    use Mix.Config

    config :yggdrasil,
      rabbitmq: [host: "localhost",
                port: 5672,
                username: "guest",
                password: "guest",
                virtual_host: "/"]
    ```
    The default RabbitMQ adapter uses `AMQP`, so the configuration parameters have
    the same name as the ones in `AMQP`. By default connects to
    `amqp://guest:guest@localhost/`

    * To configure `Yggdrasil` with the provided PostgreSQL adapter
    (`Yggdrasil.Adapter.Postgres`):

    ```elixir
    use Mix.Config

    config :yggdrasil,
      postgres: [hostname: "localhost",
                port: 5432,
                username: "postgres",
                password: "postgres",
                database: "yggdrasil"]
    ```
    The default PostgreSQL adapter uses `Postgrex`, so the configuration parameters
    have the same name as the ones in `Postgrex`.
  """
  use Application

  alias Yggdrasil.Channel

  @backend YProcess.Backend.PG2
  @generator Yggdrasil.Publisher.Generator
  @broker Yggdrasil.Broker

  #############
  # Client API.

  @doc """
  Subscribes to a `channel`.
  """
  def subscribe(channel) do
    Yggdrasil.Backend.join(channel, self())
  end

  @doc """
  Unsubscribe from a `channel`.
  """
  def unsubscribe(channel) do
    Yggdrasil.Backend.leave(channel, self())
  end

  @doc """
  Emits a `message` in a `channel`. Bypasses the adapter.
  """
  def publish(%Channel{channel: channel, decoder: decoder}, message) do
    registry = apply(@generator, :get_registry, [])
    publisher = {:via, registry, {Yggdrasil.Publisher, channel, decoder}}
    Yggdrasil.Publisher.sync_notify(publisher, channel, message)
  end
  def publish(channel, message) do
    Yggdrasil.Backend.emit(channel, message)
  end

  #################
  # Version checks.

  require Logger
  @version Mix.Project.config[:version]

  ##
  # Current Yggdrasil version.
  defp current_version, do: @version

  ##
  # Checks version.
  defp check_version() do
    Hex.start()
    Hex.Utils.ensure_registry!()

    all_versions =
      :yggdrasil
      |> Atom.to_string()
      |> Hex.Registry.get_versions()
    current = current_version()

    if should_update?(all_versions, current) do
      latest = latest_version(all_versions, current)
      Logger.warn("A new Yggdrasil version is available (#{latest} > #{current}).")
    else
      Logger.debug("Using the lastest version of Yggdrasil (#{current}).")
    end
  end

  ##
  # Whether Yggdrasil should be updated or not.
  defp should_update?(all_versions, current) do
    latest = latest_version(all_versions, current)
    Hex.Version.compare(current, latest) == :lt
  end

  ##
  # Gets the latest version.
  defp latest_version(all_versions, default) do
    including_pre_versions? = pre_version?(default)
    latest = highest_version(all_versions, including_pre_versions?)
    latest || default
  end

  ##
  # Whether it allows previous versions or not.
  defp pre_version?(version) do
    {:ok, version} = Hex.Version.parse(version)
    version.pre != []
  end

  ##
  # Gets the highest version.
  defp highest_version(versions, including_pre_versions?) do
    if including_pre_versions? do
      versions |> List.last
    else
      versions |> Enum.reject(&pre_version?/1) |> List.last
    end
  end

  ##############
  # Application.

  ##
  # Monitors table.
  defp get_monitors_table do
    :ets.new(:monitors, [:set, :public, write_concurrency: false,
                         read_concurrency: true])
  end

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

    check_version()

    monitors = get_monitors_table()

    children = [
      supervisor(@generator, [[name: @generator]]),
      worker(@broker, [@generator, monitors, [name: @broker]])
    ] 

    children = case Application.get_env(:y_process, :backend, @backend) do
      YProcess.Backend.PG2 -> children
      YProcess.Backend.PhoenixPubSub ->
        [supervisor(YProcess.PhoenixPubSub, []) | children]
    end

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