adapters/amqp — AMQP 0.9.1 (RabbitMQ) Adapter¶

// AMQPChannel wraps the amqp091.Channel methods used by this adapter.
// *amqp091.Channel satisfies this interface.
type AMQPChannel interface {
    ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args amqp091.Table) error
    QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp091.Table) (amqp091.Queue, error)
    QueueBind(queue, key, exchange string, noWait bool, args amqp091.Table) error
    Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp091.Table) (<-chan amqp091.Delivery, error)
    Publish(exchange, key string, mandatory, immediate bool, msg amqp091.Publishing) error
    Ack(tag uint64, multiple bool) error
    Nack(tag uint64, multiple, requeue bool) error
    Close() error
}

// ExchangeBinding describes the topology for a channel.
// Pass in SubscribeOptions or PublishOptions (Phase 1).
// Phase 2: promote to events.ChannelOpt to flow into AsyncAPI spec.
type ExchangeBinding struct {
    Exchange     string       // exchange name; "" = default exchange
    ExchangeType ExchangeType // Direct | Fanout | Topic | Headers
    RoutingKey   string       // routing key; "" = use channel address
    Queue        QueueConfig  // consumer-side queue
    Durable      bool
    AutoDelete   bool
    VHost        string       // defaults to "/"
}

type QueueConfig struct {
    Name       string         // "" = server-assigned (for exclusive queues)
    Durable    bool
    Exclusive  bool
    AutoDelete bool
    Args       amqp091.Table  // x-dead-letter-exchange, x-message-ttl, etc.
}

type ExchangeType string
const (
    Direct  ExchangeType = "direct"
    Fanout  ExchangeType = "fanout"
    Topic   ExchangeType = "topic"
    Headers ExchangeType = "headers"
)

// Subscribe declares topology, binds queue, and starts a consume loop.
// Cancelling ctx stops the loop and unregisters the consumer.
// When AutoAck is false (default), the adapter Acks on fn success and
// Nacks (with requeue) on fn error.
func Subscribe[T any](
    ctx context.Context,
    ch AMQPChannel,
    handle *events.ChannelHandle[T],
    fn func(context.Context, T) error,
    opts SubscribeOptions,
    formats ...format.Format[T],
) error

// Publish encodes msg and publishes it to the exchange with the routing key.
// ExchangeDeclare runs lazily on first call (idempotent).
func Publish[T any](
    ctx context.Context,
    ch AMQPChannel,
    handle *events.ChannelHandle[T],
    msg T,
    vars map[string]string,
    opts PublishOptions,
    formats ...format.Format[T],
) error

type SubscribeOptions struct {
    Binding     ExchangeBinding
    OnError     func(SubscribeError)
    Observer    stats.Observer
    AutoAck     bool   // default false — manual ack/nack
    NackRequeue bool   // default true — requeue on handler error
    ConsumerTag string // "" = server-assigned
    PrefetchCount int  // Qos prefetch; 0 = no limit
}

type PublishOptions struct {
    Binding      ExchangeBinding
    Observer     stats.Observer
    Mandatory    bool
    ContentType  string
    Headers      amqp091.Table
    DeliveryMode byte // 1=transient, 2=persistent (default)
}

// Serve consumes from the request queue, decodes each delivery, calls fn,
// encodes the response, and publishes it to delivery.ReplyTo.
// Topology setup (ExchangeDeclare + QueueDeclare + QueueBind) runs automatically.
// When fn or encoding fails, an error reply is published so the caller does not block.
func Serve[Req, Resp any](
    ctx context.Context,
    ch AMQPChannel,
    handle *reqreply.RouteHandle[Req, Resp],
    fn func(context.Context, Req) (Resp, error),
    opts ServeOptions,
) error

// Call encodes req, publishes it to the request exchange with ReplyTo and CorrelationId set,
// then waits for a matching reply on the auto-declared exclusive reply queue.
// On success, returns the decoded response.
// On timeout, returns CallError{Kind: KindTimeout}.
// On server error reply, returns CallError{Kind: KindHandler}.
func Call[Req, Resp any](
    ctx context.Context,
    ch AMQPChannel,
    handle *reqreply.RouteHandle[Req, Resp],
    req Req,
    opts CallOptions,
) (Resp, error)

type ServeOptions struct {
    // Binding describes the topology for the request queue.
    Binding     ExchangeBinding
    OnError     func(ServeError)
    Observer    stats.Observer
    // PrefetchCount limits concurrent in-flight requests. Default 1 (serialise).
    PrefetchCount int
}

type CallOptions struct {
    // Binding describes the exchange and routing key to publish the request to.
    Binding     ExchangeBinding
    Observer    stats.Observer
    Timeout     time.Duration        // default 30s
    Headers     amqp091.Table        // optional AMQP headers on outgoing request
    // Vars substitutes {varName} placeholders in the route topic before publishing.
    Vars        map[string]string
}

// ServeError — per-request decode/handler/encode failure on the responder side.
type ServeError struct {
    Kind       ErrorKind // KindDecode | KindHandler | KindEncode
    RoutingKey string
    Err        error
}
// LogValue emits: {kind, routing_key, err}

// CallError — caller-side failure: encode, timeout, server error, decode.
type CallError struct {
    Kind ErrorKind // KindEncode | KindTimeout | KindHandler | KindDecode
    Err  error
}
// LogValue emits: {kind, err}

// Layer 2: declare route (same as MQTT5 / ZeroMQ)
var ComputeRoute = reqreply.NewRoute[ComputeReq, ComputeResp](
    "compute.add",
    computeReqCodec, computeRespCodec,
    reqreply.RouteMeta{OperationID: "computeAdd"},
)

b := reqreply.NewBuilder(reqreply.Info{Title: "Compute API", Version: "1.0.0"})
b.AddServer("rabbitmq", reqreply.Server{URL: "amqp://localhost:5672", Protocol: "amqp"})
handle, _ := ComputeRoute.Register(b)

// Responder — runs in a goroutine
go amqpadapter.Serve(ctx, ch, handle,
    func(ctx context.Context, req ComputeReq) (ComputeResp, error) {
        return ComputeResp{Sum: req.X + req.Y}, nil
    },
    amqpadapter.ServeOptions{
        Binding: amqpadapter.ExchangeBinding{
            Exchange:     "compute",
            ExchangeType: amqpadapter.Direct,
            RoutingKey:   "compute.add",
            Queue: amqpadapter.QueueConfig{
                Name:    "compute-add-workers",
                Durable: true,
            },
            Durable: true,
        },
        Observer:      obs,
        PrefetchCount: 4, // handle up to 4 concurrent requests
    },
)

// Caller
resp, err := amqpadapter.Call(ctx, ch, handle,
    ComputeReq{X: 3, Y: 4},
    amqpadapter.CallOptions{
        Binding: amqpadapter.ExchangeBinding{
            Exchange:     "compute",
            ExchangeType: amqpadapter.Direct,
            RoutingKey:   "compute.add",
            Durable:      true,
        },
        Timeout:  5 * time.Second,
        Observer: obs,
    },
)
// resp.Sum == 7

type ErrorKind int
const (
    KindDecode  ErrorKind = iota // payload could not be decoded
    KindHandler                  // application handler returned error
    KindEncode                   // response/outgoing payload failed encode
    KindTimeout                  // no reply received within deadline (Call only)
    KindAck                      // Delivery.Ack/Nack failed after successful handler
)

// SubscribeError — per-message failure during consume loop.
// Carries exchange, queue, and routing key for correlation with topology.
type SubscribeError struct {
    Kind       ErrorKind
    Exchange   string
    Queue      string
    RoutingKey string
    Err        error
}
// LogValue emits: {kind, exchange, queue, routing_key, err}

// PublishEncodeError — payload encode failure before the message is sent.
type PublishEncodeError struct {
    Exchange   string
    RoutingKey string
    Err        error
}
// LogValue emits: {exchange, routing_key, err}

// BrokerError — AMQP infrastructure failure (topology setup or publish/consume).
type BrokerError struct {
    Op  string // "exchange_declare" | "queue_declare" | "queue_bind" | "consume" | "publish"
    Err error
}
// LogValue emits: {op, err}

import (
    amqpadapter "github.com/DaniDeer/go-codex/adapters/amqp"
    amqp091 "github.com/rabbitmq/amqp091-go"
    "github.com/DaniDeer/go-codex/api/events"
)

// Layer 1: declare codec
var SensorReadingCodec = codex.Struct[SensorReading](...)

// Layer 2: declare channel (same as MQTT)
var SensorChannel, _ = events.NewChannel[SensorReading](
    "sensors.readings",
    SensorReadingCodec,
    events.Subscribe{Summary: "Sensor readings from the field."},
    events.Publish{Summary: "Publish a sensor reading."},
).Register(eventsBuilder)

// Layer 3: connect and adapt
conn, _ := amqp091.Dial("amqp://guest:guest@localhost:5672/")
ch, _ := conn.Channel()

// Subscribe — topology setup runs automatically
err := amqpadapter.Subscribe(ctx, ch, SensorChannel,
    func(ctx context.Context, r SensorReading) error {
        return store.Save(ctx, r)
    },
    amqpadapter.SubscribeOptions{
        Binding: amqpadapter.ExchangeBinding{
            Exchange:     "sensor-events",
            ExchangeType: amqpadapter.Topic,
            RoutingKey:   "sensors.#",
            Queue: amqpadapter.QueueConfig{
                Name:    "sensor-readings-consumer",
                Durable: true,
            },
            Durable: true,
        },
        Observer: obs,
        OnError:  func(e amqpadapter.SubscribeError) { slog.Warn("subscribe error", "error", e) },
    },
)

// Publish — exchange declared lazily
err = amqpadapter.Publish(ctx, ch, SensorChannel, reading, nil,
    amqpadapter.PublishOptions{
        Binding: amqpadapter.ExchangeBinding{
            Exchange:     "sensor-events",
            ExchangeType: amqpadapter.Topic,
            RoutingKey:   "sensors.readings",
            Durable:      true,
        },
        Observer: obs,
    },
)