Subscribe to queue events in real-time.

QueueEvents uses Redis Streams to listen for job lifecycle events. This enables reactive patterns and monitoring of job processing.

Usage

# Start the event listener
{:ok, pid} = BullMQ.QueueEvents.start_link(
  queue: "my_queue",
  connection: :redis
)

# Subscribe to events
BullMQ.QueueEvents.subscribe(pid, self())

# Receive events
receive do
  {:bullmq_event, :completed, %{job_id: id, returnvalue: value}} ->
    IO.puts("Job #{id} completed with #{value}")

  {:bullmq_event, :failed, %{job_id: id, failed_reason: reason}} ->
    IO.puts("Job #{id} failed: #{reason}")
end

Events

The following events are emitted:

• :added - Job was added to the queue
• :waiting - Job is waiting to be processed
• :active - Job started processing
• :progress - Job progress was updated
• :completed - Job completed successfully
• :failed - Job failed
• :delayed - Job was delayed
• :stalled - Job was detected as stalled
• :removed - Job was removed
• :drained - Queue has no more waiting jobs
• :paused - Queue was paused
• :resumed - Queue was resumed

Handler Module

For more structured event handling, implement a handler module:

defmodule MyApp.QueueHandler do
  use BullMQ.QueueEvents.Handler

  @impl true
  def handle_event(:completed, %{job_id: id, returnvalue: value}, state) do
    Logger.info("Job #{id} completed")
    {:ok, state}
  end

  @impl true
  def handle_event(:failed, %{job_id: id, failed_reason: reason}, state) do
    Logger.error("Job #{id} failed: #{reason}")
    {:ok, state}
  end

  @impl true
  def handle_event(_event, _data, state) do
    {:ok, state}
  end
end

# Use the handler
BullMQ.QueueEvents.start_link(
  queue: "my_queue",
  connection: :redis,
  handler: MyApp.QueueHandler
)