# Kombu for Elixir

[![Hex.pm](https://img.shields.io/hexpm/v/kombu.svg)](https://hex.pm/packages/kombu)
[![HexDocs](https://img.shields.io/badge/hex-docs-blue.svg)](https://hexdocs.pm/kombu)
[![License: BSD-3-Clause](https://img.shields.io/badge/License-BSD_3--Clause-blue.svg)](https://opensource.org/licenses/BSD-3-Clause)

**Kombu** is an Elixir port of the Python [Celery Kombu](https://github.com/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**:
  - `Kombu.Connection`, `Kombu.Exchange`, `Kombu.Queue`, `Kombu.Binding`
  - `Kombu.Producer` (including batch publishing) & `Kombu.Consumer`
  - `Kombu.Simple.SimpleQueue` & `Kombu.Simple.SimpleBuffer`
  - `Kombu.Pools` (`ConnectionPool` & `ProducerPool`)
  - `Kombu.ConsumerServer` (`kombu.mixins.ConsumerMixin` OTP equivalent)
  - `Kombu.Pidbox` (`Mailbox` and `Node` for Celery worker control/inspection RPC)
- **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`:

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

---

## Quick Start

### Publishing and Consuming over Redis

```elixir
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:

```elixir
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

```elixir
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:

```elixir
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:**
```python
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:**
```elixir
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):
  ```bash
  mix test
  ```
- **Live Redis integration tests** (requires Redis on `localhost:6379`):
  ```bash
  mix test --include integration
  ```
- **Python ↔ Elixir cross-language interop tests**:
  ```bash
  mix test --include interop
  ```

---

## License

Released under the **BSD 3-Clause License**, matching the original Python [celery/kombu](https://github.com/celery/kombu/blob/main/LICENSE) license. See [LICENSE](LICENSE) for full details.