Go MQTT Vending Backend · Lesson 16 · 15 minutes

Choose MQTT controls from measured behavior

Experiment with subscription options, shared consumers, flow limits, packet sizes, aliases, and reason codes without confusing protocol controls with application policy.

📬 Subscription options shape delivery

Change one option at a time with a retained status topic. Record whether the publication was a retained replay or live delivery and inspect the packet flags. Never set No Local as a substitute for ACLs.

⚖️ Shared subscriptions and limits

Run two ingestion instances in one shared group. Observe distribution and what happens when one stops. Do not assume per-device ordering or affinity; both workers must use durable idempotency. MQTT Receive Maximum limits unacknowledged QoS 1/2 protocol exchanges in a direction. It is not your Go queue capacity, and it does not limit QoS 0 in the same way.

Advertised Maximum Packet Size limits packet size in the direction advertised. Test client-to-broker and broker-to-client separately. Compare protocol inflight count with application queue depth; they can differ sharply if callbacks acknowledge before queued work is durable.

🏷️ Aliases, reason codes, and support matrix

Topic Alias mappings are scoped to a network connection. A reconnect needs a new mapping even if the MQTT session resumes. Confirm the pinned Paho and Mosquitto support and lifecycle reset behavior before treating aliases as an optimization; a TopicAlias field existing in a type is not evidence that a full flow works.

Interpret reason codes in their packet context. PUBACK “No matching subscribers” is not a device response. A SUBACK result applies to its filter. A context deadline means the local wait ended and may leave business outcome uncertain. Use bounded reconnect backoff with jitter for transient failures; repair authentication/configuration errors instead of retrying faster.

🧭 File overview: internal/mqtt/controls.go

Create this complete helper file to build MQTT 5 CONNECT and SUBSCRIBE options without mixing broker protocol limits with local queue capacity. The caller still inspects actual CONNACK/SUBACK results and checks version-matched Paho support.

🧰 NewConnectProperties — short summary

Builds validated client Receive Maximum and Maximum Packet Size properties.

📬 NewSubscription — short summary

Builds one subscription and rejects invalid QoS, Retain Handling, and No Local on a shared filter.

🔎 CapabilitiesFromConnack — short summary

Copies optional server capability values safely, preserving nil as “not advertised.”

package mqtt

import (
	"fmt"
	"strings"

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

type ServerCapabilities struct {
	SessionPresent bool
	MaximumQoS *byte
	ReceiveMaximum *uint16
	MaximumPacketSize *uint32
	TopicAliasMaximum *uint16
	SharedSubscriptions bool
	RetainedMessages bool
}

func NewConnectProperties(receiveMaximum uint16, maximumPacketSize uint32) (*paho.ConnectProperties, error) {
	if receiveMaximum == 0 { return nil, fmt.Errorf("receive maximum must be at least one") }
	if maximumPacketSize == 0 { return nil, fmt.Errorf("maximum packet size must be at least one byte") }
	packetLimit := maximumPacketSize
	flightLimit := receiveMaximum
	return &paho.ConnectProperties{ReceiveMaximum: &flightLimit, MaximumPacketSize: &packetLimit}, nil
}

func NewSubscription(filter contracts.TopicFilter, qos, retainHandling byte, noLocal, retainAsPublished bool) (paho.SubscribeOptions, error) {
	value := filter.String()
	if value == "" { return paho.SubscribeOptions{}, fmt.Errorf("subscription filter is required") }
	if qos > 2 { return paho.SubscribeOptions{}, fmt.Errorf("QoS must be 0, 1, or 2") }
	if retainHandling > 2 { return paho.SubscribeOptions{}, fmt.Errorf("Retain Handling must be 0, 1, or 2") }
	if noLocal && strings.HasPrefix(value, "$share/") {
		return paho.SubscribeOptions{}, fmt.Errorf("No Local cannot be used with a shared subscription")
	}
	return paho.SubscribeOptions{
		Topic: value, QoS: qos, RetainHandling: retainHandling,
		NoLocal: noLocal, RetainAsPublished: retainAsPublished,
	}, nil
}

func CapabilitiesFromConnack(ack *paho.Connack) ServerCapabilities {
	if ack == nil { return ServerCapabilities{} }
	result := ServerCapabilities{SessionPresent: ack.SessionPresent}
	if ack.Properties == nil { return result }
	result.MaximumQoS = copyByte(ack.Properties.MaximumQoS)
	result.ReceiveMaximum = copyUint16(ack.Properties.ReceiveMaximum)
	result.MaximumPacketSize = copyUint32(ack.Properties.MaximumPacketSize)
	result.TopicAliasMaximum = copyUint16(ack.Properties.TopicAliasMaximum)
	result.SharedSubscriptions = ack.Properties.SharedSubAvailable
	result.RetainedMessages = ack.Properties.RetainAvailable
	return result
}

func copyByte(value *byte) *byte {
	if value == nil { return nil }; copy := *value; return &copy
}
func copyUint16(value *uint16) *uint16 {
	if value == nil { return nil }; copy := *value; return &copy
}
func copyUint32(value *uint32) *uint32 {
	if value == nil { return nil }; copy := *value; return &copy
}

🧪 Build a feature decision matrix

1. Record Paho and broker versions and the exact option/API field.
2. For each experiment record purpose, packet and direction, prediction, actual observation, and whether you would use the feature.
3. Test No Local, Retain As Published, and each Retain Handling value with stored and live messages.
4. Run two shared consumers; stop one during traffic and inspect distribution, duplicates, and result processing.
5. Test Receive Maximum and packet-size negotiation in both directions. Measure protocol inflight and application queue separately.
6. Review Topic Alias and reason-code paths against the pinned source; mark reading-only items as not experimentally verified.

📚 Detailed experiments and Paho v0.23.0 source pointers: Lesson 16 source lesson. Primary references: OASIS MQTT 5.0 Standard and Paho v0.23.0 API.

Ask me to map an observed control or reason code to its packet and owning layer.