MQTT 5 Examples¶

import (
    "net"
    "github.com/eclipse/paho.golang/paho"
)

conn, err := net.Dial("tcp", "localhost:1883")
router := paho.NewStandardRouter()
client := paho.NewClient(paho.ClientConfig{
    Conn:   conn,
    Router: router,
    OnClientError:      func(err error) { log.Error("client error", "err", err) },
    OnServerDisconnect: func(d *paho.Disconnect) { log.Warn("disconnected") },
})

// Connect
if _, err := client.Connect(ctx, &paho.Connect{
    KeepAlive:  60,
    ClientID:   "my-service",
    CleanStart: true,
}); err != nil {
    log.Fatal(err)
}

import (
    "context"

    mqtt5adapter "github.com/DaniDeer/go-codex/adapters/mqtt5"
    "github.com/DaniDeer/go-codex/api/events"
    "github.com/DaniDeer/go-codex/stats"
)

// NewSubscribeTransport/NewPublishTransport — the spec-free, no-*Client-needed,
// handle-based call surface Decision 7 inverted into api/events itself
// (docs/design/d-0002-pubsub-workflow-simplification.md). Fully typed generic
// constructors, no reflection; use these for custom OnError/Observer/security
// impls or wildcard topics. The simple case uses Client.Attach/.Subscribe below
// instead.
subTransport := mqtt5adapter.NewSubscribeTransport[SensorReading](client, router,
    mqtt5adapter.SubscribeOptions{Observer: obs})

sub := contract.ReadingsChannel.WithSubscribe(events.Subscribe{})
if err := events.SubscribeHandle(ctx, sub, subTransport,
    func(ctx context.Context, r SensorReading) error {
        return store.Save(ctx, r)
    },
); err != nil {
    log.Fatal(err)
}

// Publish
pubTransport := mqtt5adapter.NewPublishTransport[SensorReading](client,
    mqtt5adapter.PublishOptions[SensorReading]{
        Observer:    obs,
        ContentType: "application/json", // sets MQTT 5 ContentType property
        UserProperties: []mqtt5adapter.UserProperty{
            {Key: "TenantID", Value: "acme"},
        },
    },
)
pub := contract.ReadingsChannel.WithPublish(events.Publish{})
err := events.PublishHandle(ctx, pub, pubTransport, reading)

client := events.NewClient(events.WithInfo(events.Info{Title: "Sensor Network", Version: "1.0.0"}))
if err := client.Attach(mqtt5.NewTransport(mqtt5.TransportOptions{Client: mqttClient, Router: router})); err != nil {
    log.Fatal(err)
}

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

go func() {
    _ = client.Subscribe(ctx, sub, func(ctx context.Context, r SensorReading) error {
        log.Printf("received: %+v", r)
        return nil
    })
}()

err := client.Publish(ctx, pub, reading) // "one struct, one call"

eventsClient := events.NewClient(events.WithInfo(events.Info{Title: "Sensor Network", Version: "1.0.0"}))
eventsClient.AddServer("mqtt5", events.Server{URL: "mqtt://broker:1883", Protocol: "mqtt5"})
handle, _ := ReadingsChannel.WithSubscribe(events.Subscribe{}).Handle(eventsClient)
spec, _ := eventsClient.AsyncAPISpec()

// Static topic — no template variables.
var ComputeRoute = reqreply.NewRoute[ComputeReq, ComputeResp](
    "compute/add",
    computeReqCodec, computeRespCodec,
    reqreply.RouteMeta{OperationID: "computeAdd"},
)

// Template topic — {tenantID} is validated AND auto-merged into/from
// ComputeReq.TenantID via NewTopicParam (assumes ComputeReq has a
// TenantID string field) — the client derives the topic from the
// request struct automatically; the server receives it already merged.
var TenantComputeRoute = reqreply.NewRoute[ComputeReq, ComputeResp](
    "compute/{tenantID}/add",
    computeReqCodec, computeRespCodec,
    reqreply.RouteMeta{OperationID: "computeAdd"},
    reqreply.NewTopicParam("tenantID", codex.String().Refine(validate.NonEmptyString),
        func(r ComputeReq) string { return r.TenantID },
        func(r *ComputeReq, v string) { r.TenantID = v },
    ),
)

// Server side: WithHandler + Register is ONE fluent chain — the route is
// dispatchable the moment it's registered, no separate Handle step.
server := reqreply.NewServer(reqreply.Info{Title: "Compute API", Version: "1.0.0"})
server.AddServer("mqtt5", reqreply.ServerEntry{URL: "mqtt://broker:1883", Protocol: "mqtt5"})
handle, err := ComputeRoute.WithHandler(func(ctx context.Context, req ComputeReq) (ComputeResp, error) {
    return ComputeResp{Sum: req.X + req.Y}, nil
}).Register(server)
if err != nil {
    log.Fatal(err)
}
if err := server.Attach(mqtt5adapter.NewServerTransport(mqtt5adapter.ServerTransportOptions{Client: client, Router: router})); err != nil {
    log.Fatal(err)
}
go server.Serve(ctx) // dispatches every registered route concurrently, blocks until ctx is cancelled

// Client side: Attach once, then plain Call/CallAsync — no *RouteHandle needed
// for a raw Route call (GlobalSecurity is invisible in that mode, same
// accepted limitation as REST's own Route.ClientHandle()).
reqreplyClient := reqreply.NewClient()
if err := reqreplyClient.Attach(mqtt5adapter.NewClientTransport(mqtt5adapter.ClientTransportOptions{Client: client, Router: router})); err != nil {
    log.Fatal(err)
}
respAny, err := reqreplyClient.Call(ctx, ComputeRoute, ComputeReq{X: 3, Y: 4})
resp := respAny.(ComputeResp)

// Async: CallAsync returns a *reqreply.Future[ComputeResp] immediately;
// resolve it later, from a different call site if needed.
futureAny, err := reqreplyClient.CallAsync(ctx, ComputeRoute, ComputeReq{X: 10, Y: 20})
future := futureAny.(*reqreply.Future[ComputeResp])
// ... do other independent work ...
resp2, err := future.Wait(ctx)

transport := mqtt5adapter.NewServerTransport(mqtt5adapter.ServerTransportOptions{
    Client: client, Router: router,
    Serve: mqtt5adapter.ServeOptions{Observer: obs},
})
if err := server.Attach(transport); err != nil {
    log.Fatal(err)
}
go server.Serve(ctx)

transport := mqtt5adapter.NewClientTransport(mqtt5adapter.ClientTransportOptions{
    Client: client, Router: router,
    Call: mqtt5adapter.CallOptions{
        ReplyTopicPrefix: "replies",    // generates: "replies/<uuid>"
        Timeout:          5 * time.Second,
        Observer:         obs,
    },
})
if err := reqreplyClient.Attach(transport); err != nil {
    log.Fatal(err)
}

// Static topic — no template vars involved.
respAny, err := reqreplyClient.Call(ctx, ComputeRoute, ComputeReq{X: 3, Y: 4})
if err != nil {
    var reqErr mqtt5adapter.CallError
    if errors.As(err, &reqErr) && reqErr.Kind == mqtt5adapter.KindTimeout {
        log.Warn("request timed out")
    }
}
resp := respAny.(ComputeResp)

// Template topic — TenantComputeRoute's NewTopicParam merge field derives
// the topic from the request struct automatically; no separate vars
// argument needed. Setting req.TenantID is enough:
respAny, err = reqreplyClient.Call(ctx, TenantComputeRoute, ComputeReq{X: 3, Y: 4, TenantID: "acme"})
// On validation failure: CallError wrapping reqreply.RouteParamError
// or reqreply.MissingRouteParamError — both errors.As-navigable.

// Built-in default — explicit form (identical to not setting ReplyTopicBuilder)
transport := mqtt5adapter.NewClientTransport(mqtt5adapter.ClientTransportOptions{
    Client: client, Router: router,
    Call: mqtt5adapter.CallOptions{
        ReplyTopicBuilder: mqtt5adapter.UUIDReplyTopic("replies"),
    },
})

// Shared subscription — scale reply consumers horizontally.
// The ResponseTopic sent to the responder is "replies/<uuid>" (plain publish topic).
// The local subscribe uses "$share/gateway-pool/replies/<uuid>".
// The broker delivers each reply to exactly one subscriber in the group.
transport = mqtt5adapter.NewClientTransport(mqtt5adapter.ClientTransportOptions{
    Client: client, Router: router,
    Call: mqtt5adapter.CallOptions{
        ReplyTopicBuilder: mqtt5adapter.SharedReplyTopic("replies", "gateway-pool"),
    },
})

// Fully custom builder — client-ID + monotonic counter, no uuid dependency.
var seq int64
transport = mqtt5adapter.NewClientTransport(mqtt5adapter.ClientTransportOptions{
    Client: client, Router: router,
    Call: mqtt5adapter.CallOptions{
        ReplyTopicBuilder: func() (string, string) {
            t := fmt.Sprintf("replies/gw-1/%d", atomic.AddInt64(&seq, 1))
            return t, t
        },
    },
})

server := reqreply.NewServer(reqreply.Info{Title: "Compute API", Version: "1.0.0"})
server.AddServer("mqtt5", reqreply.ServerEntry{URL: "mqtt://broker:1883", Protocol: "mqtt5"})
handle, _ := ComputeRoute.Register(server)

doc, _ := server.AsyncAPISpec()  // AsyncAPI 3.0 with reply: block

scheme := route.SecurityScheme{Type: "http", Scheme: "bearer"}
sub := contract.ReadingsChannel.WithSubscribe(events.Subscribe{}).
    SubscribeMW(events.SecurityMiddleware[struct{}, struct{}]("bearerAuth", scheme, nil),
        func(ctx context.Context, msg *paho.Publish, r *SensorReading) (map[string][]string, error) {
            for _, p := range msg.Properties.User {
                if p.Key == "Authorization" {
                    return verifyJWT(strings.TrimPrefix(p.Value, "Bearer "), []route.SecurityRequirement{{"bearerAuth": nil}})
                }
            }
            return nil, errors.New("missing Authorization User Property")
        })

transport := mqtt5adapter.NewSubscribeTransport[SensorReading](client, router,
    mqtt5adapter.SubscribeOptions{})
err := events.SubscribeHandle(ctx, sub, transport, fn)

// Access User Properties inside the handler:
func(ctx context.Context, r SensorReading) error {
    props, ok := mqtt5adapter.UserPropertiesFromContext(ctx)
    if ok {
        tenantID := ""
        for _, p := range props {
            if p.Key == "TenantID" {
                tenantID = p.Value
            }
        }
    }
    return nil
}

transport := mqtt5adapter.NewSubscribeTransport[SensorReading](client, router,
    mqtt5adapter.SubscribeOptions{
        UserPropertyParams: []mqtt5adapter.UserPropertyParam{
            // Required bearer token — validated with a codec:
            mqtt5adapter.UserPropertyParam{Name: "Authorization", Required: true}.
                WithCodec(codex.String().Refine(validate.BearerToken)),
            // Optional tenant ID — present must be non-empty:
            mqtt5adapter.UserPropertyParam{Name: "TenantID", Required: false}.
                WithCodec(codex.String().Refine(validate.NonEmptyString)),
        },
    })
err := events.SubscribeHandle(ctx, sub, transport, fn)

// Error handling:
opts.OnError = func(e mqtt5adapter.SubscribeError) {
    var missing mqtt5adapter.MissingUserPropertyError
    if errors.As(e, &missing) {
        slog.Warn("required property absent", "name", missing.Name)
        return
    }
    var propErr mqtt5adapter.UserPropertyError
    if errors.As(e, &propErr) {
        slog.Warn("property validation failed", "error", propErr)
        return
    }
}

transport := mqtt5adapter.NewSubscribeTransport[SensorReading](client, router,
    mqtt5adapter.SubscribeOptions{},
    format.JSON(sensorCodec),   // ContentType: "application/json"
    format.YAML(sensorCodec),   // ContentType: "application/yaml"
)
err := events.SubscribeHandle(ctx, sub, transport, fn)

obs := stats.NewFanout(
    metricsObserver,
    stats.NewLoggingObserver(slog.Default()),
    tracer,
)

subTransport := mqtt5adapter.NewSubscribeTransport[SensorReading](client, router, mqtt5adapter.SubscribeOptions{Observer: obs})
err := events.SubscribeHandle(ctx, sub, subTransport, fn)

pubTransport := mqtt5adapter.NewPublishTransport[SensorReading](client, mqtt5adapter.PublishOptions[SensorReading]{Observer: obs})
err = events.PublishHandle(ctx, pub, pubTransport, msg)

mqtt5adapter.Serve(ctx, client, router, handle, fn, mqtt5adapter.ServeOptions{Observer: obs})
mqtt5adapter.Call(ctx, client, router, handle, req, mqtt5adapter.CallOptions{Observer: obs})

// Subscribe / Serve — delivered to OnError callback
var subErr mqtt5.SubscribeError
if errors.As(err, &subErr) {
    switch subErr.Kind {
    case mqtt5.KindDecode:    // payload validation failed
    case mqtt5.KindHandler:   // application handler error
    case mqtt5.KindSecurity:  // SecurityFunc rejected the message
    }
    slog.Warn("subscribe failed", "error", subErr) // emits kind, topic, err
}

// Call — returned directly
var reqErr mqtt5.CallError
if errors.As(err, &reqErr) {
    switch reqErr.Kind {
    case mqtt5.KindTimeout:   // no reply within deadline
    case mqtt5.KindDecode:    // reply could not be decoded
    case mqtt5.KindHandler:   // server returned an error
    case mqtt5.KindEncode:    // request encoding failed or subscribe failed
    }
    slog.Error("request failed", "error", reqErr)
}

// Publish — returned directly
var encErr mqtt5.PublishEncodeError
if errors.As(err, &encErr) {
    slog.Error("publish encode failed", "error", encErr) // emits topic, err
}