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.

🎯 One skill: measure before adopting a feature🧠 Predict packet effectsπŸ§ͺ Version-pinned lab

❓ Start with the system story

Two ingestion processes share a subscription. A command response reaches the instance that did not create the original process-local waiter. Which state design still lets that instance process the result correctly?

Answer with the shared durable source of truth, then explore the protocol controls.

πŸ“¬ Subscription options shape delivery

OptionChangesDoes not do
No LocalSuppresses same-client-originated publications for that subscriptionAuthorize a topic or filter a device identity
Retain As PublishedPreserves incoming RETAIN bit on forwarded live trafficChoose whether broker stores retained state
Retain HandlingControls retained replay when subscription is establishedBlock future live retained publications
UNSUBSCRIBERemoves the broker subscription after UNSUBACKErase retained state or unregister a local Go handler

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.

🧠 Check scaling behavior: What makes a shared-subscription response safe when it reaches a different ingestion instance?




βš–οΈ 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.”

πŸ“„ Complete file contents
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
}

πŸ”Ž Details: limits and capability defaults

The helper does not resize your application queue or guarantee that the broker supports a requested behavior. The server's absent optional numeric properties stay nil; callers must apply the MQTT-defined defaults for the relevant feature. This small struct records the fields needed by the experiment and does not claim to capture every negotiated capability. Topic Alias state must be rebuilt after a new network connection.

πŸ§ͺ 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.

βœ… Done when

Your matrix distinguishes MQTT specification, broker, client-library, and application behavior. Every feature has evidence or is marked unverified; no optimization is selected just because the API exposes a field.

πŸ“‹ Short self-check answers
  1. Retain Handling controls replay at subscribe time; Retain As Published controls a flag on forwarded live traffic.
  2. Shared subscriptions distribute work and do not guarantee affinity; durable state lets any instance process safely.
  3. Receive Maximum limits in-flight QoS 1/2 packets, not an application's total queue.
  4. Aliases need to be re-established after reconnect.

πŸ“š 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.