Event Channels — MQTT & AsyncAPI¶

See also: api/events on pkg.go.dev · adapters/mqtt on pkg.go.dev · Feature: Protocol-Native Capabilities (QoS/Retained/HWM/Conflate)

api/events is a transport-agnostic event channel client. The same events.Client that drives typed decode/encode also generates a complete AsyncAPI 3.0 spec.

Declaring channels¶

client := events.NewClient(
    events.WithInfo(events.Info{Title: "Sensor Platform", Version: "1.0.0"}),
    events.WithTopicConstraints(validate.MQTTPublishTopic),
)
client.AddServer("production", events.Server{URL: "mqtt://broker.example.com:1883", Protocol: "mqtt"})

// Static topic — NewChannel opts are ChannelMeta/TopicParam only;
// fork into a role via WithSubscribe/WithPublish, then Register (handler
// attached) or Handle (spec-only) against the client.
userCreatedChannel := events.NewChannel[UserCreatedEvent]("user/created", userCreatedCodec)
userCreatedSub := userCreatedChannel.WithSubscribe(events.Subscribe{
    OperationID: "receiveUserCreated", Summary: "A user was created", SchemaName: "UserCreatedEvent",
})
handle, err := userCreatedSub.Handle(client) // spec-only registration; returns *ChannelHandle[T]

// Template topic with parameter codec
sensorUUIDCodec := codex.String().Refine(validate.UUID)
sensorMeasurement := events.NewChannel[Measurement]("sensors/{sensorID}/measurements",
    measurementCodec,
    events.TopicParam{
        Name:        "sensorID",
        Description: "UUID of the sensor.",
    }.WithCodec(sensorUUIDCodec),
)
measurementSub := sensorMeasurement.WithSubscribe(events.Subscribe{OperationID: "receiveMeasurement", SchemaName: "Measurement"})
measurementPub := sensorMeasurement.WithPublish(events.Publish{OperationID: "publishMeasurement"})

Attaching security or general-purpose middleware (Observer-style logging, etc.) to a channel uses the SAME role-scoped path shown above — Channel.WithSubscribe(events.Subscribe{...})/Channel.WithPublish(events.Publish{...}) each return a Subscriber[T]/Publisher[T] builder with its own .Use(mws ...middleware.Middleware) and a terminal .Handle(client) (spec-only) or .WithHandler(fn).Register(client) (spec + local dispatch registration, requires a handler):

handle, err := sensorMeasurement.
    WithSubscribe(events.Subscribe{OperationID: "receiveMeasurement"}).
    Use(events.SecurityMiddleware[struct{}, struct{}]("apiKey", scheme, nil)).
    Handle(client)

Parameter types¶

TopicParam has no Required field — topic variables must always be present (same rationale as PathParam).

BuildTopic — type-safe topic construction¶

measurementHandle, err := measurementSub.Handle(client) // *ChannelHandle[Measurement]
topic, err := measurementHandle.BuildTopic(map[string]string{"sensorID": "f47ac10b-..."})
// → "sensors/f47ac10b-.../measurements"
// err: events.TopicParamError or events.MissingTopicVarError on failure

Paho MQTT adapter — subscribing¶

import (
    amqtt "github.com/DaniDeer/go-codex/adapters/mqtt"
    "github.com/DaniDeer/go-codex/api/events"
)

sub := sensorMeasurement.WithSubscribe(events.Subscribe{})
transport := amqtt.NewSubscribeTransport[Measurement](client,
    amqtt.SubscribeOptions{
        // Capabilities declares QoS/Retained protocol-natively — see
        // [Feature: Protocol-Native Capabilities](capabilities.md).
        Capabilities: []amqtt.Capability{amqtt.QoS(1)},
        Observer:     obs,
        OnError: func(e amqtt.SubscribeError) {
            switch e.Kind {
            case amqtt.KindDecode:
                logger.Warn("decode error", "topic", e.Topic, "error", e.Err)
            case amqtt.KindHandler:
                logger.Error("handler error", "topic", e.Topic, "error", e.Err)
            }
        },
    },
)
// mqtt v3 has no broker router, so events.SubscribeHandle registers then
// blocks on ctx.Done() — run it in a goroutine.
go func() {
    _ = events.SubscribeHandle(ctx, sub, transport,
        func(ctx context.Context, m Measurement) error {
            // MessageFromContext gives access to QoS, retained flag, etc.
            if msg, ok := amqtt.MessageFromContext(ctx); ok {
                if msg.Retained() { return nil } // skip stale retained messages
            }
            return svc.HandleMeasurement(ctx, m)
        },
    )
}()

Paho MQTT adapter — publishing¶

alertPub := alertChannel.WithPublish(events.Publish{})
sensorPub := sensorMeasurement.WithPublish(events.Publish{})

// Static topic — vars derived automatically from the channel's merge fields
pubTransport := amqtt.NewPublishTransport[Alert](client, amqtt.PublishOptions[Alert]{
    Capabilities: []amqtt.Capability{amqtt.QoS(1)},
    Observer:     obs,
})
err := events.PublishHandle(ctx, alertPub, pubTransport, alert)

// Template topic — BuildTopic called internally from m's own merge-capable fields
measurementTransport := amqtt.NewPublishTransport[Measurement](client, amqtt.PublishOptions[Measurement]{
    Capabilities: []amqtt.Capability{amqtt.QoS(1)},
    Observer:     obs,
})
err = events.PublishHandle(ctx, sensorPub, measurementTransport, m)

SubscribeError.Topic — always the concrete incoming message topic, even for template channels (e.g. sensors/abc-123/measurements, never sensors/{sensorID}/measurements). Use this in OnError logging to identify the exact message that failed.

SubscribeError.As(target any) bool (adapters/mqtt/adapters/mqtt5/adapters/zeromq, Topic 7 of docs/design/d-0005-error-handling.md) — a thin convenience wrapper over errors.As(e.Err, target), collapsing the two-step errors.As(subErr.Err, &target) dance into subErr.As(&target) from inside OnError:

opts.OnError = func(subErr mqtt5.SubscribeError) {
    var conflict domain.EmailConflictError
    if subErr.As(&conflict) {
        // ...
    }
}

Unlike Topic 6's ErrorPatternAs (which decodes WIRE BYTES into a typed value), no decode is involved here — Err is already a native Go error value; the subscriber never encoded/decoded it across a wire boundary for this purpose. This mostly matters for failures that matched NO declared ErrorChannel/DeadLetter — once one is declared, most failures never reach OnError at all.

TopicVarsFromMessage — wildcard subscription¶

Extracts and validates {varName} from the concrete received topic — the inverse of BuildTopic:

// Channel: "sensors/{sensorID}/measurements"
// Subscribed to: "sensors/+/measurements"
// Message arrives on: "sensors/f47ac10b-.../measurements"

vars, err := amqtt.TopicVarsFromMessage(sensorMeasurement, msg)
// vars["sensorID"] == "f47ac10b-..."

Wildcard capture rules: - {varName} — captures the corresponding topic segment into the variable - + in the subscription pattern — matches one level (anonymous; not captured into vars) - # as the last segment — matches all remaining levels; captured under key "#"

Validation chain (in order): 1. Structural match (segment count + literals) → TopicMismatchError 2. Builder-level topic codec → InvalidTopicError 3. TopicParam.Codec per variable → TopicParamError

Topic vars with automatic merge — NewTopicParam¶

TopicParam above is validate-only — it checks a topic variable against a codec but leaves extracting it into your payload struct to hand-written code. events.NewTopicParam declares the SAME spec TopicParam AND a merge field in one call — the handler receives an already-merged, already-validated payload, no manual TopicVarsFromMessage extraction:

type SensorReading struct {
    SensorID string
    Value    float64
}

var sensorChannel = events.NewChannel[SensorReading]("sensors/{sensorID}/readings", sensorCodec,
    events.NewTopicParam("sensorID", codex.String().Refine(validate.UUID),
        func(r SensorReading) string { return r.SensorID },
        func(r *SensorReading, v string) { r.SensorID = v },
    ),
)
handle, _ := sensorChannel.WithSubscribe(events.Subscribe{}).Handle(client)

// adapters/mqtt5's subscribe dispatch (reached via events.Client.Subscribe
// or events.SubscribeHandle+NewSubscribeTransport) calls
// ChannelHandle.DecodeMerged automatically whenever handle.MergeFields()
// is non-empty — the handler function just receives a fully populated,
// validated SensorReading:
func(ctx context.Context, r SensorReading) error {
    store.Save(r) // r.SensorID already validated as a UUID
    ...
}

This is the PRIMARY, recommended way to declare a topic variable — but not the SOLE way: the plain TopicParam struct literal remains available for validate-only variables a handler never reads directly. A channel can freely mix both styles.

ChannelHandle.MergeFields() returns the registered merge fields directly (no role-aware split needed — unlike REST's path/query/header/cookie, events has exactly ONE var destination, the topic, so a single flat slice is always safe for both directions). ChannelHandle.DecodeMerged(payload, topicVars) (T, error) closes the loop: decodes the payload (via ChannelHandle.Decode, JSON) AND merges every topic variable into the SAME value, in one call.

events.PublishHandle + each adapter's NewPublishTransport[T] is the single-call publisher convenience (Decision 7 of docs/design/d-0002-pubsub-workflow-simplification.md) — mirrors rest.CallWithTransport: the transport's Publish method derives the topic vars from the payload struct automatically via codex.EncodeVars(msg, handle.MergeFields()...) internally. The same convenience exists for every transport with a pub/sub event surface — mqtt5.NewPublishTransport[T], mqtt.NewPublishTransport[T] (MQTT 3.1.1), and zeromq.NewPublishTransport[T] — identical shape and semantics, one per transport package.

transport := mqtt5.NewPublishTransport[SensorReading](client, mqtt5.PublishOptions[SensorReading]{})
err := events.PublishHandle(ctx, sensorChannel.WithPublish(events.Publish{}), transport, reading)
// topic + payload both derived from the SAME reading value — no manual vars map.

adapters/mqtt5.TopicVarsFromMessage is the mqtt5-specific equivalent of the (Paho MQTT 3.1.1) mqtt.TopicVarsFromMessage shown above — used internally by the subscribe-side auto-merge wiring, and directly usable by hand-rolled mqtt5 consumers. adapters/zeromq.TopicVarsFromMessage is the ZeroMQ equivalent, adapted for zeromq's plain-string topic (the first frame of a [topic, payload] PUB/SUB message) rather than a message struct; zeromq.NewSubscribeTransport's Subscribe calls it internally whenever the channel declares merge fields, exactly like mqtt5/mqtt's subscribe transports do.

Every port-binding SinkAdapter/IOAdapter constructed for one of these transports (nethttp.DrainCallAdapter/CallAdapter, mqtt5.PublishAdapter/CallAdapter, zeromq.PublishAdapter/CallAdapter, mqtt.PublishAdapter) derives vars PER-ITEM automatically via the corresponding *Handle function whenever its Vars option is left nil — the "one struct, one call" convenience extends all the way through ports.SinkPort/ports.IOPort, not just the standalone functions. Set Vars explicitly (even to an empty, non-nil map) to keep today's static, same-vars-for-every-item behavior instead.

Nested structs & non-JSON payloads¶

The merge-field convenience is neither JSON-specific nor flat-struct-specific — the same guarantees documented for REST (Nested structs & binary body formats) apply here:

• Merge-field get/set are plain closures, so a topic variable can target a NESTED sub-struct field (func(r Reading) string { return r.Meta.SensorID }) with zero framework changes.
• Payload decode/encode is orthogonal to topic-var merge, so format.Gob/ format.Binary/any custom format composes unchanged — but the SAME format.Gob(codec) caveat applies: it serialises the WHOLE typed value directly via encoding/gob's reflection, bypassing the codec entirely for the wire bytes. Use format.NewTyped with a custom marshal/ unmarshal to project the wire bytes onto ONLY a nested payload sub-field.

See examples/events-nested-binary for the full runnable version (nested Meta/Value payload, Gob body projected via format.NewTyped, topic merge into the nested field).

Multi-format MQTT payloads¶

MQTT 3.1.1 carries no content-type metadata — format is agreed out-of-band. Configure the default format once on the handle; WithFormats also updates the AsyncAPI spec: the first format's content type is written to message.contentType on each registered operation.

Structured logging: all codec error types (ValidationErrors, ConstraintError, TypeMismatchError, etc.) implement slog.LogValuer. Pass them directly to slog.Any(...) for full nested structured output — field names, constraint details, type mismatches — without any string parsing.

// Configure YAML as default — adapter picks it up automatically
yamlChannel := measurementCh.WithFormats(format.YAML(measurementCodec))

sub := yamlChannel.WithSubscribe(events.Subscribe{})
pub := yamlChannel.WithPublish(events.Publish{})

transport := amqtt.NewSubscribeTransport[Measurement](client, opts)
go func() { _ = events.SubscribeHandle(ctx, sub, transport, handler) }()

pubTransport := amqtt.NewPublishTransport[Measurement](client, amqtt.PublishOptions[Measurement]{})
err := events.PublishHandle(ctx, pub, pubTransport, m)

// Call-time format override still works — passed as trailing variadic to the constructor
jsonTransport := amqtt.NewSubscribeTransport[Measurement](client, opts, format.JSON(measurementCodec))
go func() { _ = events.SubscribeHandle(ctx, sub, jsonTransport, handler) }()

Format priority: call-time variadic → handle.SubscribeFormats/PublishFormats → handle.Formats → JSON fallback.

events.Formats/SubscribeFormats/PublishFormats declared INLINE as NewChannel opts apply identically whether the channel is registered with a Client (Subscriber.Handle(client)/ Publisher.Handle(client)) or built client-only via Handle(nil) — both populate the SAME ChannelHandle.Formats/SubscribeFormats/PublishFormats fields mqtt5/zeromq's client-side Subscribe/Publish read. A type mismatch (formats declared for the wrong payload type) panics on Handle(nil) (infallible — no error return) with the same message Handle(client) would return as FormatOptError.

TopicParam schema → AsyncAPI spec¶

TopicParam.Codec schema flows automatically into the AsyncAPI parameters: block. Every {varName} placeholder gets a parameter entry: - No TopicParam declared → auto-generated as {type: string} - TopicParam with .WithCodec(c) → codec schema in parameters: + runtime validation at BuildTopic time - TopicParam with .Description only → enriches the spec without runtime validation

// With codec: UUID schema in spec + UUID validation at BuildTopic time
events.TopicParam{Name: "sensorID", Description: "Sensor UUID"}.WithCodec(
    codex.String().Refine(validate.UUID),
)
// Without codec: description only, no runtime validation
events.TopicParam{Name: "region", Description: "Geographic region"}

Client options¶

Template-transparent validation: constraints run on the structural shape of the topic, not the literal template. {varName} placeholders are replaced with x before validation — sensors/{sensorID}/readings → sensors/x/readings. The stored ChannelHandle.Topic is always the original template.

Final topic re-validation: BuildTopic re-validates the fully assembled topic against the builder-level codec after substitution. Returns events.InvalidTopicError{Topic, Err} with the concrete topic on failure.

AsyncAPI spec generation¶

doc, err := client.AsyncAPISpec()
yamlBytes, _ := doc.MarshalYAML()

AsyncAPI 3.0: separate channels and operations top-level keys, action: receive / action: send. The render/asyncapi/v2 package generates AsyncAPI 2.6 for existing users.

Codec-backed middleware (SubscribeMW/PublishMW)¶

events.Middleware[In, Out] mirrors rest.Middleware[In, Out] (see Feature: REST API) for pub/sub's asymmetric shape: subscribe is the RECEIVING role (In decoded from incoming topic vars, Out unused — no reply channel to encode into), publish is the SENDING role (Out encoded into outgoing topic vars via WithPublishTopic, In unused). Reuses the SAME events.NewTopicParam[T,V] constructor a channel's own Item already uses. Transform/ClientTransform (the former route/channel-BOUND, free-function attachment point) were REMOVED (Rollout Phase B) — folded directly into Subscriber.SubscribeMW/ Publisher.PublishMW, which detect a codec-backed Middleware[In,Out]'s bound shape via fn's own reflected signature:

regionPolicy := events.NewMiddleware(
    middleware.NewDeclaration("region-policy", regionInCodec, regionOutCodec),
).WithSubscribeTopic(events.NewTopicParam("region", codex.String(),
    func(in RegionIn) string { return in.Region },
    func(in *RegionIn, v string) { in.Region = v },
))

subscriber = subscriber.SubscribeMW(regionPolicy,
    func(ctx context.Context, msg *SensorReading, in RegionIn) error {
        msg.Region = in.Region
        return nil
    })

Route/channel-AGNOSTIC reuse works the same way as REST, via Middleware.WithReceive/Middleware.WithSend + plain .Use(mw) on Subscriber[T]/Publisher[T]. D6(b)/D7 (name uniqueness, ambiguous dual attachment) and the ErrorChannel-eligible fn-error fallback (events.MiddlewareError) mirror REST exactly. Dispatch is wired into adapters/mqtt, adapters/mqtt5, and adapters/zeromq — subscribe dispatch shares the SAME pre-handler point security enforcement already runs at; publish dispatch derives middleware-contributed topic vars BEFORE BuildTopic, following the FULL 3-tier precedence: explicit (a per-call Vars override) wins over middleware-derived, which in turn wins over the channel's own derived vars, on a key collision.

A SECOND, protocol-neutral property vocabulary axis also exists alongside topic vars, shared with api/reqreply — see Feature: Codec-Declared Middleware for the full explanation (NewPropertyParam/NewOptionalPropertyParam, direct attachment, write-side wiring, AsyncAPI rendering) and D-0003 — Codec-Declared Middlewares for the full design.

Protocol-native QoS and Retained (Capability mechanism)¶

The declarative, transport-agnostic events.MQTTQoS/Subscribe.QoS/ Publisher.WithAttributes design described in earlier revisions of this page was DELETED — it bypassed the sealed Capability mechanism's "zero backdoor between the api layer and the adapters" guardrail. QoS and Retained are now declared per-adapter via Capabilities []<adapter>.Capability on SubscribeOptions/PublishOptions (shown in the Paho MQTT subscribe/publish sections above) — see Feature: Protocol-Native Capabilities for the full mechanism, adapter coverage table, and rationale. adapters/zeromq has no QoS/Retained equivalent (no broker-mediated semantics in ZeroMQ PUB/SUB); its own capabilities (HWM, Conflate) are documented on the same page.

Error types¶

Codec-as-contract pattern¶

// contract/contract.go
var ReadingsChannel = events.NewChannel[SensorReading](
    "sensors/{sensorID}/readings", sensorReadingCodec,
    events.TopicParam{Name: "sensorID"}.WithCodec(SensorIDCodec),
)

// producer/main.go — Handle(producerClient) registers the spec once (e.g. for AsyncAPISpec());
// events.PublishHandle's own transport call is spec-free (Publisher.Handle(nil) internally).
pub := contract.ReadingsChannel.WithPublish(events.Publish{...})
_, _ = pub.Handle(producerClient) // optional — only needed if producerClient generates a spec
transport := amqtt.NewPublishTransport[SensorReading](mqttClient, amqtt.PublishOptions[SensorReading]{})
err := events.PublishHandle(ctx, pub, transport, reading)

// consumer/main.go
sub := contract.ReadingsChannel.WithSubscribe(events.Subscribe{...})
_, _ = sub.Handle(consumerClient) // optional — same spec-registration purpose
subTransport := amqtt.NewSubscribeTransport[SensorReading](mqttClient, opts)
go func() { _ = events.SubscribeHandle(ctx, sub, subTransport, fn) }()

Error-path ergonomics — ErrorChannel¶

Pub/sub channels have no synchronous caller to respond to — events.ErrorChannel is the pub/sub analogue of rest.ErrorPattern: declare a codec-backed typed error payload for a matched error type, published to a dedicated error-output topic instead of an HTTP status/body.

type ValidationError struct{ Reason string }
func (e ValidationError) Error() string { return "validation: " + e.Reason }

type ErrorPayload struct {
    Code    string
    Message string
}

handle, err := events.NewChannel[SensorReading]("sensors/{id}/data", sensorCodec,
    events.ErrorChannel[ValidationError, ErrorPayload](
        "sensors/{id}/errors", errorPayloadCodec,
        func(e ValidationError) (ErrorPayload, error) {
            return ErrorPayload{Code: "validation", Message: e.Reason}, nil
        },
    ),
).WithSubscribe(events.Subscribe{}).Handle(client)

• Direct mode (no map function): E must itself be assignable to the declared payload type B.
• Mapped mode (map function provided): the map function converts E into B.
• Matching: type-only via errors.As; the first declared ErrorChannel (in NewChannel option order) whose type matches wins — the same deterministic precedence used by REST.
• ChannelHandle.ErrorResponseFor(err) (ErrorChannelResponse, bool, error) looks up the first matching pattern — adapters call this before falling back to their own default error handling.
• ChannelHandle.ObserveErrorResponseFor(ctx, obs, err) (ErrorChannelResponse, bool, error) is the RECOMMENDED single call site: performs the SAME match as ErrorResponseFor, but ALSO reports stats.ErrorPatternObserver/ stats.SpanTagger observability internally (see Observer guide) — no separate adapter-side wiring needed.
• ErrorChannel is eligible at EVERY subscribe-side dispatch failure point, not just the handler's own return — payload decode, topic-var extraction/merge, property-var merge (mqtt5), User Property param validation (mqtt5), middleware DecodeIn, and security middleware Fn errors are all ErrorChannel-eligible, exactly like a handler error. The PUBLISH side stays correctly excluded — a publisher plays the same "client/sender" role a REST/reqreply client plays, which never consults a declared error pattern either (see docs/design/d-0005-error-handling.md's Topic 7).

Action model — respond / handle / log¶

A matched pattern executes exactly one action, never an implicit chain:

events.ErrorChannel[ValidationError, ErrorPayload]("sensors/{id}/errors", errorPayloadCodec, mapFn).
    WithAction(events.ErrorLog)

Adapter wiring (adapters/mqtt5, adapters/mqtt, adapters/zeromq)¶

mqtt5.PublishAdapter, mqtt.PublishAdapter, and zeromq.PublishAdapter all consult handle.ErrorResponseFor(err) for every upstream stream error before falling back to their own OnError option:

• matched + respond → publishes the encoded payload to the declared topic (does not call OnError);
• matched + handle → falls through to OnError (unchanged existing behavior — OnError already IS the "handle" realization for these adapters);
• matched + log, or unmatched → falls through to OnError unchanged.

Dead-letter fallback — DeadLetter¶

events.DeadLetter(topic, opts...) declares an OPTIONAL, last-resort sink attempted immediately after ErrorChannel fails to match (or none is declared at all) — a fixed, non-codec-backed envelope, since by this point no reliable business type exists to encode:

type DeadLetterEnvelope struct {
    SourceTopic string
    Payload     []byte // original, undecoded message bytes
    Error       string
    Timestamp   time.Time
}

handle, err := events.NewChannel[SensorReading]("sensors/{id}/data", sensorCodec,
    events.DeadLetter("sensors/dead-letter"),
).WithSubscribe(events.Subscribe{}).Handle(client)

• A DeadLetter topic is ALWAYS a plain, literal string — never a template. Unlike the channel's OWN topic (which may declare {varName} placeholders resolved per-message via NewTopicParam), the dead-letter destination is resolved ONCE at declaration time and used verbatim on every dead-letter publish; {varName}-shaped substrings in a declared DeadLetter topic are NOT substituted and would appear literally in both the published topic and the generated AsyncAPI channel address. This mirrors ErrorChannel's own topic field, which is equally literal-only.
• Client.AddGlobalDeadLetter(topic, opts...) sets an application-wide default; a channel with no DeadLetter declared inherits it. An explicit DeadLetter("") (empty topic) opts a channel OUT of the inherited default — mirrors AddGlobalSecurity's nil-inherit/empty-override convention exactly. ClientHandle() never inherits the global default (same asymmetry as GlobalSecurity).
• ChannelHandle.DeadLetterFor(obs, sourceTopic, rawPayload, err) (topic string, body []byte, ok bool) is the single call site adapters consult — ok is false when no dead-letter rule resolves for this channel.
• Eligible at the SAME dispatch points as ErrorChannel (decode, topic/property-var merge, security middleware Fn, Transform middleware DecodeIn/Fn/EncodeOut, handler error) — on BOTH the subscribe side AND a FAILED publish (the message never reached the broker, or its own encode failed).
• A TYPE-MATCHED ErrorChannel is NEVER ALSO dead-lettered, even when its declared action is ErrorHandle/ErrorLog (not the default ErrorRespond). DeadLetter is scoped to the genuinely UNMATCHED case only (a decode failure before any business error exists, or a business error whose type matches NO declared ErrorChannel) — a type match via errors.As means the failure WAS anticipated and IS declared for, regardless of which action was chosen, so it correctly falls through directly to OnError without also reaching the dead-letter destination.
• On the publish side specifically, dead-lettering is intentionally narrower than the subscribe side's full Category-A coverage. Only a ClientTransform/general-purpose middleware Fn/encode failure and the final broker-rejection are dead-lettered (consistent across adapters/mqtt/mqtt5/zeromq). A BuildTopic failure (invalid topic-var value) or a client-side security implementation Fn rejection are NOT — these are pre-transmission validation/authorization failures on the CALLER's own outgoing message, closer in spirit to REST's client-side param/credential validation (Category C, permanently excluded) than to a message that reached dispatch and then failed — the caller already receives the synchronous error directly in both cases, so nothing is silently lost either way.
• DeadLetter DOES generate its own AsyncAPI channel entry — a receive-only channel at the declared destination topic, carrying the FIXED DeadLetterEnvelope schema. WithCode remains inert metadata (no wire/spec effect); WithDescription/WithSchemaName/ WithChannelAddress/WithOperationID all DO affect the rendered spec, mirroring reqreply.ErrorPatternOpt's equivalents exactly:

events.NewChannel[Reading]("sensors/{id}/data", readingCodec,
    events.DeadLetter("sensors/dead-letter").
        WithDescription("Undeliverable sensor readings.").
        WithSchemaName("SensorDeadLetter"),
).WithSubscribe(events.Subscribe{}).Handle(client)

When several channels share ONE dead-letter destination (e.g. via Client.AddGlobalDeadLetter), the shared topic is registered as a SINGLE spec channel entry (first-registered-wins dedup by topic) — not once per channel that resolves it.

See Error handling guide — dead-letter fallback for the full adapter-reachability caveats (MQTT vs. ZeroMQ REQ/REP).

WebSocket error frames¶

See WebSocket — error frames for the duplex-socket analogue (websocket.ErrorFrame), which broadcasts a typed error frame to every connected session instead of publishing to a declared topic (sockets have no dedicated error-output channel — broadcast IS the notification path).

See also¶

• Feature: Security & Auth — MQTT security via SubscribeMW/PublishMW-paired implementations
• Concept: Go Library as Contract — shared contract pattern
• examples/events-api — the comprehensive multi-adapter pub/sub demo: three-layer domain-boundary pipeline, shared routes/ contract package across mqtt v3/mqtt5/zeromq, SubscribeMW/PublishMW security, events.Observability[T]
• examples/api-events — event builder + AsyncAPI spec