Kombu for Elixir

Hex.pm HexDocs License: BSD-3-Clause

Warning

This port is partial and only supports functionality to be able to write OpenRelik workers!

Kombu is an Elixir port of the Python Celery Kombu messaging library (v5.7.0rc1), designed for byte-for-byte wire-format and Redis data-structure compatibility with Python Kombu and Celery.

Publish messages from Python Kombu/Celery and consume them in Elixir—or publish from Elixir and consume in Python—over Redis without any custom adapters or bridges.

Features

Installation

Add kombu to your list of dependencies in mix.exs:

def deps do
  [
    {:kombu, "~> 0.1.0"}
  ]
end

Quick Start

Publishing and Consuming over Redis

alias Kombu.{Connection, Exchange, Queue, Producer, Consumer, Message}

# 1. Connect to Redis
conn = Kombu.connection("redis://localhost:6379/0")

# 2. Define an Exchange and Queue
exchange = Kombu.exchange("media", type: "direct", durable: true)
queue = Kombu.queue("video", exchange, "video", durable: true)

# 3. Start a Consumer
consumer =
  Kombu.consumer(conn, [queue],
    accept: ["json"],
    callbacks: [
      fn body, message ->
        IO.inspect(body, label: "Received payload")
        Message.ack(message)
      end
    ]
  )

Consumer.consume(consumer)

# 4. Publish a Message (with optional zlib compression and priority)
producer = Kombu.producer(conn, exchange: exchange, routing_key: "video", serializer: "json")

Producer.publish(
  producer,
  %{"name" => "movie.mp4", "size" => 1024},
  compression: "zlib",
  headers: %{"x-source" => "elixir"},
  priority: 3
)

# 5. Drain events
Connection.drain_events(conn, timeout: 5.0)

Consumer.cancel(consumer)
Connection.close(conn)

SimpleQueue API

For simple one-off point-to-point queues without explicit Exchange/Consumer boilerplate:

alias Kombu.{Message, Simple.SimpleQueue}

conn = Kombu.connection("redis://localhost:6379/0")
sq = Kombu.simple_queue(conn, "simple_jobs")

SimpleQueue.put(sq, %{"task" => "send_email", "to" => "user@example.com"})

message = SimpleQueue.get(sq, timeout: 5.0)
IO.inspect(Message.payload(message))
Message.ack(message)

SimpleQueue.close(sq)
Connection.close(conn)

Topic Exchanges & Wildcard Routing

conn = Kombu.connection("redis://localhost:6379/0")
exchange = Kombu.exchange("stocks", type: "topic")
nyse_queue = Kombu.queue("nyse_feed", exchange, "stock.*.nyse")

# Bind & declare queue on the channel
ch = Kombu.Connection.default_channel(conn)
nyse_queue |> Kombu.Queue.bind(ch) |> Kombu.Queue.declare()

producer = Kombu.producer(conn, exchange: exchange)
Kombu.Producer.publish(producer, %{"symbol" => "AAPL", "price" => 195.5}, routing_key: "stock.tech.nyse")

Pidbox (Remote Control & Inspection Mailbox)

Compatible with kombu.pidbox used by Celery for broadcasting control commands (ping, stats, revoke, etc.) to worker nodes:

alias Kombu.Pidbox.{Mailbox, Node}

conn = Kombu.connection("redis://localhost:6379/0")
mailbox = Mailbox.new("celery", connection: conn)

# Worker node registering a command handler
node =
  Node.new("worker1@host", mailbox, state: %{processed: 42})
  |> Node.handler("status", fn state, _args ->
    %{"ok" => true, "processed" => state.processed}
  end)

consumer = Node.listen(node)

# Client calling the worker node and collecting replies
replies = Mailbox.call(mailbox, ["worker1@host"], "status", %{}, timeout: 2.0)
# => [%{"worker1@host" => %{"ok" => true, "processed" => 42}}]

Python ↔ Elixir Interoperability

Because Kombu for Elixir uses the exact same Redis keys, Lua scripts, and JSON serialization format as Python kombu, Python producers and Elixir consumers (or vice versa) work together out of the box:

Python Producer:

from kombu import Connection, Exchange, Queue, Producer

exchange = Exchange("tasks", type="direct")
queue = Queue("default", exchange=exchange, routing_key="default")

with Connection("redis://localhost:6379/0") as conn:
    producer = Producer(conn, exchange=exchange, routing_key="default", serializer="json", compression="zlib")
    producer.publish({"hello": "from_python"}, declare=[queue], priority=6)

Elixir Consumer:

conn = Kombu.connection("redis://localhost:6379/0")
exchange = Kombu.exchange("tasks", type: "direct")
queue = Kombu.queue("default", exchange, "default")

consumer =
  Kombu.consumer(conn, [queue],
    callbacks: [
      fn body, msg ->
        # body == %{"hello" => "from_python"}
        Kombu.Message.ack(msg)
      end
    ]
  )

Kombu.Consumer.consume(consumer)
Kombu.Connection.drain_events(conn, timeout: 5.0)

Testing

License

Released under the BSD 3-Clause License, matching the original Python celery/kombu license. See LICENSE for full details.