gramnad_rpc

defmodule Arland.RabbitMQ.Gramnad do
  alias :crypto, as: Crypto
  alias :ets, as: ETS
  require Logger
  use GenServer
  use AMQP

  @gramnad_main_queue "gramnad_queue"

  def start_link(init_args) do
    GenServer.start_link(__MODULE__, init_args, name: __MODULE__)
  end

  def init(args) do
    Logger.info("Starting RabbitMQ connection client")

    conn_link =
      "amqp://#{args.rabbitmq_username}:#{args.rabbitmq_password}@#{args.rabbitmq_url}:#{args.rabbitmq_port}"

    {:ok, conn} = Connection.open(conn_link)
    {:ok, chan} = Channel.open(conn)
    {:ok, queue} = setup_queue(chan)
    :ok = Basic.qos(chan, prefetch_count: 5)
    {:ok, _consumer_tag} = Basic.consume(chan, queue, self(), no_ack: false)

    ets_ref = ETS.new(:gramnad_rpc_pids, [:set])

    {:ok, %{chan: chan, queue: queue, ets_ref: ets_ref}}
  end

  def handle_info({:basic_consume_ok, _data}, state) do
    Logger.info("Consumer process registered")
    {:noreply, state}
  end

  def handle_info({:basic_cancel, _data}, state) do
    Logger.warning("Consumer process cancel")
    {:stop, :normal, state}
  end

  def handle_info({:basic_cancel_ok, _data}, state) do
    Logger.warning("Consumer process cancel ok")
    {:noreply, state}
  end

  def handle_info({:basic_deliver, payload, attributes}, state) do
    pid_from_corr_id =
      gen_pid_from_corr_id(
        state.ets_ref,
        attributes.correlation_id
      )

    GenServer.reply(pid_from_corr_id, payload)
    unregister_corr_id(state.ets_ref, attributes.correlation_id)
    Basic.ack(state.chan, attributes.delivery_tag)

    {:noreply, state}
  end

  def handle_call({:rpc, data}, from, state) do
    correlation_id = generate_correlation_id()

    Basic.publish(
      state.chan,
      "",
      @gramnad_main_queue,
      data |> Jason.encode!(),
      correlation_id: correlation_id,
      reply_to: state.queue
    )

    register_corr_id(state.ets_ref, correlation_id, from)

    {:noreply, state}
  end

  def handle_call(msg, _from, state) do
    Logger.warn("Got unknown message: #{inspect(msg)}")
    {:reply, {:error, "Method undefined"}, state}
  end

  def handle_cast(msg, state) do
    Logger.warn("Got unknown message: #{inspect(msg)}")
    {:noreply, state}
  end

  defp register_corr_id(ets_ref, correlation_id, pid) do
    ETS.insert(ets_ref, {correlation_id, pid})
  end

  defp unregister_corr_id(ets_ref, correlation_id) do
    ETS.delete(ets_ref, correlation_id)
  end

  defp gen_pid_from_corr_id(ets_ref, correlation_id) do
    [{_, pid}] = ETS.lookup(ets_ref, correlation_id)
    pid
  end

  defp setup_queue(chan, queue_name \\ "") do
    {:ok, responce} =
      Queue.declare(
        chan,
        queue_name,
        durable: false,
        auto_delete: true
      )

    {:ok, responce.queue}
  end

  defp generate_correlation_id() do
    Crypto.strong_rand_bytes(16) |> Base.encode16()
  end
end