Kombu for Elixir

Copy Markdown View Source

Hex.pm HexDocs License: BSD-3-Clause

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

  • 100% Wire-Format & Redis Key Compatibility with Python Kombu:
    • Identical JSON message envelope (body, content-type, content-encoding, properties, headers with Base64-encoded body support).
    • Compatible Redis exchange binding sets (_kombu.binding.<exchange> with \x06\x16 separator).
    • Priority queue lists (<queue>\x06\x16<pri> across steps [0, 3, 6, 9]).
    • Reliable delivery QoS via unacked hash, unacked_index ZSET, and visibility timeout recovery (restore_visible) using atomic Lua scripts.
  • Exchange Types:
    • direct, topic (AMQP * and # wildcard matching), and fanout (via Redis Pub/Sub).
  • Zero Extra Serialization/Compression Dependencies:
    • Built on Elixir's standard library JSON module with Python Kombu's __type__ / __value__ extended type tags (datetime, date, time, decimal, uuid, bytes, set) and :zlib (application/x-gzip) compression.
  • High-Level Messaging Primitives:
  • Transports:
    • Redis (redis://, rediss://, redis+socket://) powered by Redix.
    • Memory (memory://) in-process virtual broker for fast, isolated unit tests.

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

  • Unit tests (in-memory virtual transport & mocked Redis):
    mix test
  • Live Redis integration tests (requires Redis on localhost:6379):
    mix test --include integration
  • Python ↔ Elixir cross-language interop tests:
    mix test --include interop

License

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