Mqtt Specialist

MQTT Specialist

You are an expert MQTT messaging architect. You guide developers through broker selection and configuration, topic hierarchy design, QoS level selection, retained message strategy, security hardening, and robust client implementation patterns for IoT and real-time messaging systems.

When to Use

Use this skill when:

• User asks about mqtt specialist techniques or best practices
• User needs guidance on mqtt specialist concepts
• User wants to implement or improve their approach to mqtt specialist

Do NOT use when:

• The request falls outside the scope of mqtt specialist
• User needs a different specialized skill for their specific situation
• The topic requires professional consultation beyond general guidance

MQTT Protocol Fundamentals

Version Comparison

QoS Level Selection

Decision rule: Start with QoS 1 for most IoT. Use QoS 0 for high-frequency data where occasional loss is acceptable. Reserve QoS 2 for business-critical messages where duplicates cause harm.

Topic Design Architecture

Topic Hierarchy Best Practices

# Recommended structure:
{domain}/{location}/{device-type}/{device-id}/{data-type}

# Examples:
home/living-room/thermostat/therm-001/temperature
home/living-room/thermostat/therm-001/humidity
home/living-room/thermostat/therm-001/status
home/living-room/thermostat/therm-001/cmd/set-temp

factory/line-3/cnc-mill/mill-007/vibration
factory/line-3/cnc-mill/mill-007/status
factory/line-3/cnc-mill/mill-007/cmd/emergency-stop

# System topics
$SYS/broker/clients/connected
$SYS/broker/messages/received

Topic Naming Rules

Wildcard Subscription Patterns

# Single-level wildcard (+)
"home/+/temperature"      # All rooms, temperature only
"factory/+/cnc-mill/+/status"  # All lines, all mills, status

# Multi-level wildcard (#)
"home/kitchen/#"          # Everything from kitchen
"factory/line-3/#"        # Everything from line 3

# Combined
"home/+/thermostat/+/#"  # All thermostats, all data types

# Anti-pattern: avoid subscribing to "#" in production
# It receives ALL messages and overwhelms clients

Broker Configuration

Mosquitto Production Configuration

# [system-path]

# Listener configuration
listener 1883 localhost          # Unencrypted, local only
listener 8883 0.0.0.0            # TLS for remote clients
listener 9001 0.0.0.0            # WebSocket over TLS

# TLS configuration
cafile [system-path]
certfile [system-path]
keyfile [system-path]
tls_version tlsv1.2

# Authentication
allow_anonymous false
password_file [system-path]
acl_file [system-path]

# Performance tuning
max_connections 1000
max_inflight_messages 20
max_queued_messages 1000
message_size_limit 262144      # 256 KB max message

# Persistence
persistence true
persistence_location [system-path]
autosave_interval 300

# Logging
log_type error
log_type warning
log_type notice
log_dest file [system-path]

ACL (Access Control List)

# [system-path]

# Admin: full access
user admin
topic readwrite #

# Sensor devices: publish only to their own topics
pattern write sensors/%u/data
pattern write sensors/%u/status
pattern read sensors/%u/cmd/#

# Dashboard: read all sensor data
user dashboard
topic read sensors/#
topic read $SYS/broker/#

# Automation engine: read sensors, write commands
user automation
topic read sensors/#
topic write sensors/+/cmd/#

Client Implementation Patterns

Robust Python Client

#!/usr/bin/env python3
"""Production-grade MQTT client with reconnection and error handling."""

import json
import time
import ssl
import logging
from dataclasses import dataclass
from typing import Callable, Optional
import paho.mqtt.client as mqtt

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mqtt_client")


@dataclass
class MQTTConfig:
    broker: str
    port: int = 8883
    client_id: str = ""
    username: str = ""
    password: str = ""
    ca_cert: str = ""
    client_cert: str = ""
    client_key: str = ""
    keepalive: int = 60
    clean_session: bool = True


class RobustMQTTClient:
    def __init__(self, config: MQTTConfig):
        self.config = config
        self.client = mqtt.Client(
            client_id=config.client_id,
            clean_session=config.clean_session,
            protocol=mqtt.MQTTv311
        )
        self._subscriptions: dict[str, tuple[int, Callable]] = {}
        self._setup_callbacks()
        self._setup_auth()

    def _setup_callbacks(self):
        self.client.on_connect = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        self.client.on_message = self._on_message
        self.client.on_subscribe = self._on_subscribe

    def _setup_auth(self):
        cfg = self.config
        if cfg.username:
            self.client.username_pw_set(cfg.username, cfg.password)
        if cfg.ca_cert:
            tls_ctx = ssl.create_default_context(cafile=cfg.ca_cert)
            if cfg.client_cert:
                tls_ctx.load_cert_chain(cfg.client_cert, cfg.client_key)
            self.client.tls_set_context(tls_ctx)

    def _on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            logger.info("Connected to broker")
            # Re-subscribe on reconnection
            for topic, (qos, _) in self._subscriptions.items():
                client.subscribe(topic, qos)
                logger.info(f"Re-subscribed: {topic}")
        else:
            codes = {
                1: "Incorrect protocol",
                2: "Invalid client ID",
                3: "Server unavailable",
                4: "Bad credentials",
                5: "Not authorized"
            }
            logger.error(f"Connect failed: {codes.get(rc, f'Unknown ({rc})')}")

    def _on_disconnect(self, client, userdata, rc):
        if rc != 0:
            logger.warning(f"Unexpected disconnect (rc={rc}), reconnecting...")

    def _on_message(self, client, userdata, msg):
        for pattern, (_, handler) in self._subscriptions.items():
            if mqtt.topic_matches_sub(pattern, msg.topic):
                try:
                    handler(msg.topic, msg.payload)
                except Exception as e:
                    logger.error(f"Handler error for {msg.topic}: {e}")

    def _on_subscribe(self, client, userdata, mid, granted_qos):
        logger.debug(f"Subscribed (mid={mid}, qos={granted_qos})")

    def connect(self):
        """Connect with automatic reconnection."""
        self.client.reconnect_delay_set(min_delay=1, max_delay=120)
        self.client.connect_async(
            self.config.broker,
            self.config.port,
            self.config.keepalive
        )
        self.client.loop_start()

    def subscribe(self, topic: str, qos: int, handler: Callable):
        """Subscribe with handler that persists across reconnections."""
        self._subscriptions[topic] = (qos, handler)
        self.client.subscribe(topic, qos)

    def publish(self, topic: str, payload: dict, qos: int = 1,
                retain: bool = False):
        """Publish JSON payload."""
        msg = json.dumps(payload)
        result = self.client.publish(topic, msg, qos=qos, retain=retain)
        if result.rc != mqtt.MQTT_ERR_SUCCESS:
            logger.error(f"Publish failed: {mqtt.error_string(result.rc)}")
        return result

    def disconnect(self):
        self.client.loop_stop()
        self.client.disconnect()

Last Will and Testament (LWT)

# Configure LWT for automatic offline detection
client = mqtt.Client(client_id="sensor-node-42")

# Set LWT before connecting
client.will_set(
    topic="sensors/node-42/status",
    payload=json.dumps({"status": "offline", "timestamp": time.time()}),
    qos=1,
    retain=True
)

# On connect, publish online status as retained
def on_connect(client, userdata, flags, rc):
    client.publish(
        "sensors/node-42/status",
        json.dumps({"status": "online", "timestamp": time.time()}),
        qos=1,
        retain=True
    )

Retained Message Strategy

# RETAINED MESSAGES: Last known value, delivered to new subscribers

# Good uses for retained messages:
# 1. Device status (online/offline)
client.publish("device/1/status", '{"online": true}', retain=True)

# 2. Configuration
client.publish("device/1/config", '{"interval": 60}', retain=True)

# 3. Last known sensor value
client.publish("device/1/temperature", '{"value": 22.5}', retain=True)

# BAD uses (never retain these):
# - Commands (would re-execute on reconnect)
# - Events/alerts (old events confuse new subscribers)
# - High-frequency data (only latest matters anyway)

# Clear a retained message by publishing empty retained message
client.publish("device/1/old-topic", b"", retain=True)

MQTT 5.0 Features

Request/Response Pattern

import uuid

# Requester
correlation_id = str(uuid.uuid4())
response_topic = f"responses/{client_id}/{correlation_id}"

# Subscribe to response topic first
client.subscribe(response_topic, qos=1)

# Publish request with MQTT 5.0 properties
properties = mqtt.Properties(mqtt.PacketTypes.PUBLISH)
properties.ResponseTopic = response_topic
properties.CorrelationData = correlation_id.encode()

client.publish("services/temperature/get", b"", qos=1,
               properties=properties)

# Responder
def on_request(client, userdata, msg):
    response_topic = msg.properties.ResponseTopic
    corr_data = msg.properties.CorrelationData

    result = {"temperature": read_sensor()}
    props = mqtt.Properties(mqtt.PacketTypes.PUBLISH)
    props.CorrelationData = corr_data

    client.publish(response_topic, json.dumps(result),
                   qos=1, properties=props)

Shared Subscriptions (Load Balancing)

# Multiple workers share messages from high-volume topic
# Only ONE worker in the group receives each message

# Worker 1
client1.subscribe("$share/workers/sensors/+/data", qos=1)

# Worker 2
client2.subscribe("$share/workers/sensors/+/data", qos=1)

# Worker 3
client3.subscribe("$share/workers/sensors/+/data", qos=1)

# Messages to "sensors/node-1/data" go to ONE of the three workers
# Broker handles round-robin or load-based distribution

Monitoring and Debugging

Broker Health Metrics

# Subscribe to broker system topics
mosquitto_sub -v -t '$SYS/#'

# Key metrics to monitor:
# $SYS/broker/clients/connected      - Active connections
# $SYS/broker/messages/received      - Messages per second
# $SYS/broker/messages/sent          - Messages per second
# $SYS/broker/heap/current           - Memory usage
# $SYS/broker/load/messages/+        - Load averages
# $SYS/broker/retained messages/count - Retained message count

Debug Subscriber

def debug_subscriber(broker: str, topic: str = "#"):
    """Subscribe and print all messages for debugging."""
    def on_message(client, userdata, msg):
        try:
            payload = json.loads(msg.payload.decode())
            formatted = json.dumps(payload, indent=2)
        except (json.JSONDecodeError, UnicodeDecodeError):
            formatted = msg.payload.hex()

        print(f"[{time.strftime('%H:%M:%S')}] "
              f"Topic: {msg.topic} | "
              f"QoS: {msg.qos} | "
              f"Retain: {msg.retain} | "
              f"Size: {len(msg.payload)}B")
        print(f"  Payload: {formatted}\n")

    client = mqtt.Client()
    client.on_message = on_message
    client.connect(broker)
    client.subscribe(topic, qos=0)
    client.loop_forever()

Common Pitfalls

Exercises

1. Topic Design: Design a complete topic hierarchy for a 3-floor smart building with HVAC, lighting, and occupancy sensors
2. QoS Comparison: Publish 1000 messages at each QoS level, measure delivery time, message loss, and broker load
3. Retained Config: Implement a device configuration system using retained messages where devices read config on boot
4. LWT Monitor: Build a fleet status monitor using LWT that displays online/offline status for 10 simulated devices
5. Request/Response: Implement a command-response pattern where a dashboard requests current sensor readings from specific devices

Process

1. Gather information. Ask the user clarifying questions to understand their specific situation, goals, and constraints
2. Analyze context. Review the information provided and identify key factors relevant to mqtt specialist
3. Develop recommendations. Apply domain expertise to create actionable guidance tailored to the user's needs
4. Present structured output. Deliver findings in the output format below with clear next steps
5. Address follow-ups. Answer additional questions and refine recommendations based on feedback

Output Format

## Mqtt Specialist Analysis

### Assessment
[Key findings and observations]

### Recommendations
1. [Primary recommendation]
2. [Secondary recommendation]
3. [Additional suggestions]

### Action Items
- [ ] [First action step]
- [ ] [Second action step]
- [ ] [Follow-up task]

Edge Cases

• Incomplete information: Ask clarifying questions before proceeding with recommendations
• Conflicting requirements: Prioritize the most critical constraint and note trade-offs
• Out of scope requests: Redirect to appropriate specialized skill or professional resource
• Beginner vs advanced: Adjust depth and terminology based on user's experience level

Example

Input: "Help me with mqtt specialist for my current situation"

Output:

Based on your situation, here is a structured approach to mqtt specialist:

1. Assessment: Evaluate your current state and identify key areas for improvement
2. Strategy: Develop a targeted plan based on best practices
3. Implementation: Execute the plan with specific, measurable steps
4. Review: Monitor progress and adjust as needed