pipeline.ex

defmodule ExJob.Pipeline do
  @moduledoc """
  A Pipeline represents the workflow for enqueueing and consuming work
  from a single queue.

  Each pipeline can have different queue, worker pool and dequeue strategies.
  """

  use Supervisor

  alias ExJob.Pipeline.{Queue, Runner}

  def start_link(job_module, opts \\ [])  do
    Supervisor.start_link(__MODULE__, job_module, opts)
  end

  def init(job_module) do
    children = children_for(job_module)
    Supervisor.init(children, strategy: :one_for_one)
  end

  defp children_for(job_module) do
    [{Queue, name: queue_name(job_module)}, {Runner, subscribe_to: [queue_name(job_module)]}]
  end

  defp queue_name(job_module), do: String.to_atom("#{job_module}-queue")

  def enqueue(pid, job) do
    queue = queue_stage(pid)
    Queue.enqueue(queue, job)
  end

  defp queue_stage(supervisor) do
    {_, pid, _, _} = supervisor
    |> Supervisor.which_children
    |> Enum.find(fn {_, _, _, modules} -> modules == [Queue] end)
    pid
  end
end