retort

defmodule Retort.Channel do
  @moduledoc """
  Establishes a channel
  """

  alias Retort.Connection

  require Logger

  @typedoc """
  The name of a RabbitMQ queue
  """
  @type queue :: String.t

  # Functions

  @doc """
  Closes the `channel` if it is not already closed
  """
  @spec ensure_closed(%AMQP.Channel{}) :: :already_closed | :ok
  def ensure_closed(channel = %AMQP.Channel{}) do
    AMQP.Channel.close(channel)
  catch
    :exit, {:noproc, _} -> :already_closed
  else
    :ok -> :ok
  end

  @doc """
  Opens a new channel on `queue` with the calling process as the consumer
  """
  @spec init(queue) :: {:ok, %AMQP.Channel{}}
  @spec init(queue, module) :: {:ok, %AMQP.Channel{}}
  def init(queue, amqp \\ Retort.AMQP.Default) when is_binary(queue) and is_atom(amqp) do
    retryable_init(queue, amqp, 1)
  end

  ## Private Functions

  @spec maybe_init(%AMQP.Connection{}, queue, module) :: {:ok, %AMQP.Channel{}} | :error
  defp maybe_init(connection = %AMQP.Connection{}, queue, amqp) when is_binary(queue) and is_atom(amqp) do
    with {:ok, channel} <- maybe_open(connection, amqp) do
      # Limit unacknowledged messages to 1 to allow round-robinning by the broker
      amqp.prefetch_count(channel, 1)
      {:ok, %{queue: ^queue}} = amqp.declare_queue(channel, queue)
      {:ok, _consumer_tag} = amqp.consume(channel, queue)

      channel
    end
  catch
    # Connection or Channel died while opening and consuming
    :exit, reason ->
      Logger.error fn ->
        ["Exit (", inspect(reason), ") while initializing channel"]
      end

      :error
  else
    channel = %AMQP.Channel{} ->
      _reference = monitor(channel)

      {:ok, channel}
    :error ->
      :error
  end

  @spec maybe_open(%AMQP.Connection{}, module) :: {:ok, %AMQP.Channel{}} | :error
  defp maybe_open(connection = %AMQP.Connection{}, amqp) when is_atom(amqp) do
    case amqp.open(connection) do
      success = {:ok, _channel} -> success
      :closing ->
        Logger.error("Connection is closing while initializng channel")

        :error
    end
  end

  @spec monitor(%AMQP.Channel{}) :: reference
  defp monitor(%AMQP.Channel{pid: pid}), do: Process.monitor(pid)

  # Keeps retrying to init channel until it succeeds
  @spec retryable_init(queue, module, non_neg_integer) :: {:ok, %AMQP.Channel{}}
  defp retryable_init(queue, amqp, retry_count)
       when is_binary(queue) and is_atom(amqp) and is_integer(retry_count) and retry_count > 0 do
    {:ok, connection} = Connection.await()

    retryable_init(connection, queue, amqp, retry_count)
  end

  @spec retryable_init(%AMQP.Connection{}, queue, module, non_neg_integer) :: {:ok, %AMQP.Channel{}}
  defp retryable_init(connection = %AMQP.Connection{}, queue, amqp, retry_count)
       when is_binary(queue) and is_atom(amqp) and is_integer(retry_count) and retry_count > 0 do
    case maybe_init(connection, queue, amqp) do
      success = {:ok, %AMQP.Channel{}} ->
        success
      :error ->
        Logger.info fn ->
          ["Retry (", to_string(retry_count), ") to initialize channel"]
        end

        retryable_init(queue, amqp, retry_count + 1)
    end
  end
end