Documentation
¶
Overview ¶
Package events provides a transport-agnostic event channel builder for go-codex.
Define channels with codec-backed payload types; the builder returns 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 2.6 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",
})
userCreated := events.AddChannel[UserCreated](b, "user/created", userCreatedCodec,
events.ChannelConfig{
Subscribe: &events.OperationConfig{
Summary: "A user was created",
SchemaName: "UserCreatedEvent",
},
})
// In your broker callback (any library):
event, err := userCreated.Decode(msg.Payload()) // JSON → UserCreated, validates
payload, err := userCreated.Encode(event) // UserCreated → JSON
// AsyncAPI 2.6 spec:
doc, err := b.AsyncAPISpec()
yaml, _ := doc.MarshalYAML()
Encoding is JSON only. AddChannel uses format.JSON internally; for other formats construct a format.Format directly and call its Unmarshal/Marshal.
Index ¶
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) AddSchema ¶
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) AsyncAPISpec ¶
AsyncAPISpec builds a complete AsyncAPI 2.6 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 AddChannel. If the topic is invalid, AddChannel 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 ChannelConfig ¶
type ChannelConfig struct {
Description string
// Subscribe describes the operation where the application receives messages.
// Set to nil to omit the subscribe operation from the spec.
Subscribe *OperationConfig
// Publish describes the operation where the application sends messages.
// Set to nil to omit the publish operation from the spec.
Publish *OperationConfig
// TopicParamCodecs maps {varName} template variables in the topic to codecs
// that validate concrete values at runtime via [ChannelHandle.BuildTopic].
//
// Keys must correspond to {varName} placeholders in the topic template;
// unknown keys cause [AddChannel] to return an error immediately.
TopicParamCodecs map[string]codex.Codec[string]
}
ChannelConfig holds metadata for a channel registration.
At least one of Subscribe or Publish must be non-nil. When both are set, the same payload codec is used for both directions.
type ChannelHandle ¶
type ChannelHandle[T any] struct { // Topic is the channel name (e.g. "user/created", "orders.placed"). Topic string // Descriptor is the frozen asyncapi.ChannelItem built at registration 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) // contains filtered or unexported fields }
ChannelHandle is returned by AddChannel. It holds the frozen spec descriptor and codec-backed Decode/Encode helpers.
func AddChannel ¶
func AddChannel[T any]( b *Builder, topic string, codec codex.Codec[T], config ChannelConfig, ) (*ChannelHandle[T], error)
AddChannel registers a channel with the builder and returns a ChannelHandle.
codec is used to decode and validate incoming payloads and to encode outgoing messages. The same codec applies to both subscribe and publish directions.
If the builder was created with WithTopicCodec or WithTopicConstraints, the topic is validated immediately. An error is returned if validation fails — no channel is registered in that case.
If config.TopicParamCodecs is non-empty, each key is verified to be a {varName} present in the topic template. A key that does not appear in the topic is a programming error and causes AddChannel to return an error.
AddChannel is a free function (not a method) because Go requires type parameters to appear on free functions, not on method receivers.
The descriptor is built and frozen at call time; later mutations to config do not affect the registered channel or the returned handle.
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.
Example:
topic, err := sensorChannel.BuildTopic(map[string]string{"sensorID": "f47ac10b-..."})
// topic = "sensors/f47ac10b-.../measurements"
type 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 AddChannel 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 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 OperationConfig ¶
type OperationConfig struct {
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
}
OperationConfig holds metadata for one direction (subscribe or publish) on a channel. It controls the operation entry in the AsyncAPI spec.
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 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, ¶mErr) {
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