MQTT Support

// MQTT protocol level (mqtt.go:59)
const mqttProtoLevel = byte(0x4)  // MQTT 3.1.1

CONNECT     - Client connection request
CONNACK     - Connection acknowledgment
PUBLISH     - Publish message
PUBACK      - QoS 1 acknowledgment
PUBREC      - QoS 2 delivery part 1
PUBREL      - QoS 2 delivery part 2
PUBCOMP     - QoS 2 delivery part 3
SUBSCRIBE   - Subscribe to topics
SUBACK      - Subscribe acknowledgment
UNSUBSCRIBE - Unsubscribe from topics
UNSUBACK    - Unsubscribe acknowledgment
PINGREQ     - Keepalive ping
PINGRESP    - Ping response
DISCONNECT  - Clean disconnect

MQTT Topic         NATS Subject
──────────────     ──────────────
sensors/temp       sensors.temp
device/+/status    device.*.status
alerts/#           alerts.>

mqtt {
    # Listen on standard MQTT port
    port: 1883
    
    # TLS configuration (optional)
    # tls {
    #     cert_file: "/path/to/cert.pem"
    #     key_file: "/path/to/key.pem"
    # }
}

# JetStream required for QoS 1 and 2
jetstream {
    store_dir: "/var/lib/nats/jetstream"
}

mqtt {
    port: 1883
    
    # Authentication
    # Supports users, tokens, or account authentication
    
    # TLS configuration
    tls {
        cert_file: "/path/to/mqtt-cert.pem"
        key_file: "/path/to/mqtt-key.pem"
        ca_file: "/path/to/ca.pem"
        verify: true
    }
    
    # QoS settings
    ack_wait: "30s"              # Time to wait for QoS acks
    max_ack_pending: 1024        # Max unacknowledged messages
    
    # Session settings
    # Note: Session persistence uses JetStream
    
    # WebSocket support (see websocket section)
    # Clients connect to /mqtt on the WebSocket port
}

jetstream {
    store_dir: "/var/lib/nats/jetstream"
    max_memory_store: 1GB
    max_file_store: 10GB
}

websocket {
    port: 8080
    no_tls: true
}

mqtt {
    port: 1883
}

import paho.mqtt.client as mqtt

# Connect MQTT device to NATS
client = mqtt.Client("sensor-001")
client.connect("nats-server.local", 1883, 60)

# Publish sensor data (QoS 1 for reliability)
client.publish("sensors/temperature", "22.5", qos=1)

# Subscribe to commands
def on_message(client, userdata, msg):
    print(f"Command: {msg.payload.decode()}")

client.on_message = on_message
client.subscribe("devices/sensor-001/commands", qos=1)
client.loop_forever()

// NATS application receives MQTT sensor data
nc, _ := nats.Connect("nats://localhost:4222")

// Subscribe to MQTT sensor topics
nc.Subscribe("sensors.>", func(msg *nats.Msg) {
    // Process data from MQTT devices
    fmt.Printf("Sensor data: %s from %s\n", msg.Data, msg.Subject)
    
    // Send commands back to MQTT devices
    nc.Publish("devices.sensor-001.commands", []byte("RESET"))
})

# Publish with retain flag
client.publish("status/service", "online", retain=True)

# New subscribers immediately receive the retained message
client.subscribe("status/#")  # Gets "online" immediately

client.will_set(
    topic="status/sensor-001",
    payload="offline",
    qos=1,
    retain=True
)

client.connect("nats-server.local", 1883)
# Will message sent automatically on unexpected disconnect

# Clean session (default)
client = mqtt.Client("device-001", clean_session=True)
# Session state discarded on disconnect

# Persistent session
client = mqtt.Client("device-001", clean_session=False)
# Subscriptions and QoS messages persist across reconnects

curl http://localhost:8222/connz?mqtt_client=sensor-001

# List all MQTT connections
curl http://localhost:8222/connz | jq '.connections[] | select(.mqtt_client)'

curl http://localhost:8222/jsz?accounts=true | \
  jq '.account_details[].stream_detail[] | select(.name | startswith("$MQTT"))'