LavinMQ for IoT: from device to dashboard in minutes
Whether IoT devices go offline, networks wobble, or data arrives in bursts, your consumer shouldn’t even notice. That’s the job of the message broker, like LavinMQ. A broker sits in the middle to decouple publishers and consumers. It buffers during outages, smooths spikes, and delivers each message to every service that needs it.
LavinMQ is a lightweight message broker. You can install it in minutes, connect your IoT devices, and start publishing data right away. Consumers subscribe at their own pace, dashboards update in real time, and backends process messages reliably without being overloaded.
Why brokers matter for IoT
IoT devices lose power, roam out of coverage, reboot for updates, or sleep to save battery. In short, devices go offline—a lot. When they reconnect, traffic often arrives in bursts: thousands of sensors may wake up together, or alarms may flood the system at once.
Without a broker, these messages either vanish or overwhelm backends. A broker solves this by: - Decoupling devices and services so senders don’t depend on receivers being online - Persisting messages until they’re acknowledged, guaranteeing delivery - Smoothing spikes with backpressure (prefetch limits, pacing) - Fanning out one publish to many consumers (dashboards, storage, alerts) - Enforcing security boundaries (auth, TLS, per-queue permissions) - Exposing metrics so you can spot issues and recover fast
And when you need it, LavinMQ supports real-time streaming for dashboards and control loops.
Where LavinMQ fits
Think of LavinMQ as the messaging hub in your IoT system. Devices publish messages to a topic in LavinMQ, and consumers subscribe at their own pace. This decoupling is what makes the system resilient and scalable.
If you connect devices directly to consumers, any outage means lost data. With LavinMQ in the middle, both sides work independently and reliably. Here’s why that matters:
- Reliability IoT messages are often tiny but critical. A heart-rate monitor, for example, must not lose readings just because a dashboard went down. LavinMQ stores each message until it’s acknowledged. Even when devices reconnect after downtime, LavinMQ absorbs bursts and delivers data at a steady pace.
- Scalability What happens when you grow from 10 devices to 10,000—or from one consumer to a dozen? Direct connections turn into a fragile tangle. With LavinMQ, devices publish once, and any number of consumers can subscribe. Adding more consumers doesn’t add load to devices.
- Publish once, consume many times A single temperature reading can update a live dashboard, trigger alerts, and store itself in a time-series database simultaneously. Devices don’t duplicate work—LavinMQ handles the fan-out.
- Security LavinMQ enforces a clear boundary between the edge (devices) and the core (consumers). Devices can connect with TLS, authenticate, and publish only to allowed queues.
Protocols at the edge vs. at the backend
IoT systems stretch from tiny sensors at the edge to powerful services in the cloud. The same protocol may not fit everywhere. LavinMQ supports many, so you can pick the right one for each part of your stack.
MQTT for devices – Designed for constrained devices and unstable networks. Lightweight, battery-friendly, with QoS levels and retained messages.
Example, a soil-moisture sensor publishes to a topic farm/field3/soil. LavinMQ receives it, stores it, and makes it available to multiple consumers.
AMQP for backends – Durable queues, acknowledgments, consumer prefetch, and flexible routing.
Example, a temperature reading fans out to a dashboard, a time-series DB, and an anomaly detector—all at once.
MQTT/AMQP over WebSocket for dashboards – Ideal for browsers and quick consumers that need live streaming without extra libraries.
Consumer code samples
Here are some sample consumers in different languages:
Python code using the paho-mqtt client library:
import paho.mqtt.client as mqtt
BROKER = "<host-name>"
PORT = 1883 # use 8883 + TLS block if your instance requires TLS
TOPICT = "lavinmq/home/temperature "
TOPICH = " lavinmq/home/humidity"
# IMPORTANT: vhost goes after @ in the username (RabbitMQ/LavinMQ)
USERNAME = "<vhost:username>" # change vhost if needed
PASSWORD = "<password>"
def on_connect(client, userdata, flags, reason_code, properties=None):
if reason_code == 0:
print("Connected to broker.")
client.subscribe([(TOPICT, 1), (TOPICH, 1)])
print(f"Subscribed to: {TOPICT}, {TOPICH}")
else:
print(f"Connect failed. reason_code={reason_code}")
def on_message(client, userdata, msg):
print(f"[{msg.topic}] {msg.payload.decode('utf-8', errors='replace')}")
def on_disconnect(client, userdata, disconnect_flags, reason_code, properties=None):
print(f"Disconnected. reason_code={reason_code}, flags={disconnect_flags}")
client = mqtt.Client(
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
protocol=mqtt.MQTTv311
)
client.username_pw_set(USERNAME, PASSWORD)
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=30)
client.connect(BROKER, PORT, keepalive=60)
client.loop_forever()
C# code using the .NET MQTTnet client library:
using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Client.Options;
using MQTTnet.Protocol;
class Program
{
// ---- Config ----
private const string BROKER = "<broker-name>"; // e.g., "mqtt.example.com"
private const int PORT = 1883; // 1883 (plain) or 8883 (TLS)
private const string TOPICT = "lavinmq/home/temperature";
private const string TOPICH = "lavinmq/home/humidity";
// IMPORTANT: vhost goes after @ in the username (RabbitMQ/LavinMQ)
private const string USERNAME = "<vhost:username>"; // e.g., "sensor-user@/"
private const string PASSWORD = "<password>";
static async Task Main()
{
var factory = new MqttFactory();
using var client = factory.CreateMqttClient();
// Handlers
client.UseConnectedHandler(async e =>
{
Console.WriteLine("Connected to broker.");
// Subscribe to both topics with QoS 1 (AtLeastOnce)
var subResult = await client.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
.WithTopicFilter(f => f.WithTopic(TOPICT).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
.WithTopicFilter(f => f.WithTopic(TOPICH).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
.Build());
Console.WriteLine($"Subscribed to: {TOPICT}, {TOPICH}");
});
client.UseDisconnectedHandler(async e =>
{
Console.WriteLine($"Disconnected. Reason: {e.Reason}");
// Simple backoff reconnect (1s -> 5s)
await Task.Delay(TimeSpan.FromSeconds(1));
try
{
await client.ConnectAsync(BuildOptions(), CancellationToken.None);
}
catch (Exception ex)
{
Console.WriteLine($"Reconnect failed: {ex.Message}");
await Task.Delay(TimeSpan.FromSeconds(5));
}
});
client.UseApplicationMessageReceivedHandler(e =>
{
var topic = e.ApplicationMessage.Topic;
var payload = e.ApplicationMessage.Payload == null
? string.Empty
: Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
Console.WriteLine($"[{topic}] {payload}");
});
// Connect (keepalive ~ 60s)
var options = BuildOptions();
try
{
await client.ConnectAsync(options, CancellationToken.None);
}
catch (Exception ex)
{
Console.WriteLine($"Connect failed: {ex.Message}");
}
// Keep the process alive like loop_forever()
await Task.Delay(Timeout.InfiniteTimeSpan);
}
private static IMqttClientOptions BuildOptions()
{
// MQTT v3.1.1 with keepalive 60s and clean session (like your Python sample)
return new MqttClientOptionsBuilder()
.WithClientId($"csharp-subscriber-{Guid.NewGuid():N}")
.WithTcpServer(BROKER, PORT)
.WithCredentials(USERNAME, PASSWORD)
.WithCleanSession() // start fresh (set to false if you want durable subscriptions)
.WithKeepAlivePeriod(TimeSpan.FromSeconds(60))
.WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V311)
//.WithTls() // uncomment if using TLS on 8883
.Build();
}
}
JavaScript code using the Node.js MQTT client library
const mqtt = require("mqtt");
// --- Config ---
const BROKER = "<broker-host>"; // e.g. "mqtt.example.com"
const PORT = 1883; // 1883 (plain) or 8883 (TLS)
const TOPICT = "lavinmq/home/temperature";
const TOPICH = "lavinmq/home/humidity";
// IMPORTANT: vhost goes after @ in the username (RabbitMQ/LavinMQ)
const USERNAME = "<vhost:username>"; // e.g. "sensor-user@/"
const PASSWORD = "<password>";
// Optional: give the client a stable ID if you want durable sessions (clean:false)
const CLIENT_ID = "terminal-consumer-" + Math.random().toString(16).slice(2);
const client = mqtt.connect({
host: BROKER,
port: PORT,
protocol: PORT === 8883 ? "mqtts" : "mqtt",
username: USERNAME,
password: PASSWORD,
clientId: CLIENT_ID,
clean: true, // set to false + stable clientId for durable session
keepalive: 60,
reconnectPeriod: 1000, // auto-reconnect every 1s
// ca: fs.readFileSync("ca.pem"), // uncomment if using custom CA with TLS
});
client.on("connect", () => {
console.log("Connected to broker.");
client.subscribe(
[{ topic: TOPICT, qos: 1 }, { topic: TOPICH, qos: 1 }],
(err, granted) => {
if (err) {
console.error("Subscribe error:", err.message);
} else {
console.log(
"Subscribed:",
granted.map(g => `${g.topic} (QoS ${g.qos})`).join(", ")
);
}
}
);
});
client.on("message", (topic, payload) => {
const text = payload.toString("utf8");
const ts = new Date().toISOString();
console.log(`[${ts}] ${topic}: ${text}`);
});
client.on("reconnect", () => console.log("Reconnecting..."));
client.on("close", () => console.log("Connection closed."));
client.on("error", (err) => console.error("MQTT error:", err.message));
// Graceful exit
process.on("SIGINT", () => {
console.log("\nDisconnecting...");
client.end(true, () => process.exit(0));
});
Ruby code using the thmqtt gem
require 'mqtt'
require 'securerandom'
require 'time'
# --- Config ---
BROKER = "<broker-host>" # e.g. "mqtt.example.com"
PORT = 1883 # 1883 (plain) or 8883 (TLS)
TOPICS = { # topic => QoS
"lavinmq/home/temperature" => 1,
"lavinmq/home/humidity" => 1
}
# IMPORTANT: vhost goes after @ in the username (RabbitMQ/LavinMQ)
USERNAME = "<vhost:username>" # e.g. "sensor-user@/"
PASSWORD = "<password>"
CLIENT_ID = "terminal-consumer-#{SecureRandom.hex(4)}"
KEEPALIVE_SEC = 60
CLEAN_SESSION = true # set to false + stable CLIENT_ID for durable session
USE_TLS = (PORT == 8883)
def connect_and_stream
MQTT::Client.connect(
host: BROKER,
port: PORT,
username: USERNAME,
password: PASSWORD,
client_id: CLIENT_ID,
keep_alive: KEEPALIVE_SEC,
clean_session: CLEAN_SESSION,
ssl: USE_TLS
) do |c|
c.subscribe(TOPICS)
puts "Connected. Subscribed to: #{TOPICS.keys.join(', ')}"
c.get do |topic, message|
ts = Time.now.utc.iso8601
puts "[#{ts}] #{topic}: #{message}"
end
end
end
# Auto-reconnect with simple backoff (1s -> 30s)
backoff = 1
loop do
begin
connect_and_stream
backoff = 1 # reset if we exit normally
rescue Interrupt
puts "\nDisconnecting..."
break
rescue => e
warn "Disconnected: #{e.class} - #{e.message}"
sleep backoff
backoff = [backoff * 2, 30].min
end
end
Replace the placeholders (<broker-name>, <vhost:username>, <password>) with the values from your LavinMQ instance’s Overview → MQTT details.
Conclution
IoT devices come and go, but your data shouldn’t. LavinMQ makes sure every reading gets through—buffering outages, smoothing bursts, and fanning out messages to dashboards, alerts, and storage. It’s lightweight, easy to set up, and ready to scale with your IoT system.
Want to get hands-on? Try our real-time temperature monitoring with LavinMQ and IoT guide.