# πŸ“‘ Section 4 β€” Publish, subscribe, and build a Go device simulator Prerequisites: Sections 1–3 complete. Keep Section 3's configuration package and TOML file. This lesson supplies complete files for a reusable MQTT transport, backend, and simulator. Implement each file in stages and explain it before running both programs. **Version baseline:** Paho `v0.23.0`, verified from upstream release tags on 2026-10-07. It requires Go 1.24 or newer; Go 1.25 is compatible. The matching local reference is `references/paho.golang-v0.23.0/` (commit `1aa03969c5932464359b3e4ed752d9ac61abccc9`). ## 🎬 Start with the system story > πŸ’‘ **Story question:** The backend and device are separate MQTT clients. How do they exchange a command without knowing each other's address, and how can the backend tell whether the device actually handled it? The broker is the meeting point. The backend connects and subscribes to telemetry and responses; the simulator connects with its own identity and subscribes to commands. Each publication goes to the broker, which routes it to matching subscriptions. A device response is a new application message, separate from the broker's acknowledgement of the backend's publish. ```mermaid sequenceDiagram participant G as Go backend participant B as Mosquitto broker participant D as Go device simulator G->>B: CONNECT B-->>G: CONNACK accepted D->>B: CONNECT B-->>D: CONNACK accepted G->>B: SUBSCRIBE telemetry, responses B-->>G: SUBACK D->>B: SUBSCRIBE commands B-->>D: SUBACK G->>B: PUBLISH ping, QoS 1 par Backend-to-broker acknowledgement B-->>G: PUBACK and Broker-to-device delivery B->>D: PUBLISH ping, QoS 1 D->>D: Paho callback queues message D-->>B: PUBACK after callback returns end D->>D: Application handles ping D->>B: PUBLISH pong response, QoS 1 par Device-to-broker acknowledgement B-->>D: PUBACK and Broker-to-backend delivery B->>G: PUBLISH pong response, QoS 1 G->>G: Paho callback queues message G-->>B: PUBACK after callback returns end ``` The broker, backend, and device have distinct setup. Mosquitto owns its listener, password source, and ACL. Each Go process loads its own broker URL, client ID, and username. The backend and simulator use different usernames for their broker permissions; both use the disposable test password from `configs/lab.toml`. The code below first builds the shared transport that connects, subscribes, receives, and publishes; then the two application loops use it. At the end of this lesson, ping/pong proves the transport path works, but the backend still treats received bytes as demonstrations. Section 5 adds a shared contract; Section 5A turns a validated heartbeat into a device-view update. Keep this order clear: a network callback receives packets, a contract gives them shape, and an application service gives them meaning. Before each implementation part, use the question there to predict which process or protocol layer is responsible. Follow [Lesson 0](00-system-lifecycle.md) for the full lifecycle map. ## πŸ—ΊοΈ 1. Map the conversation > πŸ’‘ **Story question:** Which client publishes each kind of message, which client needs it, and what does the broker use to route it? Map the telemetry, command, and response paths first. This identifies which process must subscribe before we write transport code. The lab has one deliberately small hierarchy: `lab` names the isolated environment, `v1` is the topic-contract version, `devices` is the entity collection, `{deviceID}` scopes device traffic, and the last level names a flow. `v1` versions the route vocabulary; the payload's `schema_version` versions JSON shape, so the two versions can evolve independently. A `TopicName` is one concrete publish destination; a `TopicFilter` is a subscription pattern. The broker accepts `+` and `#` only in filters, so the Go API keeps these concepts as different types. This hierarchy is the learning contract, not a proposed migration of the existing vending backend. | Topic | Publisher | Subscriber | Example | | --- | --- | --- | --- | | `lab/v1/devices/device-001/telemetry` | Simulator | Backend | Timestamped heartbeat | | `lab/v1/devices/device-001/commands` | Backend | Simulator | Harmless ping with ID | | `lab/v1/devices/device-001/responses` | Simulator | Backend | Pong echoing that ID | A broker subscription and a local Go callback are different things. SUBSCRIBE asks the broker which messages to forward. `OnPublishReceived` tells your local client how to process forwarded PUBLISH packets. You need both. This lab's topic convention is deliberately separate from the production contracts. ### πŸ”„ Trace two independent delivery hops ```mermaid sequenceDiagram participant B as Go backend participant M as Broker participant S as Go simulator S->>M: SUBSCRIBE commands, maximum QoS 1 M-->>S: SUBACK with grant or rejection B->>M: PUBLISH command, QoS 1 par Backend publish exchange M-->>B: PUBACK and Broker forwards to simulator M->>S: PUBLISH command, granted delivery QoS S-->>M: PUBACK for delivery hop end S->>S: Application handles command S->>M: PUBLISH application result, QoS 1 par Simulator publish exchange M-->>S: PUBACK and Broker forwards result to backend M->>B: PUBLISH application result B-->>M: PUBACK for delivery hop end ``` The `par` blocks show that a publisher-side PUBACK and broker-to-subscriber delivery are independent work on separate client connections; their observed order can vary. Each QoS 1 PUBACK confirms only its own hop. The application result is a separate publicationβ€”the broker does not create it automatically, and neither PUBACK proves that business work completed. ### πŸ“¨ Choose QoS by the guarantee you need | QoS | Successful exchange on one hop | What the Go application must still handle | | --- | --- | --- | | 0 | PUBLISH, no MQTT acknowledgement | Loss is possible; nil publish error does not prove receipt | | 1 | PUBLISH β†’ PUBACK | Redelivery is possible; use stable business IDs | | 2 | PUBLISH β†’ PUBREC β†’ PUBREL β†’ PUBCOMP | Protocol duplicate suppression needs session state; application retries and external effects still need idempotency | The subscriber requests a **maximum** delivery QoS. A QoS 1 publication cannot become QoS 2 merely because the subscription requests 2. With a QoS 0 subscription grant, a publisher's QoS 1 exchange with the broker still ends in PUBACK while the broker-to-subscriber hop has no PUBACK. QoS 2 does not make a database update and a physical dispense one transaction. For a QoS 2 experiment, change both publication and requested subscription QoS to 2, accept SUBACK grants 0–2, and inspect the four-packet exchange. The original helper intentionally accepts only grants up to its requested QoS 1. Do not change only the publish field and call the resulting lower-QoS subscriber path an end-to-end QoS 2 test. ### 🏷️ Topic filters are routing rules `lab/v1/devices/+/telemetry` matches one device segment; `lab/v1/devices/#` matches that subtree. `+` occupies an entire level and `#` is the final level. Topics are case-sensitive, so `device-001` and `Device-001` differ. Wildcards belong in subscriptions, never publish topic names. A root `#` filter does not include `$SYS/...`; broker system topics require an explicit matching filter and permission. A SUBACK proves the subscription request received a result. Topic ACLs still govern delivery: an accepted broad filter does not prove access to every matching topic. Check an authorized observer, reason codes, and broker logs when testing denial. ## πŸ› οΈ 2. Create the transport helper > πŸ’‘ **Story question:** What should be shared between the backend and simulator, and what must remain specific to each running client? The transport helper can share connection lifecycle, subscription handling, message copying, and publishing. Each caller still supplies its configured username, Client ID, topic list, and receive queue. Keeping those identities separate lets the broker apply different permissions and prevents Client ID takeover. Topic construction belongs in `internal/contracts/topic.go`, which is introduced here and reused by later lessons. Application composition still chooses which flows each process subscribes to; the contract package only builds and parses names and filters. Mosquitto ACLs remain a separate broker authorization layer. ```sh mkdir -p internal/contracts internal/mqtt cmd/simulator ``` Create `internal/contracts/topic.go`: ```go package contracts import ( "fmt" "strings" ) type TopicName string type TopicFilter string type Kind string const ( KindTelemetry Kind = "telemetry" KindCommands Kind = "commands" KindResponses Kind = "responses" KindSales Kind = "sales" ) const topicPrefix = "lab/v1/devices/" type Topic struct { DeviceID string Kind Kind } func (t TopicName) String() string { return string(t) } func (f TopicFilter) String() string { return string(f) } func DeviceTopic(deviceID string, kind Kind) (TopicName, error) { if !validID(deviceID) { return "", fmt.Errorf("invalid device ID %q", deviceID) } if !supportedKind(kind) { return "", fmt.Errorf("unsupported topic kind %q", kind) } return TopicName(topicPrefix + deviceID + "/" + string(kind)), nil } func DeviceFilter(kind Kind) (TopicFilter, error) { if !supportedKind(kind) { return "", fmt.Errorf("unsupported topic kind %q", kind) } return TopicFilter(topicPrefix + "+/" + string(kind)), nil } func DeviceSubscription(deviceID string, kind Kind) (TopicFilter, error) { topic, err := DeviceTopic(deviceID, kind) if err != nil { return "", err } return TopicFilter(topic.String()), nil } func ParseTopic(value string) (Topic, error) { if !strings.HasPrefix(value, topicPrefix) { return Topic{}, fmt.Errorf("topic must be %s{deviceID}/{kind}", topicPrefix) } parts := strings.Split(strings.TrimPrefix(value, topicPrefix), "/") if len(parts) != 2 || !validID(parts[0]) { return Topic{}, fmt.Errorf("topic must be %s{deviceID}/{kind}", topicPrefix) } kind := Kind(parts[1]) if !supportedKind(kind) { return Topic{}, fmt.Errorf("unsupported topic kind %q", kind) } return Topic{DeviceID: parts[0], Kind: kind}, nil } func supportedKind(kind Kind) bool { switch kind { case KindTelemetry, KindCommands, KindResponses, KindSales: return true default: return false } } func validID(value string) bool { if value == "" || len(value) > 128 { return false } for _, r := range value { if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || strings.ContainsRune("._-", r)) { return false } } return true } ``` ### πŸ” `topic.go` β€” file overview This file is the source of truth for the lab's MQTT routing shape. `DeviceTopic` creates a concrete publish name; `DeviceFilter` creates a fleet-wide per-flow subscription filter; `DeviceSubscription` creates an exact per-device filter; and `ParseTopic` turns an incoming concrete topic into routing data. `TopicName` and `TopicFilter` are distinct string types so code must consciously convert them at the Paho boundary. #### 🧱 `DeviceTopic` β€” build one publish destination **Summary:** Validates the device ID and flow kind, then returns a topic with no wildcard levels. **Details:** Device IDs must fit one topic level and may contain only ASCII letters, digits, dot, underscore, and hyphen. This avoids slash injection and prevents a wildcard from accidentally becoming part of a publish name. It is a shape check, not proof that a device is authorized to publish there. #### πŸ”Ž `DeviceFilter` and `DeviceSubscription` β€” choose filter scope **Summary:** `DeviceFilter` subscribes to one flow across devices; `DeviceSubscription` subscribes to one exact device and flow. **Details:** These functions control wildcard placement instead of letting callers concatenate arbitrary strings. The backend uses fleet filters where it must ingest multiple devices; the simulator uses an exact device subscription for least privilege. Mosquitto ACLs still decide whether either role may subscribe. #### 🧭 `ParseTopic` β€” extract route identity **Summary:** Checks the shared `topicPrefix`, then validates the remaining device ID and flow kind. **Details:** `ParseTopic` first requires the incoming name to start with `topicPrefix`, then splits only the suffix into exactly two levels. The parser does not repeat the prefix's individual segments, so changing the constant updates builders and parsing together. The returned device ID is routing metadata to compare with the payload claim; it does not authenticate the publishing client. #### βœ… `supportedKind` and `validID` β€” share the contract rules **Summary:** Centralize the allowed final segments and safe single-level identifier syntax. **Details:** Both building and parsing use these rules, preventing the producer and consumer from quietly accepting different topic vocabularies. Create `internal/mqtt/client.go`: ```go package mqtt import ( "context" "fmt" "log" "net/url" "time" "example.com/mqtt5-learning/internal/config" "example.com/mqtt5-learning/internal/contracts" "github.com/eclipse/paho.golang/autopaho" "github.com/eclipse/paho.golang/paho" ) type Message struct { Topic string Payload []byte Retain bool } // Start creates the MQTT connection, subscribes to the requested topics, and // waits until the initial connection and subscriptions are ready. func Start(ctx context.Context, cfg config.Config, topics []contracts.TopicFilter, inbox chan<- Message) (*autopaho.ConnectionManager, error) { // Parse the configured broker URL before creating the reconnecting client. broker, err := url.Parse(cfg.MQTT.URL) if err != nil { return nil, err } // The callback reports its first SUBACK so startup can fail early if needed. initialSubscription := make(chan error, 1) manager, err := autopaho.NewConnection(ctx, autopaho.ClientConfig{ ServerUrls: []*url.URL{broker}, KeepAlive: 20, CleanStartOnInitialConnection: true, SessionExpiryInterval: 0, ConnectUsername: cfg.MQTT.Username, ConnectPassword: []byte(cfg.MQTT.Password), // Recreate subscriptions after every successful connection or reconnect. OnConnectionUp: func(cm *autopaho.ConnectionManager, _ *paho.Connack) { // This runs again on reconnect; reestablish broker subscriptions. subCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() subscriptions := make([]paho.SubscribeOptions, 0, len(topics)) for _, topic := range topics { subscriptions = append(subscriptions, paho.SubscribeOptions{Topic: topic.String(), QoS: 1, RetainAsPublished: true}) } // Ask for QoS 1 and preserve RETAIN so consumers can identify stored data. ack, subErr := cm.Subscribe(subCtx, &paho.Subscribe{Subscriptions: subscriptions}) if subErr == nil { if ack == nil || len(ack.Reasons) != len(topics) { subErr = fmt.Errorf("missing or incomplete SUBACK") } else { for i, reason := range ack.Reasons { if reason > 1 { // We requested QoS 1; valid grants are 0 or 1. subErr = fmt.Errorf("subscription %s rejected: 0x%02x", topics[i], reason) break } } } } if subErr != nil { log.Printf("subscribe failed: %v", subErr) } else { log.Printf("subscriptions ready: %v", topics) } select { case initialSubscription <- subErr: default: } }, OnConnectionDown: func() bool { log.Print("disconnected; reconnecting"); return true }, OnConnectError: func(err error) { log.Printf("connect error: %v", err) }, // Keep the packet callback short: copy the message and enqueue it for work. ClientConfig: paho.ClientConfig{ ClientID: cfg.MQTT.ClientID, OnClientError: func(err error) { log.Printf("client error: %v", err) }, OnServerDisconnect: func(d *paho.Disconnect) { log.Printf("disconnect reason=0x%02x", d.ReasonCode) }, OnPublishReceived: []func(paho.PublishReceived) (bool, error){ func(received paho.PublishReceived) (bool, error) { packet := received.Packet message := Message{Topic: packet.Topic, Payload: append([]byte(nil), packet.Payload...), Retain: packet.Retain} // Never block the network callback when the bounded inbox is full. select { case inbox <- message: default: // Deliberately lossy early demo: automatic MQTT ACK is still possible. log.Printf("lab inbox full; message dropped topic=%s", packet.Topic) } return true, nil }, }, }, }) if err != nil { return nil, fmt.Errorf("start AutoPaho: %w", err) } // Bound initial connection and subscription setup by the configured timeout. startup, cancel := context.WithTimeout(ctx, cfg.ConnectTimeout()) defer cancel() if err := manager.AwaitConnection(startup); err != nil { Stop(manager) return nil, fmt.Errorf("await connection: %w", err) } select { case err := <-initialSubscription: if err != nil { Stop(manager) return nil, err } case <-startup.Done(): Stop(manager) return nil, fmt.Errorf("await subscriptions: %w", startup.Err()) } return manager, nil } func Publish(ctx context.Context, manager *autopaho.ConnectionManager, topic contracts.TopicName, payload []byte) error { // Publishing waits only for this bounded operation, not for device completion. operation, cancel := context.WithTimeout(ctx, 3*time.Second) defer cancel() ack, err := manager.Publish(operation, &paho.Publish{ Topic: topic.String(), QoS: 1, Retain: false, Payload: payload, Properties: &paho.PublishProperties{ContentType: "application/json"}, }) if err != nil { return fmt.Errorf("publish: %w", err) } if ack == nil { return fmt.Errorf("missing QoS 1 publication response") } if ack.ReasonCode >= 0x80 { return fmt.Errorf("publish rejected: 0x%02x", ack.ReasonCode) } log.Printf("broker publish acknowledgement topic=%s reason=0x%02x", topic, ack.ReasonCode) return nil } func Stop(manager *autopaho.ConnectionManager) { // Give the MQTT disconnect handshake a short, independent shutdown window. shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := manager.Disconnect(shutdown); err != nil { log.Printf("shutdown: %v", err) } } ``` ### πŸ” `internal/mqtt/client.go` β€” file overview This file is the transport adapter shared by the backend and simulator. It starts AutoPaho, restores subscriptions after connection-up, copies received packet bytes into an application-owned inbox, publishes QoS 1 messages with a deadline, and disconnects cleanly. It does not validate vending contracts or decide what a device message means. #### πŸ”Œ `Start` β€” connect and establish subscriptions **Summary:** Creates the connection manager, subscribes to the requested filters, checks each SUBACK reason, and waits for initial readiness. **Details:** Connection-up can happen again after reconnect, so subscription setup belongs in that callback. The initial readiness channel lets the caller distinguish CONNACK from successful subscriptions. The publish-receive callback copies the payload before it leaves the packet context and uses a nonblocking inbox send; if the bounded inbox is full, this early example drops the message and logs the loss. #### πŸ“€ `Publish` β€” publish one bounded operation **Summary:** Sends a QoS 1 JSON message and returns only after the publish operation reports its broker acknowledgement or an error. **Details:** The operation context bounds the wait. A successful acknowledgement covers this publish hop; it does not prove a subscriber received, validated, or applied the message. Timeout can leave the delivery outcome uncertain. #### πŸ›‘ `Stop` β€” disconnect within a deadline **Summary:** Gives AutoPaho a fresh, bounded shutdown context and logs a disconnect error. **Details:** The fresh context remains usable after the signal context has been cancelled. This stops transport lifecycle work; it does not drain a future application worker queue, which Section 6 adds. ## πŸ”— 3. Map the helper to MQTT behavior > πŸ’‘ **Story question:** When `Start` says subscriptions are ready, which broker exchange has happened, and what is the local callback's separate job? The broker has received `SUBSCRIBE` and returned an acceptable `SUBACK`; the callback is configured in the Go process to receive broker-forwarded PUBLISH packets. Read the implementation against these two sides of the receive path. | Code decision | Why it matters | | --- | --- | | Subscribe inside `OnConnectionUp` | The early disposable session loses its broker subscriptions on disconnect | | Inspect every `ack.Reasons` entry | Each requested filter has its own result; connection success is insufficient | | `RetainAsPublished: true` | Preserve the publisher's RETAIN flag on live forwarding, so the simulator can reject live retained commands as well as retained replay | | Copy payload into `Message` | The application queue owns bytes that will be processed after the receive callback returns | | Nonblocking inbox send | Bound callback time and memory; dropping on overflow is deliberately lossy | | `return true, nil` | Report local handling to subsequent callbacks; this is not a negative/positive business acknowledgement | | `QoS: 1`, `Retain: false` | Ask for a broker acknowledgement without storing the command as the topic's retained value | | `ContentType: "application/json"` | Describe the encoding; neither broker nor library validates the application's schema for you | | Initial SUBACK channel | Delay startup until this client's receive path is established; it says nothing about the peer's readiness | SUBACK grants `0`, `1`, or `2` for maximum delivery QoS. Since this program requests QoS 1, it accepts only 0 or 1. Failure reason codes start at `0x80`. Receiving a successful call return without inspecting subscription results can leave you connected but receiving no useful messages. For QoS 1, `Publish` returns a response describing the immediate broker acknowledgement. Success codes can include `0x10` (no matching subscribers); this does not prove the simulator is online. Transport errors and context timeouts can leave the delivery outcome uncertain. Never retry a real action with a new business ID just because your waiting context expired. `return true, nil` means the callback handled the message for local routing. It is not a device acceptance/completion response or durable-storage guarantee. With automatic acknowledgement, Paho can acknowledge after callback processing while your inbox has not yet been handled. Inspect `paho/client.go` in the tagged source to understand callback/ack order. The initial-subscription channel allows startup to wait for SUBACK as well as CONNACK. The callback subscribes again on reconnect. This helper checks initial readiness only; reconnect subscription failures are logged, not represented as continuous health state. Add that state later rather than treating initial readiness as permanent. ## πŸ’» 4. Replace the backend main program > πŸ’‘ **Story question:** After configuration, connection, and subscriptions are ready, how does a backend timer become an MQTT command? ### 🦟 Trace one ping from Go value to broker On each timer tick, the backend first builds an ordinary Go value. `json.Marshal` turns that value into the byte sequence sent as the MQTT payload: ```go struct { ID string `json:"id"` Command string `json:"command"` }{ID: "ping-123-1", Command: "ping"} ``` The JSON field tags set the wire names, so the payload bytes represent: ```json {"id":"ping-123-1","command":"ping"} ``` The backend asks `contracts.DeviceTopic(deviceID, contracts.KindCommands)` for the concrete destination. That constructor validates the device ID as one safe topic level. It passes the resulting `TopicName` and JSON bytes to `mqtt.Publish`; that helper builds a Paho `paho.Publish` with **QoS 1**, **Retain false**, and Content Type `application/json`, then calls AutoPaho's `manager.Publish`. `manager.Publish` waits for the QoS 1 publish exchange or an error/timeout. When it receives a successful `PUBACK`, the helper logs the reason code and returns nil. That confirms the backend-to-broker QoS exchange. A success reason of `0x10` means the broker had no matching subscribers; it is still a successful MQTT acknowledgement, so it does not prove the simulator received anything. A timeout can also leave the outcome uncertain. The broker independently forwards a matching PUBLISH to the simulator's `.../commands` subscription. The simulator's Paho callback copies it into an inbox; the simulator's main loop decodes the JSON, checks that it is a non-retained `ping` with an ID, and publishes a `pong` carrying that same ID to `.../responses`. That new response is application-level evidence the simulator handled the ping. The backend is subscribed to the response topic and currently logs the received bytes; matching responses to pending commands is added in Lesson 7. ```text Go struct β†’ JSON payload bytes β†’ QoS 1 PUBLISH β†’ broker PUBACK β””β†’ simulator receives and handles ping β†’ separate pong PUBLISH β†’ backend receives ``` The two arrows after the backend's PUBLISH are separate outcomes: PUBACK may arrive before or after the simulator's response. Retain is false because a ping is a one-time action; storing it as retained state could replay the command to a later subscriber. Create `cmd/backend/main.go`: ```go package main import ( "context" "encoding/json" "fmt" "log" "os" "os/signal" "syscall" "time" "example.com/mqtt5-learning/internal/config" "example.com/mqtt5-learning/internal/contracts" mqtt "example.com/mqtt5-learning/internal/mqtt" ) func main() { // Convert a returned application error into a visible nonzero process exit. if err := run(); err != nil { log.Print(err) os.Exit(1) } } func run() error { // Load settings and bind cancellation to Ctrl+C or the container stop signal. cfg, err := config.Load(os.Args[1:]) if err != nil { return err } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() // Build subscription filters and a concrete command destination. telemetryFilter, err := contracts.DeviceFilter(contracts.KindTelemetry) if err != nil { return err } responsesFilter, err := contracts.DeviceFilter(contracts.KindResponses) if err != nil { return err } commandTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindCommands) if err != nil { return err } inbox := make(chan mqtt.Message, 16) manager, err := mqtt.Start(ctx, cfg, []contracts.TopicFilter{telemetryFilter, responsesFilter}, inbox) if err != nil { return err } defer mqtt.Stop(manager) // Publish synthetic ping commands periodically while processing received data. ticker := time.NewTicker(10 * time.Second) defer ticker.Stop() runID := time.Now().UnixNano() sequence := 0 for { // The select keeps shutdown, inbound messages, and periodic commands responsive. select { case <-ctx.Done(): return nil case message := <-inbox: // Synthetic lab payloads only; production payload logging needs redaction. log.Printf("received topic=%s retained=%t payload=%s", message.Topic, message.Retain, message.Payload) case <-ticker.C: sequence++ payload, err := json.Marshal(struct { ID string `json:"id"` Command string `json:"command"` }{ID: fmt.Sprintf("ping-%d-%d", runID, sequence), Command: "ping"}) if err != nil { return err } if err := mqtt.Publish(ctx, manager, commandTopic, payload); err != nil { log.Printf("command publication failed or outcome uncertain: %v", err) } } } } ``` #### 🧭 `main` and `run` β€” assemble the backend process **Summary:** `main` turns a returned error into a failing process; `run` loads settings, starts MQTT, then handles inbound messages and periodic pings until cancellation. **Details:** `main` is intentionally small so startup errors have one exit path. `run` owns resources in reverse order: it creates the signal context, starts the manager, and defers shutdown; it also stops the ticker. Its `select` keeps messages, timer work, and cancellation responsive. A publish error is logged as uncertain because the client may not know whether the broker accepted the packet. The ticker sends one ping every ten seconds. It starts after the backend's own subscriptions are ready. A timestamp and sequence make IDs distinct enough for this demonstration, not a production idempotency system. The backend logs the response but does not yet implement pending requests or validate completion; Section 7 adds those semantics. ## πŸ› οΈ 5. Create the device simulator > πŸ’‘ **Story question:** How does a received MQTT packet become a device action, and how does the simulator report its result without blocking Paho's network callback? The callback copies the packet into a bounded inbox. The simulator's main loop reads it, validates the harmless ping, and publishes a pong. This draws a deliberate boundary between packet reception and application behavior; later lessons examine the loss window and acknowledgement timing. Create `cmd/simulator/main.go`: ```go package main import ( "context" "encoding/json" "log" "os" "os/signal" "syscall" "time" "example.com/mqtt5-learning/internal/config" "example.com/mqtt5-learning/internal/contracts" mqtt "example.com/mqtt5-learning/internal/mqtt" ) func main() { // Convert a returned application error into a visible nonzero process exit. if err := run(); err != nil { log.Print(err) os.Exit(1) } } func run() error { // Load settings and bind cancellation to Ctrl+C or the container stop signal. cfg, err := config.Load(os.Args[1:]) if err != nil { return err } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() // Build one command filter and this device's concrete publish destinations. commandFilter, err := contracts.DeviceSubscription(cfg.MQTT.DeviceID, contracts.KindCommands) if err != nil { return err } telemetryTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindTelemetry) if err != nil { return err } responseTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindResponses) if err != nil { return err } inbox := make(chan mqtt.Message, 16) manager, err := mqtt.Start(ctx, cfg, []contracts.TopicFilter{commandFilter}, inbox) if err != nil { return err } defer mqtt.Stop(manager) // Heartbeats are periodic; command handling remains active between ticks. ticker := time.NewTicker(3 * time.Second) defer ticker.Stop() for { // Handle cancellation, telemetry publication, and inbound commands together. select { case <-ctx.Done(): return nil case <-ticker.C: payload, err := json.Marshal(struct { DeviceID string `json:"device_id"` Time string `json:"time"` }{cfg.MQTT.DeviceID, time.Now().UTC().Format(time.RFC3339)}) if err != nil { return err } if err := mqtt.Publish(ctx, manager, telemetryTopic, payload); err != nil { log.Printf("heartbeat: %v", err) } case message := <-inbox: // Decode into a local command value only after rejecting retained messages. var command struct { ID string `json:"id"` Command string `json:"command"` } if message.Retain { log.Print("ignoring retained command") continue } if err := json.Unmarshal(message.Payload, &command); err != nil || command.ID == "" || command.Command != "ping" { log.Print("ignoring malformed or unsupported command") continue } // Echo the command ID so the backend can associate this pong with its ping. response, err := json.Marshal(struct { ID string `json:"id"` DeviceID string `json:"device_id"` Status string `json:"status"` }{command.ID, cfg.MQTT.DeviceID, "pong"}) if err != nil { return err } if err := mqtt.Publish(ctx, manager, responseTopic, response); err != nil { log.Printf("response: %v", err) } } } } ``` #### 🧭 `main` and `run` β€” simulate a vending device **Summary:** `main` reports process failure; `run` publishes a heartbeat every three seconds and handles commands from the device's command topic. **Details:** The simulator uses its own identity and subscribes only to its command namespace. The receive callback hands messages to `inbox`; the main loop validates and responds outside that callback. Cancellation returns cleanly and deferred cleanup stops the ticker and MQTT manager. This simulator is a learning stand-in, not a real vending device driver. `json.Marshal` serializes Go structs to JSON bytes. Tags choose field names. `json.Unmarshal(message.Payload, &command)` needs a pointer so the decoder can fill the struct. The simulator's processing loop publishes responses outside the receive callback, keeping the callback short and avoiding blocking it on a network round trip. This simulator sends a heartbeat every three seconds, receives ping commands, and publishes a pong carrying the same ID. It rejects packets marked retained to avoid treating stored state as a fresh one-shot action. The subscription sets `RetainAsPublished: true`: with the default false, live forwarding can clear RETAIN and defeat a guard that checks only `message.Retain`. Keep command publishers nonretaining too; this guard does not remove a retained value already stored at the broker. Full contract validation and duplicate handling arrive in later sections; this only demonstrates the transport lifecycle. ## ▢️ 6. Run two separate identities > πŸ’‘ **Story question:** Why start two processes with different usernames and Client IDs, and what output proves each stage of the path? The simulator and backend represent independent clients. Their connection logs, subscription results, broker publish acknowledgement, and pong are separate evidence. Start the simulator first so its command subscription exists before the backend publishes. Terminal A, simulator first: ```sh go run ./cmd/simulator -username simulator -client-id lab-simulator-device-001 ``` Terminal B, backend: ```sh go run ./cmd/backend -username backend -client-id lab-backend ``` Open two independent terminal sessions and change both to the project root. The shared lab password comes from `configs/lab.toml`; `-username` selects the broker account for each process, and `-client-id` overrides the TOML client ID. Stop each foreground program with Ctrl+C when finished. Use the same configured `device-001` in both. Distinct usernames authorize different topic directions; both local accounts use the test password in `configs/lab.toml`. Distinct client IDs prevent connection takeover. A username, client ID, and payload device ID serve different purposes; the ACL is what binds this simulator credential to its permitted device namespace. Expected backend output, abbreviated (timestamps and IDs vary): ```text subscriptions ready: [lab/v1/devices/device-001/telemetry lab/v1/devices/device-001/responses] received topic=lab/v1/devices/device-001/telemetry ... payload={"device_id":"device-001","time":"..."} broker publish acknowledgement topic=lab/v1/devices/device-001/commands reason=0x00 received topic=lab/v1/devices/device-001/responses ... payload={"id":"ping-...","device_id":"device-001","status":"pong"} ``` Log order is not guaranteed: the device response may be observed around the publication acknowledgement. Separate connections and goroutines progress independently. If the simulator is not subscribed yet, a ping may have no recipient; this is expected with these zero-expiry early-lab sessions. ## πŸ§ͺ 7. Experiments to learn the API > πŸ’‘ **Story question:** Which part of the system changes in each experimentβ€”the broker's routing/state, the client's delivery setting, or the application's handling policy? Predict the effect at that boundary before changing the code, then use packet or application evidence to explain what happened. 1. Replace the backend's exact telemetry subscription with `lab/v1/devices/+/telemetry`. Explain the topic filter versus a concrete publish topic. Never publish to a topic containing `+` or `#`. 2. Add `Retain: true` only to synthetic telemetry; restart the backend and observe a retained value. Restore `false` afterward. Do not retain one-shot commands. 3. Change publication QoS to 0. Modify the helper's acknowledgement check: QoS 0 has no broker PUBACK; any returned response is not evidence of broker confirmation. Compare CLI packet traces at QoS 0 and 1. 4. Stop the simulator while the backend continues. Observe broker acknowledgement without a pong. Explain why the broker cannot tell you the vending application completed a ping. 5. Restart the broker. Observe reconnect and renewed subscription readiness. Ping delivery during the reconnect interval is not guaranteed by this helper. 6. Change simulator `-device-id device-002`. Its existing ACL must prevent the new subscriptions/publications; do not broaden its ACL to hide the failure. 7. Reduce inbox capacity and add a slow consumer. Observe explicit drops; explain why automatic acknowledgement can make those drops irreversible.
Self-check: why does QoS 1 still lose a command in the full-inbox experiment? The immediate receiver is the MQTT client. Its automatic acknowledgement can complete delivery even though your callback drops the application copy. QoS cannot recover work your own receive path has acknowledged and discarded. Preserve important work before acknowledging, with an explicit overload and recovery policy.
Self-check: why does publishing with Retain false not clear the previous retained value? The new publication does not ask the broker to change retained storage. Delete the stored value using a zero-byte retained publication to the same exact topic, then test with a fresh subscription. Stop any retained publisher during that check so it cannot immediately replace the value.
## ⚠️ Cautions: limits of this early example - The inbox is deliberately bounded and lossy. It makes overload visible without blocking receive callbacks. It is unsuitable for durable sales reports or real commands; later sections replace this policy with persistence and deliberate acknowledgement/backpressure rules. - A publish call can block this single application loop for up to three seconds; this is acceptable for a tiny demonstration. Later sections add workers and separate business processing. - QoS 1 can duplicate delivery. Ping is harmless, but real effects need stable command IDs and device-side deduplication. - Payload validation here is intentionally minimal. Content Type is descriptive metadata, not authentication or schema enforcement. - Fresh sessions and zero expiry mean offline commands are not a durable workflow. Do not infer restart guarantees from a successful reconnect. - Raw payload logs are allowed here only for small synthetic lab messages; production logs need redaction and size limits. - Never close the inbox while Paho may still send to it. These programs let the context and manager stop the producer; the channel need not be closed for process exit. ## βœ… Verification and completion ```sh gofmt -w cmd internal go test ./... go vet ./... go build ./cmd/backend ./cmd/simulator ./cmd/configcheck ``` `mkdir -p` above creates missing parent directories without creating Go files. `gofmt -w cmd internal` formats Go files under those directories. `go test ./...` runs tests in all packages in this module; `...` is Go's recursive package pattern. `go vet ./...` performs static checks. `go build` with several package paths checks all those executables without writing one combined binary. Run these from the root, after creating all referenced packages. For the retain experiment, publish a zero-byte retained payload to that exact telemetry topic to clear its retained value; merely switching future publications back to nonretained does not erase the broker's old retained message. Publish with an identity authorized to write the topic. For broker restart, use the explained lifecycle command in [Broker operations](broker-operations.md), then watch both clients reconnect and resubscribe. Packages without tests do not demonstrate runtime correctness. Also run both programs, perform the failure experiments, and verify clean Ctrl+C behavior. **Completion:** Explain the complete path from backend Go struct β†’ JSON bytes β†’ MQTT PUBLISH β†’ simulator callback β†’ application loop β†’ response PUBLISH β†’ backend callback. Distinguish broker acknowledgement from a device response, and identify the lossy queue and offline-session limitations yourself. ## πŸ”— Sources - [Paho client API v0.23.0](https://pkg.go.dev/github.com/eclipse/paho.golang@v0.23.0/paho) - [Tagged AutoPaho source](https://github.com/eclipse-paho/paho.golang/blob/v0.23.0/autopaho/auto.go) - [Tagged receive callback source](https://github.com/eclipse-paho/paho.golang/blob/v0.23.0/paho/client.go) - [MQTT 5 specification](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html) ## βœ… Result β€” the story so far The backend now subscribes to device telemetry and responses, while the simulator subscribes to that device's commands. The simulator publishes heartbeats for the backend; when the backend sends a QoS 1 ping, the broker forwards it to the simulator, which handles it and publishes a pong carrying the same command ID. The PUBACKs confirm their individual MQTT hops; the pong is the separate application evidence that the simulator handled the ping. Next: [Section 5 β€” Contracts and validation](05-contracts-validation.md).