simple_rabbitmq_consumer

defmodule SimpleRabbitMQConsumer do
  use GenServer
  require Logger

  @reconnect_interval 10_000

  def start_link(opts) do
    Logger.debug("Started SimpleRabbitMQConsumer")
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(opts) do
    send(self(), :connect)
    state = Map.put(opts, :prev, nil)
    {:ok, state}
  end

  def handle_info(:connect, state) do
    case connect(state) do
      {:ok, conn} ->
        Process.monitor(conn.pid)
        Logger.info("SimpleRabbitMQConsumer connected")
        {:noreply, state}

      _ ->
        Logger.error("SimpleRabbitMQConsumer Failed to connect. Reconnecting later...")
        Process.send_after(self(), :connect, @reconnect_interval)
        {:noreply, state}
    end
  end

  def handle_info({:DOWN, _, :process, _pid, reason}, state) do
    {:stop, {:connection_lost, reason}, state}
  end

  def handle_info({:payload, payload}, state) do
    if state.prev != payload do
      fun = state.handle_payload
      fun.(payload)
    end

    state = Map.put(state, :prev, payload)
    {:noreply, state}
  end

  defp new_message(payload, _) do
    payload = Jason.decode!(payload)
    send(__MODULE__, {:payload, payload})
  end

  defp connect(state) do
    with {:ok, conn} <- AMQP.Connection.open(state.connection),
         {:ok, chan} <- AMQP.Channel.open(conn),
         {:ok, _} <- AMQP.Queue.subscribe(chan, state.topic, &new_message/2) do
      {:ok, conn}
    else
      err -> err
    end
  end
end