events

package
v0.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 29, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Overview

Package events provides a transport-agnostic event channel builder for go-codex.

Define channels declaratively with codec-backed payload types; register them with a Builder to obtain a ChannelHandle with typed Decode and Encode helpers. Pass those helpers to any message broker (MQTT, AMQP, Kafka, NATS) — this package does not import any messaging library.

Spec generation is also available: Builder.AsyncAPISpec derives a complete AsyncAPI 3.0 document from the registered channels.

Typical usage: 

b := events.NewBuilder(events.Info{Title: "User Events", Version: "1.0.0"})
b.AddServer("production", events.Server{
    URL:      "mqtt://broker.example.com",
    Protocol: "mqtt",
})

// Declare the channel as a value — define once, pass around, register later.
var userCreated = events.NewChannel[UserCreated]("user/created", userCreatedCodec,
    events.ChannelMeta{Description: "A user was created"},
    events.Subscribe{Summary: "Receive user created events", SchemaName: "UserCreatedEvent"},
)

handle, err := userCreated.Register(b)

// In your broker callback (any library):
event, err := handle.Decode(msg.Payload())   // JSON → UserCreated, validates
payload, err := handle.Encode(event)          // UserCreated → JSON

// AsyncAPI 3.0 spec:
doc, err := b.AsyncAPISpec()
yaml, _  := doc.MarshalYAML()

Encoding is JSON only by default. For other formats construct a format.Format directly and pass it to the adapter (e.g. adapters/mqtt.SubscribeHandler).

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Builder 

type Builder struct {
	// contains filtered or unexported fields
}

Builder accumulates channel registrations and produces AsyncAPI specs. Create one with NewBuilder.

func NewBuilder 

func NewBuilder(info Info, opts ...BuilderOption) *Builder

NewBuilder returns a Builder initialised with the given API metadata.

func (*Builder) AddChannelItem added in v0.12.0 

func (b *Builder) AddChannelItem(topic string, item asyncapi.ChannelItem) *Builder

AddChannelItem registers a pre-built asyncapi.ChannelItem under topic. Use this for channels the single-codec Channel declaration cannot express — a duplex socket whose inbound and outbound frames are different types (ports.RegisterSocket builds the item from a SocketPattern).

The builder-level topic codec is NOT applied — the topic may be an HTTP upgrade path (e.g. "/live/{room}") rather than an MQTT-style topic. SchemaName references in the item's operations participate in the usual dangling-$ref validation at Builder.AsyncAPISpec time.

func (*Builder) AddGlobalSecurity added in v0.8.0 

func (b *Builder) AddGlobalSecurity(reqs ...route.SecurityRequirement) *Builder

AddGlobalSecurity appends security requirements that apply to all channels by default. The requirements flow into runtime enforcement: channels with nil Subscribe.Security or Publish.Security inherit these requirements at the adapter layer via ChannelHandle.GlobalSecurity.

AsyncAPI 3.0 has no document-level global security field; these requirements do NOT appear in the AsyncAPI spec output. To annotate per-channel security in the spec, set Subscribe.Security or Publish.Security explicitly.

To mark a specific channel as explicitly unsecured (exempt from global security), set Security to an empty slice: Security: []route.SecurityRequirement{}.

func (*Builder) AddSchema 

func (b *Builder) AddSchema(name string, s schema.Schema) *Builder

AddSchema registers a named schema in components/schemas. Use this to register reusable schemas that are referenced by SchemaName in channel configs but not inlined in any codec.

func (*Builder) AddSecurityScheme added in v0.8.0 

func (b *Builder) AddSecurityScheme(name string, s SecurityScheme) *Builder

AddSecurityScheme registers a named security scheme with the builder. The spec fields flow into the AsyncAPI document via AsyncAPISpec; Codec, when non-nil, is used by adapters to validate extracted credentials before SecurityFunc is called (MQTT adapters skip Codec validation — use SecurityFunc).

The name must match those used in route.Require calls on Subscribe/Publish Security fields and in Builder.AddGlobalSecurity.

func (*Builder) AddServer 

func (b *Builder) AddServer(name string, s Server) *Builder

AddServer registers a named server entry in the spec. Servers appear in the AsyncAPI output in registration order. If s.Description is empty, name is used as the description.

func (*Builder) AppendTo added in v0.11.0 

func (b *Builder) AppendTo(db *asyncapi.DocumentBuilder) error

AppendTo writes all channels registered on this Builder into db, which must have been created by asyncapi.NewDocumentBuilder. Servers, schemas, and security schemes owned by this Builder are NOT written — the caller is responsible for configuring those on db.

Use AppendTo to combine pub/sub channels with request-reply channels from api/reqreply.Builder in a single AsyncAPI 3.0 document: 

import asyncapi "github.com/DaniDeer/go-codex/render/asyncapi/v3"

doc := asyncapi.NewDocumentBuilder(info)
doc.AddServer("mqtt5", asyncapi.Server{URL: "mqtts://...", Protocol: "mqtt5"})

eventsB.AppendTo(doc)    // pub/sub channels
reqreplyB.AppendTo(doc)  // request-reply channels

spec, err := doc.Build()

func (*Builder) AsyncAPISpec 

func (b *Builder) AsyncAPISpec() (asyncapi.Document, error)

AsyncAPISpec builds a complete AsyncAPI 3.0 document from all registered channels. Returns an error if any non-empty SchemaName references a schema that will not be present in components/schemas (a dangling $ref).

type BuilderOption added in v0.4.0 

type BuilderOption func(*Builder)

BuilderOption configures a Builder at construction time.

func WithTopicCodec added in v0.4.0 

func WithTopicCodec(c codex.Codec[string]) BuilderOption

WithTopicCodec sets a codec used to validate every topic passed to Channel.Register. If the topic is invalid, Channel.Register returns an error immediately.

Use WithTopicConstraints for the common case of stacking one or more codex.Constraint values; use WithTopicCodec when you need a fully-custom codex.Codec.

Example — enforce MQTT publish topic rules: 

import "github.com/DaniDeer/go-codex/validate"

b := events.NewBuilder(info, events.WithTopicConstraints(validate.MQTTPublishTopic))

func WithTopicConstraints added in v0.4.0 

func WithTopicConstraints(cons ...codex.Constraint[string]) BuilderOption

WithTopicConstraints is a convenience wrapper around WithTopicCodec that builds a codec from codex.String refined with the given constraints. Multiple constraints are applied in order; all must pass.

Users can mix built-in constraints from the validate package with their own: 

sensorLevel := codex.Constraint[string]{
    Name:    "sensor-prefix",
    Check:   func(v string) bool { return strings.HasPrefix(v, "sensors/") },
    Message: func(v string) string { return fmt.Sprintf("topic must start with sensors/, got %q", v) },
}
b := events.NewBuilder(info, events.WithTopicConstraints(validate.MQTTPublishTopic, sensorLevel))

type Channel added in v0.8.0 

type Channel[T any] struct {
	// contains filtered or unexported fields
}

Channel is a declarative event channel spec: topic, codec, and options. It is a value type — define it once, store it, pass it around, and register it with one or more Builder instances via Channel.Register.

Create a Channel with NewChannel.

func NewChannel added in v0.8.0 

func NewChannel[T any](
	topic string,
	codec codex.Codec[T],
	opts ...ChannelOpt,
) Channel[T]

NewChannel creates a Channel spec from a topic, codec, and variadic opts. NewChannel is infallible — it only captures the spec. Validation (topic codec, TopicParam template consistency) runs at Channel.Register time.

Pass any combination of ChannelMeta, Subscribe, Publish, and TopicParam as opts. All opts are optional.

NewChannel is a free function (not a method) because Go requires type parameters to appear on free functions, not on method receivers.

Typical usage: 

var userCreated = events.NewChannel[UserCreated]("user/created", userCreatedCodec,
    events.ChannelMeta{Description: "A user was created"},
    events.Subscribe{Summary: "Receive user created events", SchemaName: "UserCreatedEvent"},
)

// Later, register with a builder:
handle, err := userCreated.Register(b)
Example 
package main

import (
	"fmt"

	"github.com/DaniDeer/go-codex/api/events"
	"github.com/DaniDeer/go-codex/codex"
)

func main() {
	type SensorReading struct {
		SensorID string
		Value    float64
	}

	readingCodec := codex.Struct[SensorReading](
		codex.RequiredField("sensor_id", codex.String(),
			func(r SensorReading) string { return r.SensorID },
			func(r *SensorReading, v string) { r.SensorID = v },
		),
		codex.RequiredField("value", codex.Float64(),
			func(r SensorReading) float64 { return r.Value },
			func(r *SensorReading, v float64) { r.Value = v },
		),
	)

	b := events.NewBuilder(events.Info{Title: "Sensor API", Version: "1.0.0"})

	// NewChannel declares a typed channel as a value — register with any builder.
	ch := events.NewChannel[SensorReading](
		"sensors/{sensorID}/readings",
		readingCodec,
		events.Subscribe{OperationID: "receiveSensorReading", Summary: "Receive a reading"},
	)

	handle, err := ch.Register(b)
	if err != nil {
		fmt.Println("register error:", err)
		return
	}

	// Decode a payload.
	reading, err := handle.Decode([]byte(`{"sensor_id":"s1","value":42.5}`))
	if err != nil {
		fmt.Println("decode error:", err)
		return
	}
	fmt.Printf("sensor=%s value=%.1f\n", reading.SensorID, reading.Value)
}

Output:
sensor=s1 value=42.5

func (Channel[T]) ClientHandle added in v0.11.0 

func (c Channel[T]) ClientHandle() *ChannelHandle[T]

ClientHandle returns a ChannelHandle for client-side use without registering with a Builder. No spec registration occurs (SecuritySchemes and GlobalSecurity are left empty).

Use ClientHandle when only the codec and topic definitions are needed (no AsyncAPI spec, no server), or when sharing a Channel definition between publisher and subscriber in the same binary without a builder registration.

The returned handle has the same Decode / Encode helpers and BuildTopic / ValidateTopicVars methods as a handle returned by Channel.Register.

Example — client-only usage (no builder required): 

var SensorChannel = events.NewChannel[SensorReading]("sensors/{sensorID}/data",
    sensorCodec, events.TopicParam{Name: "sensorID"}.WithCodec(sensorIDCodec),
)

handle := SensorChannel.ClientHandle()
domain.SensorReadings.Bind(ctx, mqtt5.SubscribeAdapter(client, router, handle, 0, fmt, opts))

Mirrors [rest.Route.ClientHandle] and [reqreply.Route.ClientHandle].

func (Channel[T]) Register added in v0.8.0 

func (c Channel[T]) Register(b *Builder) (*ChannelHandle[T], error)

Register registers the channel with b and returns a ChannelHandle.

If the builder was created with WithTopicCodec or WithTopicConstraints, the topic is validated immediately and an error is returned if it fails — no channel is registered in that case.

Any TopicParam entry whose name does not appear as a {varName} placeholder in the topic template causes Register to return an error immediately.

type ChannelHandle 

type ChannelHandle[T any] struct {
	// Topic is the channel name (e.g. "user/created", "orders.placed").
	Topic string

	// Descriptor is the live asyncapi.ChannelItem descriptor. It reflects the
	// current configuration and is used by the builder at AsyncAPISpec() time.
	Descriptor asyncapi.ChannelItem

	// Decode deserialises and validates a JSON payload into T.
	// All Refine constraints on the payload codec run automatically.
	Decode func(payload []byte) (T, error)

	// Encode serialises T to JSON bytes.
	Encode func(msg T) ([]byte, error)

	// Formats, when non-empty, specifies the default payload format for both
	// subscribe (decode) and publish (encode). The adapter uses Formats[0] when
	// no call-time format override is provided. Defaults to JSON when empty.
	// Configure via [ChannelHandle.WithFormats].
	// Use [ChannelHandle.WithSubscribeFormats] / [ChannelHandle.WithPublishFormats]
	// for asymmetric channels where decode and encode use different formats.
	Formats []format.Format[T]

	// SubscribeFormats, when non-empty, overrides Formats for the subscribe
	// (receive / decode) direction only. The adapter uses SubscribeFormats[0]
	// instead of Formats[0] when decoding incoming messages.
	// Configure via [ChannelHandle.WithSubscribeFormats].
	SubscribeFormats []format.Format[T]

	// PublishFormats, when non-empty, overrides Formats for the publish
	// (send / encode) direction only. The adapter uses PublishFormats[0]
	// instead of Formats[0] when encoding outgoing messages.
	// Configure via [ChannelHandle.WithPublishFormats].
	PublishFormats []format.Format[T]

	// SecuritySchemes maps scheme name to SecurityScheme (with runtime Codec).
	// Populated from Builder.AddSecurityScheme when AddChannel is called.
	// Adapters use this map to extract and validate credentials per scheme.
	SecuritySchemes map[string]SecurityScheme

	// GlobalSecurity holds the builder-level security requirements that apply
	// when the channel operation's Security field is nil (i.e. the channel
	// inherits global security). Adapters resolve the effective requirements as:
	//   reqs := handle.Descriptor.Subscribe.Security
	//   if reqs == nil { reqs = handle.GlobalSecurity }
	// Set via [Builder.AddGlobalSecurity]. nil when no global security is declared.
	GlobalSecurity []route.SecurityRequirement
	// contains filtered or unexported fields
}

ChannelHandle is returned by Channel.Register. It holds the spec descriptor and codec-backed Decode/Encode helpers.

func (*ChannelHandle[T]) BuildTopic added in v0.4.0 

func (h *ChannelHandle[T]) BuildTopic(vars map[string]string) (string, error)

BuildTopic substitutes {varName} placeholders in the channel's topic template with the values provided in vars, validating each against its registered codec (if any).

All template variables must be present in vars; missing variables return an error. Values are validated before substitution; codec failures return a TopicParamError that identifies the variable name and the failing value. Keys in vars that do not appear in the template are silently ignored.

If the builder was created with WithTopicCodec or WithTopicConstraints, the final assembled topic is also validated against that codec. A failure returns an InvalidTopicError with the concrete topic (not the template).

Example: 

topic, err := sensorChannel.BuildTopic(map[string]string{"sensorID": "f47ac10b-..."})
// topic = "sensors/f47ac10b-.../measurements"

func (*ChannelHandle[T]) DecodeMerged added in v0.12.0 

func (h *ChannelHandle[T]) DecodeMerged(payload []byte, topicVars map[string]string) (T, error)

DecodeMerged decodes payload (via the channel's registered format) AND merges every NewTopicParam-registered topic variable into the SAME T value, using codex.DecodeVars internally — the events-boundary mirror of [rest.RouteHandle.DecodeMerged]. Additive — ChannelHandle.Decode is unchanged; DecodeMerged behaves identically to a bare Decode when the channel declares no merge-capable topic params (MergeFields() is empty).

The payload decode error (if any) is returned FIRST, before the topic-var merge step runs — matching [rest.RouteHandle.DecodeMerged]'s precedent. The merge step itself collects every field's failure via codex.DecodeVars (never stops at the first one).

func (*ChannelHandle[T]) ErrorResponseFor added in v0.12.0 

func (h *ChannelHandle[T]) ErrorResponseFor(err error) (ErrorChannelResponse, bool, error)

ErrorResponseFor returns the first declared ErrorChannel pattern match for err (matching via errors.As, in declaration order), or (ErrorChannelResponse{}, false, nil) when none match.

A non-nil third return value indicates the matched pattern's mapping or encoding failed — callers should treat this as a terminal error for that pattern (do not fall through to other patterns).

func (*ChannelHandle[T]) MergeFields added in v0.12.0 

func (h *ChannelHandle[T]) MergeFields() []codex.FieldCodec[T]

MergeFields returns the merge-capable fields registered via NewTopicParam — feed them directly into codex.DecodeVars/ codex.EncodeVars, or use ChannelHandle.DecodeMerged for the closed-loop convenience method. Unlike REST's role-scoped PathMergeFields/QueryMergeFields/etc., there is only ONE var destination for events (the topic), so this single flat slice is safe for both directions — no cross-role leak risk exists here.

func (*ChannelHandle[T]) ValidateTopic added in v0.8.0 

func (h *ChannelHandle[T]) ValidateTopic(topic string) error

ValidateTopic validates a received concrete topic string against the builder-level topic codec (set via WithTopicCodec or WithTopicConstraints).

Call this after a wildcard subscription delivers a message to verify the concrete topic satisfies the same constraints applied at channel registration time. Returns InvalidTopicError on failure; returns nil if no topic codec is registered.

Note: unlike Channel.Register, which validates a template-stripped topic, this method validates the concrete topic as-is (with real segment values in place).

func (*ChannelHandle[T]) ValidateTopicVars added in v0.8.0 

func (h *ChannelHandle[T]) ValidateTopicVars(vars map[string]string) error

ValidateTopicVars validates extracted topic variable values against the registered TopicParam codecs. Call this after [TopicVarsFromMessage] has extracted the vars map to ensure each variable satisfies its codec constraints.

Returns TopicParamError for the first variable that fails its codec. Variables without a registered codec are skipped.

func (*ChannelHandle[T]) WithFormats added in v0.8.0 

func (h *ChannelHandle[T]) WithFormats(fmts ...format.Format[T]) *ChannelHandle[T]

WithFormats sets the default payload format for this channel. The adapter uses Formats[0] for both subscribe (decode) and publish (encode) when no call-time format override is provided. Defaults to JSON when empty.

WithFormats also updates the live AsyncAPI descriptor: if fmts is non-empty, Message.ContentType on each registered operation (Subscribe/Publish) is set to fmts[0].ContentType(). Calling WithFormats with no arguments clears both Formats and the content-type override (restoring the AsyncAPI default).

This mirrors [rest.RouteHandle.WithFormats] for event channels. Call it after Channel.Register to configure non-JSON payload serialisation: 

ch = ch.WithFormats(format.YAML(measurementCodec))

// Adapter uses YAML automatically — no format arg needed:
client.Subscribe(topic, 1, amqtt.SubscribeHandler(ctx, ch, fn, opts))

func (*ChannelHandle[T]) WithPublishFormats added in v0.8.0 

func (h *ChannelHandle[T]) WithPublishFormats(fmts ...format.Format[T]) *ChannelHandle[T]

WithPublishFormats sets the default payload format for the publish (send / encode) direction only, leaving the subscribe direction unchanged. The adapter uses PublishFormats[0] when encoding outgoing messages. Calling with no arguments clears the publish-specific override (Formats is used).

Use this for asymmetric channels where inbound and outbound payloads use different serialisation formats (e.g. YAML in, JSON out).

func (*ChannelHandle[T]) WithSubscribeFormats added in v0.8.0 

func (h *ChannelHandle[T]) WithSubscribeFormats(fmts ...format.Format[T]) *ChannelHandle[T]

WithSubscribeFormats sets the default payload format for the subscribe (receive / decode) direction only, leaving the publish direction unchanged. The adapter uses SubscribeFormats[0] when decoding incoming messages. Calling with no arguments clears the subscribe-specific override (Formats is used).

Use this for asymmetric channels where inbound and outbound payloads use different serialisation formats (e.g. YAML in, JSON out).

type ChannelMeta added in v0.8.0 

type ChannelMeta struct {
	Title       string
	Summary     string
	Description string
	Tags        []string
}

ChannelMeta holds channel-level metadata for a channel registration: title, summary, description, and tags. All fields are optional. The values flow into the generated AsyncAPI ChannelItem.

ChannelMeta implements ChannelOpt: pass it directly to NewChannel.

type ChannelOpt added in v0.8.0 

type ChannelOpt interface {
	// contains filtered or unexported methods
}

ChannelOpt is the sealed interface for variadic NewChannel options.

The following types implement ChannelOpt:

  • ChannelMeta — channel-level metadata (title, summary, description, tags)
  • Subscribe — subscribe operation metadata (application receives messages)
  • Publish — publish operation metadata (application sends messages)
  • TopicParam — topic template variable with optional codec and description
  • ErrorChannel — per-channel typed error pattern with declared error-output topic

func Formats added in v0.12.0 

func Formats[T any](fmts ...format.Format[T]) ChannelOpt

Formats declares the default payload format for a channel — the ChannelOpt equivalent of calling ChannelHandle.WithFormats after Channel.Register. Declarable inline in NewChannel's variadic opts, which means it also works through ports.EventPattern.Opts with zero changes to the ports package: 

events.NewChannel[Image]("images/{id}", imageCodec,
    events.Formats(format.Binary(pngCodec).WithContentType("image/png")),
)

A mismatched type (fmts holding format.Format[X] where the channel's payload type is not X) is only detectable once T is concrete — Channel.Register returns FormatOptError in that case.

func PublishFormats added in v0.12.0 

func PublishFormats[T any](fmts ...format.Format[T]) ChannelOpt

PublishFormats declares the payload format for the publish (send) direction only — the ChannelOpt equivalent of ChannelHandle.WithPublishFormats. See SubscribeFormats.

func SubscribeFormats added in v0.12.0 

func SubscribeFormats[T any](fmts ...format.Format[T]) ChannelOpt

SubscribeFormats declares the payload format for the subscribe (receive) direction only — the ChannelOpt equivalent of ChannelHandle.WithSubscribeFormats. Use for asymmetric channels (e.g. YAML in, JSON out) alongside PublishFormats.

type ErrorAction added in v0.12.0 

type ErrorAction string

ErrorAction selects how a matched error pattern is realized by the adapter. Events/pub-sub have no synchronous caller to respond to — a matched pattern executes exactly ONE of these, never an implicit handle-then-respond chain: 

const (
	// ErrorRespond publishes the declared typed error payload to the
	// pattern's declared error-output topic. This is the pub/sub analogue
	// of a caller-facing error response: since there is no synchronous
	// caller, "respond" means publish to a dedicated error channel. This is
	// the default action when [ErrorChannel] is declared without an
	// explicit action.
	ErrorRespond ErrorAction = "respond"
	// ErrorHandle performs no automatic publish; the adapter's existing
	// OnError-style callback (already accepted by the adapter's options)
	// runs instead, receiving the original error.
	ErrorHandle ErrorAction = "handle"
	// ErrorLog performs no automatic publish; the adapter forwards the
	// error through its normal observability/error-reporting path only
	// (identical to the unmatched-error fallback).
	ErrorLog ErrorAction = "log"
)

type ErrorChannelOpt added in v0.12.0 

type ErrorChannelOpt[E error, B any] struct {
	// contains filtered or unexported fields
}

ErrorChannelOpt is the ChannelOpt value returned by ErrorChannel.

func ErrorChannel added in v0.12.0 

func ErrorChannel[E error, B any](
	topic string,
	codec codex.Codec[B],
	mapFn ...func(E) (B, error),
) ErrorChannelOpt[E, B]

ErrorChannel declares, for a channel, how a matched subscribe/pipeline error type is realized — the pub/sub analogue of [rest.ErrorPattern], adapted to the fact that pub/sub channels have no synchronous caller to respond to. For the default ErrorRespond action, a codec-backed typed payload is published to topic when an error matching E occurs.

Two modes, mirroring [rest.ErrorPattern]:

  • Direct: no mapFn provided, E must be assignable to B.
  • Mapped: mapFn(E) produces B.

Matching is type-only via errors.As; the first declared ErrorChannel (in NewChannel option order) whose type matches wins — the same deterministic precedence used by REST error patterns. 

events.NewChannel[Reading]("sensors/{id}/data", readingCodec,
    events.ErrorChannel[domain.ValidationError, ErrorPayload](
        "sensors/{id}/errors", errorPayloadCodec,
        func(e domain.ValidationError) (ErrorPayload, error) {
            return ErrorPayload{Code: "validation", Message: e.Error()}, nil
        },
    ),
)

Use ErrorChannelOpt.WithAction to override the default respond action with ErrorHandle (existing adapter OnError callback runs instead, no publish) or ErrorLog (adapter's normal error-forwarding path only, no publish) — the same three-way action model used across go-codex adapters.

func (ErrorChannelOpt[E, B]) WithAction added in v0.12.0 

func (o ErrorChannelOpt[E, B]) WithAction(action ErrorAction) ErrorChannelOpt[E, B]

WithAction returns a copy of o with Action set to action, overriding the default ErrorRespond. A matched pattern executes exactly one action — never an implicit handle-then-respond chain.

type ErrorChannelResponse added in v0.12.0 

type ErrorChannelResponse struct {
	// Topic is the declared error-output topic Body should be published to.
	Topic string
	// Body is the JSON-encoded typed error payload.
	Body []byte
	// Value is the typed payload before encoding — useful for adapters
	// that want to re-encode with a non-JSON [format.Format].
	Value any
	// Action is the resolved action for the matched pattern.
	Action ErrorAction
}

ErrorChannelResponse is the result of a matched ErrorChannel pattern — the typed payload to publish, already encoded, plus the declared error-output topic and resolved action.

type FormatOptError added in v0.12.0 

type FormatOptError struct {
	// Direction is "both" ([Formats]), "subscribe", or "publish".
	Direction string
	Err       error
}

FormatOptError is returned by Channel.Register when Formats, SubscribeFormats, or PublishFormats was declared with formats for a type that does not match the channel's actual payload type parameter.

func (FormatOptError) Error added in v0.12.0 

func (e FormatOptError) Error() string

func (FormatOptError) LogValue added in v0.12.0 

func (e FormatOptError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (FormatOptError) Unwrap added in v0.12.0 

func (e FormatOptError) Unwrap() error

Unwrap allows errors.Is and errors.As to reach the underlying error.

type Info 

type Info = asyncapi.Info

Info is an alias for asyncapi.Info. Using the alias avoids duplicating fields and keeps the two in sync automatically.

type InvalidTopicError added in v0.4.0 

type InvalidTopicError struct {
	Topic string // the topic that failed validation
	Err   error  // the underlying constraint or codec error
}

InvalidTopicError is returned by Channel.Register when the topic fails builder-level topic codec validation.

Use errors.As to extract it and inspect the failing topic or the underlying constraint error: 

var topicErr events.InvalidTopicError
if errors.As(err, &topicErr) {
    log.Printf("bad topic %q: %v", topicErr.Topic, topicErr.Err)
}

func (InvalidTopicError) Error added in v0.4.0 

func (e InvalidTopicError) Error() string

func (InvalidTopicError) Unwrap added in v0.4.0 

func (e InvalidTopicError) Unwrap() error

Unwrap allows errors.As and errors.Is to traverse the underlying constraint error.

type InvalidTopicParamError added in v0.6.0 

type InvalidTopicParamError struct {
	Name  string // the variable name (without braces) that is not in the template
	Topic string // the topic template that was validated against
}

InvalidTopicParamError is returned by Channel.Register when a TopicParam entry names a variable that does not appear in the topic template.

Use errors.As to extract the offending name and the topic template: 

var paramErr events.InvalidTopicParamError
if errors.As(err, &paramErr) {
    log.Printf("TopicParam %q not in topic %q", paramErr.Name, paramErr.Topic)
}

func (InvalidTopicParamError) Error added in v0.6.0 

func (e InvalidTopicParamError) Error() string

type MergeFieldTypeError added in v0.12.0 

type MergeFieldTypeError struct {
	Err error
}

MergeFieldTypeError is returned by Channel.Register when a merge field registered via NewTopicParam has the wrong type parameter for the channel's payload type — mirrors [rest.MergeFieldTypeError] exactly.

func (MergeFieldTypeError) Error added in v0.12.0 

func (e MergeFieldTypeError) Error() string

func (MergeFieldTypeError) LogValue added in v0.12.0 

func (e MergeFieldTypeError) LogValue() slog.Value

LogValue implements slog.LogValuer for structured logging.

func (MergeFieldTypeError) Unwrap added in v0.12.0 

func (e MergeFieldTypeError) Unwrap() error

Unwrap allows errors.Is and errors.As to reach the underlying error.

type MergedTopicParam added in v0.12.0 

type MergedTopicParam[T any] struct {
	TopicParam
	// contains filtered or unexported fields
}

MergedTopicParam is returned by NewTopicParam. It is the events-boundary mirror of [rest.MergedPathParam]: on subscribe, the registered field's setter merges the extracted topic variable into the decoded payload via ChannelHandle.DecodeMerged; on publish, the field's getter extracts the topic variable's value from the payload via ChannelHandle.MergeFields + codex.EncodeVars (or [PublishHandle] in adapters/mqtt5, which does this automatically).

func NewTopicParam added in v0.12.0 

func NewTopicParam[T any](
	name string,
	codec codex.Codec[string],
	get func(T) string,
	set func(*T, string),
) MergedTopicParam[T]

NewTopicParam declares a topic variable that is BOTH validated against codec AND automatically merged into T by ChannelHandle.DecodeMerged — one declaration instead of a TopicParam plus a separate codex.Field. All topic variables are always required (a template cannot be resolved without every {varName} placeholder present), matching plain TopicParam's existing "no Required field" rationale. 

events.NewChannel[SensorReading]("sensors/{sensorID}/readings", sensorReadingCodec,
    events.NewTopicParam("sensorID", codex.String().Refine(validate.UUID),
        func(r SensorReading) string { return r.SensorID },
        func(r *SensorReading, v string) { r.SensorID = v },
    ),
)

func (MergedTopicParam[T]) WithDescription added in v0.12.0 

func (p MergedTopicParam[T]) WithDescription(desc string) MergedTopicParam[T]

WithDescription sets the PARAMETER-level description and returns the updated value.

type MissingTopicVarError added in v0.4.0 

type MissingTopicVarError struct {
	Name string // the variable name (without braces) that had no value
}

MissingTopicVarError is returned by ChannelHandle.BuildTopic when a {varName} placeholder in the topic template has no corresponding entry in the vars map.

Use errors.As to extract the missing variable name: 

var missingErr events.MissingTopicVarError
if errors.As(err, &missingErr) {
    log.Printf("caller forgot to supply topic variable {%s}", missingErr.Name)
}

func (MissingTopicVarError) Error added in v0.4.0 

func (e MissingTopicVarError) Error() string

type Publish added in v0.8.0 

type Publish struct {
	// OperationID is the unique identifier for the publish operation in the
	// AsyncAPI spec. Used by code generators and documentation tools.
	OperationID string
	Summary     string
	Description string
	Tags        []string

	// SchemaName, when non-empty, emits a $ref for the payload schema in the
	// spec and registers the schema under that name in components/schemas.
	SchemaName string

	// Security, when non-nil, overrides global security for this operation.
	// Pass an empty slice to declare "no auth required" for this publish operation.
	// nil (default) inherits global security declared via [Builder.AddGlobalSecurity].
	Security []route.SecurityRequirement
}

Publish describes the publish operation on a channel (application sends). It controls the publish entry in the AsyncAPI spec.

Publish implements ChannelOpt: pass it directly to NewChannel.

type SecurityScheme added in v0.8.0 

type SecurityScheme struct {
	route.SecurityScheme
	// Codec, when non-nil, validates the extracted raw credential string.
	// Nil means no format validation; SecurityFunc receives the message as-is.
	Codec *codex.Codec[string]
}

SecurityScheme combines route.SecurityScheme spec metadata with optional runtime credential validation for message broker adapters.

AddSecurityScheme registers it with the builder. The spec fields flow into the AsyncAPI document; Codec, when non-nil, is used by adapters to validate the raw credential string before SecurityFunc is called.

MQTT note: paho.mqtt.golang (MQTT 3.1.1) does not expose per-message credentials. Codec-level extraction is a no-op for standard MQTT. Use SecurityFunc + closure (credentials passed at MQTT CONNECT time) for runtime enforcement.

Use SecurityScheme.WithCodec to set the Codec field inline without a temporary variable: events.SecurityScheme{SecurityScheme: route.APIKeyScheme(...)}.WithCodec(c)

func (SecurityScheme) WithCodec added in v0.8.0 

func (s SecurityScheme) WithCodec(c codex.Codec[string]) SecurityScheme

WithCodec returns a copy of s with Codec set to c. It avoids the temporary-variable + address-of pattern required when setting Codec inline: 

b.AddSecurityScheme("apiKey", events.SecurityScheme{
    SecurityScheme: route.APIKeyScheme("X-API-Key", "header"),
}.WithCodec(codex.String().Refine(validate.NonEmptyString)))

type Server 

type Server = asyncapi.Server

Server is an alias for asyncapi.Server.

type Subscribe added in v0.8.0 

type Subscribe struct {
	// OperationID is the unique identifier for the subscribe operation in the
	// AsyncAPI spec. Used by code generators and documentation tools.
	OperationID string
	Summary     string
	Description string
	Tags        []string

	// SchemaName, when non-empty, emits a $ref for the payload schema in the
	// spec and registers the schema under that name in components/schemas.
	SchemaName string

	// Security, when non-nil, overrides global security for this operation.
	// Pass an empty slice to declare "no auth required" for this subscription.
	// nil (default) inherits global security declared via [Builder.AddGlobalSecurity].
	Security []route.SecurityRequirement
}

Subscribe describes the subscribe operation on a channel (application receives). It controls the subscribe entry in the AsyncAPI spec.

Subscribe implements ChannelOpt: pass it directly to NewChannel.

type TopicParam added in v0.5.0 

type TopicParam struct {
	// Name is the variable name (without braces) as it appears in the topic template.
	Name string
	// Description is shown in the AsyncAPI spec for this parameter.
	Description string
	// Codec validates topic parameter values at [ChannelHandle.ValidateTopicVars] and
	// [ChannelHandle.BuildTopic] time.
	// When non-nil, the codec's schema is also emitted in the AsyncAPI spec.
	// Nil means no runtime validation; the spec defaults to {type: string}.
	Codec *codex.Codec[string]
}

TopicParam describes a {varName} placeholder in a topic template for AsyncAPI spec generation and runtime validation.

TopicParam is the single configuration point for a topic variable: it carries spec metadata (description) and an optional codec for runtime validation. The codec schema is also used to enrich the AsyncAPI parameters: block.

TopicParam is optional: the events builder auto-derives parameters from the topic template. Use TopicParam to add a description or register a codec for a specific variable.

Note: all topic variables are always required — a template cannot be resolved without every {varName} placeholder present. There is no Required field.

TopicParam implements ChannelOpt: pass it directly to NewChannel.

Entry names must correspond to {varName} placeholders in the topic template; unknown names cause Channel.Register to return an error immediately.

func (TopicParam) WithCodec added in v0.8.0 

func (p TopicParam) WithCodec(c codex.Codec[string]) TopicParam

WithCodec sets the validation codec and returns the updated TopicParam.

type TopicParamError added in v0.4.0 

type TopicParamError struct {
	Name  string // the {varName} that failed
	Value string // the value that was rejected
	Err   error  // the underlying codec error
}

TopicParamError is returned by ChannelHandle.BuildTopic and ChannelHandle.ValidateTopicVars when a topic variable fails its registered codec check.

Use errors.As to extract it and inspect the failing variable: 

var paramErr events.TopicParamError
if errors.As(err, &paramErr) {
    log.Printf("bad value %q for {%s}: %v", paramErr.Value, paramErr.Name, paramErr.Err)
}

func (TopicParamError) Error added in v0.4.0 

func (e TopicParamError) Error() string

func (TopicParamError) Unwrap added in v0.4.0 

func (e TopicParamError) Unwrap() error

Unwrap allows errors.As and errors.Is to traverse the underlying codec error.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.