mqtt

Documentation

Overview

Package mqtt adapts api/events channel handles to [Paho MQTT] callbacks.

SubscribeHandler turns a events.ChannelHandle into an mqtt.MessageHandler that decodes and validates incoming payloads before calling the application handler. Publish encodes a value and publishes it to the broker.

Typical usage:

b := events.NewBuilder(events.Info{Title: "My Events", Version: "1.0.0"})
userCreated := events.AddChannel[UserCreated](b, "user/created", codec,
    events.ChannelConfig{Subscribe: &events.OperationConfig{...}})

// Wire to Paho on connect:
client.Subscribe(userCreated.Topic, 1,
    mqtt.SubscribeHandler(ctx, userCreated, func(ctx context.Context, e UserCreated) error {
        return svc.HandleUserCreated(ctx, e)
    }, func(e mqtt.SubscribeError) { log.Println("event error:", e) }),
)

// Publish an event:
notification := NotificationCommand{Recipient: "alice@example.com", ...}
mqtt.Publish(ctx, client, notifChannel, 1, false, notification)

Index

• func Publish[T any](ctx context.Context, client pahomqtt.Client, handle *events.ChannelHandle[T], ...) error
• func SubscribeHandler[T any](ctx context.Context, handle *events.ChannelHandle[T], ...) pahomqtt.MessageHandler
• type ErrorKind
  • func (k ErrorKind) String() string
• type SubscribeError
  • func (e SubscribeError) Error() string
  • func (e SubscribeError) Unwrap() error

Functions

func Publish

func Publish[T any](ctx context.Context, client pahomqtt.Client, handle *events.ChannelHandle[T], qos byte, retained bool, msg T, vars map[string]string) error

Publish encodes msg using handle's codec and publishes it to the broker.

vars controls the topic used for publishing:

• nil: publish to handle.Topic directly (use for static topics).
• non-nil: call handle.BuildTopic(vars) to build a concrete topic from the template, validating each variable against its registered codec. An error is returned if any variable is missing or fails validation.

Example — static topic:

err := adaptermqtt.Publish(ctx, client, notifChannel, 1, false, notification, nil)

Example — template topic (sensors/{sensorID}/alerts):

err := adaptermqtt.Publish(ctx, client, alertChannel, 1, false, alert,
    map[string]string{"sensorID": id})

Publish waits for broker acknowledgement, respecting ctx cancellation. If the context is cancelled before the broker responds, ctx.Err() is returned.

func SubscribeHandler

func SubscribeHandler[T any](
	ctx context.Context,
	handle *events.ChannelHandle[T],
	fn func(context.Context, T) error,
	onErr func(SubscribeError),
) pahomqtt.MessageHandler

SubscribeHandler returns a pahomqtt.MessageHandler that decodes the message payload using handle's codec, validates it, and calls fn.

ctx is threaded through to fn for cancellation and deadline propagation. If onErr is non-nil it is called with a typed SubscribeError containing the error kind, topic, and underlying error. If onErr is nil errors are silently discarded.

The Topic field of SubscribeError reflects the concrete topic of the incoming message (from msg.Topic()), which is useful when the channel was registered with a template topic.

Types

type ErrorKind

type ErrorKind int

ErrorKind classifies the origin of a SubscribeError.

const (
	// KindDecode indicates the message payload could not be decoded or
	// failed codec validation.
	KindDecode ErrorKind = iota

	// KindHandler indicates the application handler returned an error after
	// successful decoding.
	KindHandler
)

func (ErrorKind) String

func (k ErrorKind) String() string

type SubscribeError

type SubscribeError struct {
	Kind  ErrorKind
	Topic string
	Err   error
}

SubscribeError is returned to the onErr callback with a typed Kind so callers can distinguish decode/validation failures from application handler errors without string matching.

func (SubscribeError) Error

func (e SubscribeError) Error() string

func (SubscribeError) Unwrap

func (e SubscribeError) Unwrap() error