conduit_nsq

defmodule ConduitNSQ do
  @moduledoc """
  Implements the Conduit adapter interface for NSQ
  """

  require Logger

  use Conduit.Adapter
  use Supervisor

  alias ConduitNSQ.NSQ

  def child_spec([broker, _, _, _] = args) do
    %{
      id: broker,
      start: {__MODULE__, :start_link, args},
      type: :supervisor
    }
  end

  @doc """
  Implents Conduit.Adapter.start_link/4 callback
  """
  @impl true
  def start_link(broker, topology, subscribers, opts) do
    Supervisor.start_link(
      __MODULE__,
      [broker, topology, subscribers, opts],
      name: name(broker)
    )
  end

  @doc false
  @impl true
  def init([broker, topology, subscribers, opts]) do
    import Supervisor.Spec

    Logger.info("NSQ Adapter started!")

    children = [
      worker(ConduitNSQ.MessageProcessor, [opts]),
      worker(ConduitNSQ.MessagePublisher, [opts]),
      {ConduitNSQ.ProducerGroup, [broker, topology, opts]},
      {ConduitNSQ.ConsumerGroup, [broker, subscribers, opts]}
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end

  @doc """
  Implements Conduit.Adapter.publish/3 callback
  """
  @impl true
  def publish(broker, message, config, opts) do
    NSQ.publish(broker, message, config, opts)
  end

  defp name(broker) do
    Module.concat(broker, Adapter)
  end
end