MQTT Broker Integration & Telemetry Publishing
RUSEON Core integrates a high-performance MQTT 3.1.1 / 5.0 publisher (internal/events/mqtt.go) designed for IoT, SCADA, smart city, and industrial automation platforms.
To prevent slow or unreachable MQTT brokers from causing backpressure in the core video pipeline, the MQTT publisher uses an isolated Lock-Free Ring Buffer with non-blocking enqueue semantics.
1. Topic Hierarchy & Payload Formats
RUSEON Core publishes structured JSON messages across a standardized topic tree:
| Topic Pattern | Trigger Event | Sample JSON Payload |
|---|---|---|
ruseon/cameras/{id}/status | Camera connect / disconnect | {"status":"online","fps":25.0,"bitrate_kbps":4096,"timestamp":1786802400} |
ruseon/cameras/{id}/events | Motion / Line crossing | {"event_type":"motion_detected","zone":"gate_1","score":0.88} |
ruseon/cameras/{id}/metadata | Real-time AI detections | {"pts":1786802400120000,"detections":[{"class":"car","confidence":0.96}]} |
ruseon/system/probes | Scheduled health status | {"storage_free_percent":82.4,"active_streams":48,"goroutines":112} |
2. Configuration Reference
Configure the MQTT broker connection in config.yaml:
yaml
mqtt:
enabled: true
broker: "tcp://broker.emqx.io:1883"
client_id: "ruseon_server_prod_01"
topic_prefix: "ruseon"
# Credentials
username: "ruseon_publisher"
password: "secure_password_here"
# Quality of Service (0, 1, or 2)
qos: 1
retain_status: true
# Buffer capacity for decoupling from network latency
queue_capacity: 5000
# TLS Configuration
tls:
enabled: false
ca_file: "/etc/ruseon/ca.crt"
cert_file: "/etc/ruseon/client.crt"
key_file: "/etc/ruseon/client.key"3. Lock-Free Buffer Architecture
The MQTT subsystem utilizes internal/buffer/lockfree.go:
- Lock-Free Enqueue: Frame metadata is pushed using atomic CAS operations.
- Zero Lock Contention: The main streaming goroutine never acquires a mutex when publishing telemetry.
- Overflow Protection: If the broker connection drops, the ring buffer wraps safely, discarding oldest messages without leaking memory.