mqtt-5-learning/lessons/04-publish-subscribe-simulator.md

# 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`).


## 1. Map the conversation

| 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
    M-->>B: PUBACK for backend-to-broker hop
    M->>S: PUBLISH command, granted delivery QoS
    S-->>M: PUBACK for broker-to-simulator hop
    S->>M: PUBLISH application result
    M->>B: PUBLISH application result
```

This diagram groups exchanges by purpose; it does not promise the backend sees PUBACK before the simulator's response. Each side has its own connection and packet identifiers. The broker routes by topic, not by JSON fields. A result is another publication on another topic; MQTT does not create a response automatically.

### 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

```sh
mkdir -p internal/mqtt cmd/simulator
```

Create `internal/mqtt/client.go`:

```go
package mqtt

import (
	"context"
	"fmt"
	"log"
	"net/url"
	"time"

	"example.com/mqtt5-learning/internal/config"
	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
)

type Message struct {
	Topic   string
	Payload []byte
	Retain  bool
}

func Start(ctx context.Context, cfg config.Config, topics []string, inbox chan<- Message) (*autopaho.ConnectionManager, error) {
	broker, err := url.Parse(cfg.MQTT.URL)
	if err != nil {
		return nil, err
	}
	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),
		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, QoS: 1, RetainAsPublished: true})
			}
			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) },
		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}
					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)
	}
	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 string, payload []byte) error {
	operation, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()
	ack, err := manager.Publish(operation, &paho.Publish{
		Topic: topic, 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) {
	shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := manager.Disconnect(shutdown); err != nil {
		log.Printf("shutdown: %v", err)
	}
}
```


## 3. Map the helper to MQTT behavior

| 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

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"
	mqtt "example.com/mqtt5-learning/internal/mqtt"
)

func main() {
	if err := run(); err != nil {
		log.Print(err)
		os.Exit(1)
	}
}

func run() error {
	cfg, err := config.Load(os.Args[1:])
	if err != nil {
		return err
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	base := "lab/v1/devices/" + cfg.MQTT.DeviceID
	inbox := make(chan mqtt.Message, 16)
	manager, err := mqtt.Start(ctx, cfg, []string{base + "/telemetry", base + "/responses"}, inbox)
	if err != nil {
		return err
	}
	defer mqtt.Stop(manager)
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	runID := time.Now().UnixNano()
	sequence := 0
	for {
		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, base+"/commands", payload); err != nil {
				log.Printf("command publication failed or outcome uncertain: %v", err)
			}
		}
	}
}
```


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

Create `cmd/simulator/main.go`:

```go
package main

import (
	"context"
	"encoding/json"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"example.com/mqtt5-learning/internal/config"
	mqtt "example.com/mqtt5-learning/internal/mqtt"
)

func main() {
	if err := run(); err != nil {
		log.Print(err)
		os.Exit(1)
	}
}

func run() error {
	cfg, err := config.Load(os.Args[1:])
	if err != nil {
		return err
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	base := "lab/v1/devices/" + cfg.MQTT.DeviceID
	inbox := make(chan mqtt.Message, 16)
	manager, err := mqtt.Start(ctx, cfg, []string{base + "/commands"}, inbox)
	if err != nil {
		return err
	}
	defer mqtt.Stop(manager)
	ticker := time.NewTicker(3 * time.Second)
	defer ticker.Stop()
	for {
		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, base+"/telemetry", payload); err != nil {
				log.Printf("heartbeat: %v", err)
			}
		case message := <-inbox:
			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
			}
			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, base+"/responses", response); err != nil {
				log.Printf("response: %v", err)
			}
		}
	}
}
```


`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

Terminal A, simulator first:

```sh
export TEST_MQTT5_MQTT_USERNAME=simulator
read -rs 'TEST_MQTT5_MQTT_PASSWORD?Simulator lab password: '
export TEST_MQTT5_MQTT_PASSWORD
go run ./cmd/simulator -client-id lab-simulator-device-001
```

Terminal B, backend:

```sh
export TEST_MQTT5_MQTT_USERNAME=backend
read -rs 'TEST_MQTT5_MQTT_PASSWORD?Backend lab password: '
export TEST_MQTT5_MQTT_PASSWORD
go run ./cmd/backend -client-id lab-backend
```

Open two independent terminal sessions and change both to the project root. Their environments are separate, so the simulator's exported credentials do not overwrite the backend's. The zsh `read -rs` prompt hides the password; `export` passes it to the child process. `go run` runs each executable in the foreground; `-client-id` is a program flag overriding the TOML default, not a Docker setting. Stop each program with Ctrl+C and unset its password when finished.

Use the same configured `device-001` in both. Distinct username/password pairs authorize different topic directions. 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

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.

<details>
<summary>Self-check: why does QoS 1 still lose a command in the full-inbox experiment?</summary>

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.

</details>

<details>
<summary>Self-check: why does publishing with Retain false not clear the previous retained value?</summary>

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.

</details>

## 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)

Next: [Section 5 — Contracts and validation](05-contracts-validation.md).