rabbitex

defmodule MQ.Support.TestConsumer do
  alias Core.Name
  alias MQ.Consumer
  alias MQ.Support.TestConsumerRegistry

  @this_module __MODULE__

  def child_spec(opts) do
    :ok = TestConsumerRegistry.init()

    consumer_tag = Name.random_id()
    pid = opts |> Keyword.fetch!(:pid)

    :ok = TestConsumerRegistry.register_pid(consumer_tag, pid)

    opts =
      opts
      |> Keyword.take([:queue])
      |> Keyword.merge(
        consumer_tag: consumer_tag,
        module: @this_module,
        prefetch_count: 1
      )

    %{
      id: consumer_tag,
      start: {Consumer, :start_link, [opts]}
    }
  end

  def process_message(payload, %{consumer_tag: consumer_tag} = meta) do
    {:ok, pid} = TestConsumerRegistry.lookup_pid(consumer_tag)

    message =
      case Jason.decode(payload) do
        {:ok, decoded_message} ->
          {:json, decoded_message, meta}

        _ ->
          {:binary, payload, meta}
      end

    send(pid, message)

    :ok
  end
end