# Transports

PhoenixMicro supports five transports. Only one is active at a time (set via `:transport`).
Switching transports requires only a config change — no consumer or publisher code changes.

> **Zero native deps.** `phoenix_micro` itself compiles on any platform with no C compiler,
> no rebar3, and no native code. Kafka is fully built-in. NATS and Redis are pure Elixir.
> Only RabbitMQ requires a system toolchain (`escript.exe` / rebar3).

## Transport comparison

| Transport     | Dep to add to YOUR app | Linux/macOS | Windows                        | Notes                     |
| ------------- | ---------------------- | ----------- | ------------------------------ | ------------------------- |
| **Kafka**     | **none — built-in**    | ✅          | ✅                             | Pure Elixir wire protocol |
| **In-memory** | **none — built-in**    | ✅          | ✅                             | Dev/test only             |
| NATS          | `{:gnat, "~> 1.7"}`    | ✅          | ✅                             | Pure Elixir               |
| Redis Streams | `{:redix, "~> 1.5"}`   | ✅          | ✅                             | Pure Elixir               |
| RabbitMQ      | `{:amqp, "~> 3.3"}`    | ✅          | ⚠️ Needs `escript.exe` on PATH | Uses rebar3               |

---

## Kafka (built-in)

**No dependency required.** PhoenixMicro implements the Kafka binary wire protocol
natively over `:gen_tcp`. No `kafka_ex`, no `:brod`, no `crc32cer`, no C compiler.

### Supported Kafka APIs

| API Key | Name            | Used for                              |
| ------- | --------------- | ------------------------------------- |
| 0       | Produce         | Publishing messages                   |
| 1       | Fetch           | Consuming messages                    |
| 2       | ListOffsets     | Resolve `:latest`/`:earliest` offsets |
| 8       | OffsetCommit    | Commit consumer offsets               |
| 9       | OffsetFetch     | Fetch committed offsets               |
| 10      | FindCoordinator | Locate group coordinator broker       |
| 11      | JoinGroup       | Join consumer group                   |
| 12      | Heartbeat       | Keep group membership alive           |
| 13      | LeaveGroup      | Clean shutdown from group             |
| 14      | SyncGroup       | Receive partition assignment          |

### Configuration

```elixir
config :phoenix_micro,
  transport: :kafka,
  transports: [
    kafka: [
      # Broker connection (pick one):
      brokers: [{"localhost", 9092}],
      # OR: url: "kafka://broker1:9092,broker2:9092,broker3:9092",

      group_id:           "my_app",
      client_id:          "my_app_client",
      begin_offset:       :latest,       # :latest | :earliest | integer
      acks:               1,             # 0=none, 1=leader, -1=all replicas
      ack_timeout_ms:     5_000,
      max_bytes:          1_048_576,     # 1 MB per fetch
      fetch_wait_ms:      500,           # max block time per poll
      heartbeat_ms:       3_000,
      session_timeout_ms: 30_000
    ]
  ]
```

### Per-message publish options

```elixir
PhoenixMicro.publish("my.topic", payload)

# With options
PhoenixMicro.publish("my.topic", payload,
  partition: 2,          # target specific partition
  acks: -1,              # wait for all replicas
  ack_timeout: 10_000    # override default ack timeout
)
```

### Consumer with Kafka transport override

```elixir
defmodule MyApp.Consumers.PaymentConsumer do
  use PhoenixMicro.Consumer

  topic "payments.created"
  transport :kafka           # explicit — uses kafka even if default is :memory
  concurrency 10
  retry max_attempts: 5
  dead_letter_topic "payments.created.dlq"

  @impl PhoenixMicro.Consumer
  def handle(message, _ctx) do
    IO.inspect(message.metadata)  # %{offset: 42, partition: 0, group_id: "my_app"}
    :ok
  end
end
```

### Docker Compose

```yaml
services:
  kafka:
    image: bitnami/kafka:3.7
    ports:
      - "9092:9092"
    environment:
      KAFKA_CFG_NODE_ID: "0"
      KAFKA_CFG_PROCESS_ROLES: "broker,controller"
      KAFKA_CFG_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
      KAFKA_CFG_ADVERTISED_LISTENERS: "PLAINTEXT://localhost:9092"
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "0@kafka:9093"
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
```

### Multi-broker (production)

```elixir
config :phoenix_micro,
  transport: :kafka,
  transports: [
    kafka: [
      url: "kafka://broker1:9092,broker2:9092,broker3:9092",
      group_id: "my_app_prod",
      acks: -1,              # require all replicas for durability
      session_timeout_ms: 60_000
    ]
  ]
```

---

## NATS

**Add to YOUR app:** `{:gnat, "~> 1.7"}` — pure Elixir, no rebar3.

```elixir
config :phoenix_micro,
  transport: :nats,
  transports: [
    nats: [
      host:        "localhost",
      port:        4222,
      queue_group: "my_app",   # load-balance across instances
      username:    "user",     # optional
      password:    "pass",     # optional
      tls:         false       # optional
    ]
  ]
```

- Queue groups distribute load across consumer instances automatically
- Core NATS is fire-and-forget — ack/nack are local-only
- Supports `*` (single token) and `>` (multi-token) wildcards: `payments.*`, `events.>`
- Automatic reconnection with exponential backoff on connection loss
- Handler dispatch via `Task.Supervisor` — crashes never kill the transport

### Docker Compose

```yaml
services:
  nats:
    image: nats:2.10
    ports:
      - "4222:4222"
```

---

## Redis Streams

**Add to YOUR app:** `{:redix, "~> 1.5"}` — pure Elixir, no rebar3.

```elixir
config :phoenix_micro,
  transport: :redis_streams,
  transports: [
    redis_streams: [
      url:            "redis://localhost:6379",
      consumer_group: "my_app",
      consumer_name:  "node1",       # unique per node/instance
      batch_size:     10,
      block_ms:       1_000,         # XREADGROUP blocking timeout
      max_pending_ms: 60_000         # XAUTOCLAIM PEL recovery threshold
    ]
  ]
```

- Consumer groups via `XREADGROUP`/`XACK` — competing consumers out of the box
- PEL recovery: unacked messages reclaimed after `max_pending_ms` via `XAUTOCLAIM`
- DLQ routing to `dlq:<stream>` prefix stream

### Docker Compose

```yaml
services:
  redis:
    image: redis:7.2
    ports:
      - "6379:6379"
```

---

## RabbitMQ

**Add to YOUR app:** `{:amqp, "~> 3.3"}`

> ⚠️ **Windows:** `amqp` depends on `rabbit_common` which builds with rebar3. You need
> `escript.exe` (part of Erlang/OTP) on your `PATH`. On Linux/macOS this works automatically.

```elixir
config :phoenix_micro,
  transport: :rabbitmq,
  transports: [
    rabbitmq: [
      url:            "amqp://guest:guest@localhost",
      exchange:       "my_app",      # topic exchange name
      prefetch_count: 10,            # per-consumer QoS
      durable:        true
    ]
  ]
```

- Topic exchange with `#` (multi-word) and `*` (single-word) wildcards
- Publisher confirms for reliable publishing
- Dead-letter exchange (DLX) routing on NACK
- Automatic reconnection with exponential backoff
- Handler dispatch via `Task.Supervisor`
- `find_handler` correctly matches routing keys against subscription patterns

### Docker Compose

```yaml
services:
  rabbitmq:
    image: rabbitmq:3.13-management
    ports:
      - "5672:5672"
      - "15672:15672" # management UI
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest
```

---

## In-memory (testing)

Built-in. No dependency. No external service. Always running alongside any real transport.

```elixir
# config/test.exs
config :phoenix_micro,
  transport: :memory,
  consumers: []
```

### Test helpers

```elixir
alias PhoenixMicro.Transport.Memory

Memory.messages()               # all published messages in this session
Memory.dlq_messages()           # all dead-lettered messages
Memory.clear()                  # reset state between tests

# Block until N messages arrive on a topic (or timeout)
Memory.wait_for_messages("payments.created", 1, timeout: 2_000)
```

### Wildcards

The memory transport supports NATS-style wildcards:

- `*` — matches a single dot-separated token: `payments.*` matches `payments.created`
- `>` — matches zero or more tokens: `events.>` matches `events.a.b.c`

---

## Switching transports

Because all consumer and publisher code is transport-agnostic, switching is a
one-line config change:

```elixir
# Was: config :phoenix_micro, transport: :rabbitmq
# Now: config :phoenix_micro, transport: :kafka
```

No consumer modules, handlers, or publish calls need to change.