exkonsument

defmodule ExKonsument do
  @moduledoc false
  def open_connection(connection_string) do
    AMQP.Connection.open(connection_string)
  end

  def close_connection(connection) do
    AMQP.Connection.close(connection)
  end

  def connection_open?(connection) do
    Process.alive?(connection.pid)
  end

  def open_channel(connection) do
    case AMQP.Channel.open(connection) do
      {:ok, channel} ->
        AMQP.Basic.qos(channel, prefetch_count: 1)
        {:ok, channel}

      {:error, _} = error ->
        error
    end
  end

  def close_channel(channel) do
    AMQP.Channel.close(channel)
  end

  def declare_exchange(channel, exchange, type, opts \\ []) do
    AMQP.Exchange.declare(channel, exchange, type, opts)
  end

  def declare_queue(channel, queue \\ "", opts \\ []) do
    AMQP.Queue.declare(channel, queue, opts)
  end

  def publish(channel, exchange, routing_key, payload, options \\ []) do
    if Process.alive?(channel.pid) do
      AMQP.Basic.publish(channel, exchange, routing_key, payload, options)
    else
      :error
    end
  end

  def bind_queue(channel, queue, exchange, routing_keys) do
    Enum.each(routing_keys, fn key ->
      :ok = AMQP.Queue.bind(channel, queue, exchange, routing_key: key)
    end)

    :ok
  end

  def consume(channel, queue, consumer_pid \\ nil, opts \\ []) do
    AMQP.Basic.consume(channel, queue, consumer_pid, opts)
  end

  def ack(channel, delivery_tag, options \\ []) do
    AMQP.Basic.ack(channel, delivery_tag, options)
  end

  def reject(channel, delivery_tag, options \\ []) do
    AMQP.Basic.reject(channel, delivery_tag, options)
  end
end