DEV Community: Supun Sriyananda

Finishing What I Started — From a TODO List to a Published PyPI Package

Supun Sriyananda — Thu, 28 May 2026 13:03:03 +0000

This is a submission for the GitHub Finish-Up-A-Thon Challenge

What I Built

I built robmqtt — a resilient MQTT client for edge IoT devices, now published on PyPI.

It came out of a problem that cost me real time. I deploy sensor systems on hardware that lives in the field: Raspberry Pi units on 4G cellular, embedded gateways, battery monitoring systems. On those networks, connectivity is never stable. And the standard MQTT client, paho-mqtt, silently drops messages when the broker is unreachable — no error, no warning, no log entry unless you write one yourself.

I lost data for a long time before I understood why. And finding the cause was brutal. There are no proper logs unless you explicitly add them, so I spent hours — days — chasing a problem that left no trace. I went through forums, blogs, chat AIs, everywhere, looking for how other people had solved it. What I found was the same thing every time: people asking for a production-resilient MQTT client, and no real answer. Just scattered advice and code snippets that handled one piece and ignored the rest.

So I built it myself:

Offline queue — when the broker is unreachable, messages are written to SQLite instead of being dropped. They survive process restarts and power cycles.
Inflight tracking — messages sent but not yet acknowledged are tracked separately and re-sent on reconnect, closing a gap that even QoS 1 leaves open.
Priority eviction — when the queue fills, low-priority telemetry is evicted before critical alerts.
Exponential backoff — a fleet of devices reconnecting after an outage won't all hammer the broker at once.
TLS support — for brokers that require encrypted connections.

And then everything a library needs to actually be used in production, not just by me:

TLS and mutual TLS plus username/password auth, for secured brokers.
Bidirectional messaging — subscribe/unsubscribe with full MQTT wildcard support, and subscriptions that survive reconnects.
Observability — a built-in HTTP health check endpoint (/health returning healthy / degraded / unhealthy) with Docker HEALTHCHECK and Kubernetes liveness-probe examples.
Structured logging throughout — so the next person doesn't lose days to a problem with no trace, the way I did.
73 tests across the storage, tracking, topic-matching, and client layers.

pip install robmqtt

This matters to me because it's not a toy. It's running on real systems I maintain — battery management monitoring and robotics telemetry — where a gap in the data record has actual consequences.

Demo

PyPI: https://pypi.org/project/robmqtt/
GitHub: https://github.com/ranaweerasupun/resilient-edge-mqtt-client

The most satisfying way to see it work is to watch the offline queue survive a broker outage. The repo includes a simulation script for exactly this:

# Terminal 1 — run a simulated device publishing every 5 seconds
python test_13.py

# Terminal 2 — kill the broker mid-run
sudo systemctl stop mosquitto

The device keeps publishing. Messages start queuing to SQLite instead of erroring. Then:

# Bring the broker back
sudo systemctl start mosquitto

The queue drains automatically, in priority order. Every message that piled up during the outage is delivered. Nothing is lost.

Basic usage looks like this — the application code never has to know whether the broker is reachable:

from robmqtt import ProductionMQTTClient
import json, time

client = ProductionMQTTClient(
    client_id="field_device_001",
    broker_host="mqtt.yourdomain.com",
    broker_port=1883,
    max_queue_size=5000,
    db_path="./device.db",
)
client.connect()
client.start()

while True:
    client.publish(
        topic="sensors/temperature",
        payload=json.dumps(read_sensor()),
        qos=1,
        priority=5,
    )
    time.sleep(30)

The Comeback Story

Here's the honest before-and-after.

Before. Once I'd finally figured out the code that solved my problem, I used it — and it went into my GitHub repo, where it sat. It worked. It solved the thing that had cost me days. But no one knew it existed.

And that was the part that nagged at me. I'd keep seeing people online asking for exactly what I'd built — something resilient enough for production — and getting the same non-answers I'd gotten: random advice, half-solutions, snippets. I had the answer sitting in a corner of my repo, and a thought stuck in the corner of my head: it's right here. I need to show this off.

The things I knew I should do but hadn't:

Package it for PyPI so people could pip install it instead of cloning the repo
Add TLS/mTLS support for secure broker connections
Expose a metrics endpoint for observability

It was the classic unfinished side project. Functional, but not shipped. Not something anyone other than me could actually use. The plan was always "publish it once it's polished" — and polishing it was the part that kept not happening.

The hard part. I have a full-time job. Finishing this wasn't something I could do in work hours. So it happened on weekends, late at night, and in the gaps — I drew flowcharts and planned the package structure on my commute, in my head, on paper. Progress came in small pieces, squeezed around everything else. There were plenty of stretches where it would have been easier to just leave it in the repo and move on.

After. I pushed it over the line:

Published to PyPI as robmqtt v1.0.0. This was more work than I expected — restructuring the code into a proper installable package, writing the pyproject.toml, sorting out the module layout, testing the install in a clean environment.
Added TLS support. The client now takes use_tls, ca_certs, certfile, keyfile, and broker auth parameters, so it works with secured brokers, not just local plaintext ones.
Observability Built an HTTP health check endpoint with Docker and Kubernetes probe examples — so the library reports its own health instead of failing silently.
Structured logging so the next person doesn't suffer days to a problem with no trace, the way I did!
Data Integrity Fixed three bugs I'm glad I caught before anyone relied on it: binary payloads being corrupted by string conversion, a resend-tracking gap that could silently lose messages on a second disconnect, and a race condition in publish() that could drop a message between the connection check and the send.
Started building a real project on top of it. To put the library through its paces in a full system, I'm building an IoT fleet analytics platform around it — simulated devices publishing through robmqtt, an MQTT-to-InfluxDB bridge, Grafana dashboards, and statistical anomaly detection. It's still in progress, but it's already doing what mattered most: proving the library holds up as the foundation of a real pipeline, not just in isolated tests.

When it was finally published and I ran the test scripts and watched the offline queue fill and drain exactly the way it was supposed to — messages held through an outage, then delivered, nothing lost — there was nothing more rewarding. The thing that had cost me days of frustration was now one pip install away for anyone who hits the same wall I did.

My Experience with GitHub Copilot

I'll be honest about my workflow, because the reason I relied on Copilot is specific.

I use a few AI tools when I build things. The general-purpose chat assistants are useful for sketching out an approach or talking through a design — but when it came to actual code, they kept giving me snippets that didn't quite work. That's not surprising: they can't see my project. They don't know my file structure, my variable names, the exact version of a library I'm using, or how the piece they're suggesting fits the rest of the code. So I'd paste in a snippet, hit an error, paste the error back, get a revised snippet, hit another error. Going back and forth with a chat window that can't see my codebase got slow and frustrating.

Copilot was different because it lives in my editor and can see everything — all my open files, the actual code around the cursor, the real context. That's why it became the tool I leaned on.

Two ways I used it:

Inline completion, always on. As I typed, Copilot autocompleted the repetitive, structural parts — the device profile dictionaries, the JSON payload construction, the boilerplate around publish() calls with their QoS and priority arguments. Because it could see the patterns already in my file, its suggestions actually fit, instead of being generically plausible code I'd have to rewrite.

Copilot Chat for debugging in context. This is where it earned its place. When something broke, I'd ask it directly — "Why am I getting this error, and how can I fix it?" or "Can you check this code snippet?" — and because it could read my actual files, the answers were grounded in my real code, not a guess about what my code might look like. That's the difference that made me stop pasting snippets into external chat windows. The tool that can see your codebase gives you answers that apply to your codebase.

That freed me to spend my real thinking time on the parts that mattered: the sine-wave drift model for realistic sensor data, the priority eviction logic, and the inflight-tracking design that closes the QoS 1 gap. The hard design decisions were mine. Copilot removed the friction around them — the typing, the boilerplate, and the slow debugging loop — so I could stay focused on the actual engineering.

Built by Supun Sriyananda — R&D Engineer working on embedded and IoT systems. robmqtt is open source on GitHub and PyPI.

How to Generate Realistic IoT Sensor Data for Testing Your MQTT Pipeline

Supun Sriyananda — Thu, 28 May 2026 11:02:48 +0000

How to Generate Realistic IoT Sensor Data for Testing Your MQTT Pipeline

This is part 2 of a series on building robmqtt. Part 1 covered why paho-mqtt silently drops messages and the library I built to fix it. This part is about testing — how to exercise an MQTT pipeline without deploying physical hardware.

In the last post I wrote about robmqtt, a resilient MQTT client for edge devices. Once I had it working, I hit the obvious next problem: how do I test it properly without setting up a rack of Raspberry Pis?

I needed data flowing through the pipeline. Lots of it. From many devices. Behaving differently. And ideally surviving a broker outage so I could watch the offline queue do its job.

So I wrote a device simulator. And in writing it, I learned that generating realistic fake sensor data is harder than it looks — and that most people get it wrong in the same way.

The trap: random data isn't realistic data

The first instinct when simulating a sensor is to reach for random.uniform():

cpu = random.uniform(0, 100)   # don't do this

This produces data that looks nothing like a real sensor. Real sensor readings don't jump randomly across the whole range every second. A CPU sitting at 8% doesn't suddenly read 94% then drop to 3%. Temperature drifts slowly. Signal strength wobbles around a baseline. There's continuity from one reading to the next.

If you test your pipeline with pure random noise, your charts look like static, your anomaly detection has nothing meaningful to detect, and your dashboards are useless for spotting whether anything actually works.

I wanted data that looked like it came from a real device.

Realistic drift with a sine wave

The trick I landed on was a slow sine wave with per-device random phase, plus a little Gaussian noise on top:

def _cpu(self) -> float:
    p = self.profile
    drift = 5 * math.sin(time.time() / 300 + self._drift_phase)
    noise = random.gauss(0, p["cpu_variance"] / 2)
    value = p["cpu_base"] + drift + noise
    return round(max(1.0, min(99.0, value)), 1)

Three things are happening here:

The sine wave (math.sin(time.time() / 300 ...)) creates a slow, smooth oscillation with a period of about five minutes. This is the gradual drift you see in real systems as load rises and falls through the day.

The phase offset (self._drift_phase, a random value set once when the device starts) means every device is at a different point in its cycle. Without it, all your simulated devices would drift up and down in perfect unison, which is a dead giveaway that the data is fake.

self._drift_phase = random.uniform(0, 2 * math.pi)

The Gaussian noise (random.gauss) adds small reading-to-reading variation on top of the drift. Real sensors are never perfectly smooth — there's always measurement jitter.

The result is data that drifts, wobbles, and stays within a believable range — and each device has its own personality. When you chart it, it looks like telemetry, not like a random number generator.

Device profiles: a camera is not a sensor

A real fleet isn't 15 copies of the same device. A camera runs hot and busy. A simple sensor sips power and idles. A gateway sits in between. If your simulator treats them all identically, your test data doesn't reflect anything real.

So each device type gets a profile:

DEVICE_PROFILES = {
    "sensor": {
        "cpu_base": 8,    "cpu_variance": 6,
        "temp_base": 42.0, "temp_variance": 5.0,
        "telemetry_interval": 10,
        "failure_rate": 0.03,
    },
    "camera": {
        "cpu_base": 65,   "cpu_variance": 20,
        "temp_base": 61.0, "temp_variance": 10.0,
        "telemetry_interval": 5,
        "failure_rate": 0.02,
    },
    # gateway, controller ...
}

A camera baselines at 65% CPU and 61°C, publishing every 5 seconds. A sensor baselines at 8% CPU and 42°C, publishing every 10 seconds. When this data lands in a dashboard, the device types are visibly different — exactly like a real deployment, where you can often guess a device's role just from its resource profile.

The failure_rate field controls how often the device injects an anomaly — a sudden CPU and temperature spike — so there's something for downstream anomaly detection to actually find:

if self._is_anomaly():
    payload["cpu_percent"] = round(random.uniform(88, 99), 1)
    payload["temperature_c"] = round(random.uniform(78, 92), 2)
    payload["anomaly"] = True

Using robmqtt in the simulator

This is where the simulator doubles as a usage example for the library. Each simulated device is a real robmqtt client:

from robmqtt import ProductionMQTTClient

self.client = ProductionMQTTClient(
    client_id=f"fleet_{device_id}",
    broker_host=broker_host,
    broker_port=broker_port,
    max_queue_size=500,
    db_path=f"./data/{device_id}.db",
    min_backoff=2,
    max_backoff=30,
    log_dir=f"./logs/{device_id}",
)

self.client.connect()
self.client.start()

Each device publishes three kinds of message, and the QoS and priority differ by importance:

# Telemetry — frequent, can tolerate eviction under pressure
self.client.publish(topic=f"fleet/{id}/telemetry",
                    payload=json.dumps(payload), qos=1, priority=5)

# Status — operational health, higher priority
self.client.publish(topic=f"fleet/{id}/status",
                    payload=json.dumps(payload), qos=1, priority=8)

# Boot/alert events — must not be lost, highest priority, QoS 2
self.client.publish(topic=f"fleet/{id}/events",
                    payload=json.dumps(payload), qos=2, priority=10)

This is the priority system from part 1 in action. If the broker goes down and the offline queue fills, routine telemetry (priority 5) gets evicted before status messages (priority 8), and event messages (priority 10, QoS 2) are protected.

The status messages even report the client's own internal state, pulled straight from robmqtt:

stats = self.client.get_statistics()
payload = {
    "queue_depth":     stats.get("offline_queue_size", 0),
    "inflight_count":  stats.get("inflight_count", 0),
    "is_connected":    stats.get("is_connected", False),
    "reconnect_count": stats.get("reconnect_count", 0),
}

So the simulated fleet reports on its own connectivity health — which means you can build a dashboard that shows the offline queue filling and draining in real time.

Watching the offline queue work

This is the part I find satisfying. Start a device:

python device_simulator.py --device-id device_001 --device-type gateway

[device_001] Started — type=gateway location=warehouse_a

Now kill the broker. The device keeps publishing — but the messages are now being written to SQLite instead of sent:

sudo systemctl stop mosquitto

The device doesn't crash. It doesn't error. It just quietly queues. The queue_depth in the status payload climbs: 5, 12, 28, 45...

Bring the broker back:

sudo systemctl start mosquitto

The queue drains automatically. Every reading that piled up during the outage is delivered, in priority order. The queue_depth falls back to zero. Nothing was lost.

That's the whole point of the library, demonstrated in a way you can watch happen.

Why a simulator is worth building

Even if you have real hardware, a simulator earns its place:

It lets you test at scale you don't have hardware for. You can run 15 simulated devices on your laptop and see how your pipeline, database, and dashboards behave under fleet-level load.

It gives you reproducible failure scenarios. Killing a broker on demand is a lot easier than waiting for a real 4G connection to drop in the field.

It produces clean test data with known properties. You injected the anomalies, so you know exactly what your anomaly detection should catch.

And — as a bonus — it doubles as living documentation for how to use your client library. The simulator is the example.

What's next

In part 3 I'll cover running this at fleet scale — launching many devices at once and feeding their telemetry through an analytics pipeline into a live dashboard.

The full simulator code is on GitHub alongside robmqtt:

PyPI: pypi.org/project/robmqtt
GitHub: github.com/ranaweerasupun/resilient-edge-mqtt-client

How do you test your IoT pipelines — real hardware, simulators, or something else? I'm curious what others do — let me know in the comments.

Why Your MQTT Client Is Silently Losing Messages (And How I Fixed It) - robmqtt

Supun Sriyananda — Tue, 26 May 2026 14:01:19 +0000

Why Your MQTT Client Is Silently Losing Messages (And How I Fixed It)

I learned this the hard way.

I was building a sensor system for a field deployment — Raspberry Pi units publishing temperature and humidity data over 4G cellular to an MQTT broker. The dashboard looked fine. The graphs looked fine. Then one day I compared the raw sensor logs against what actually made it to the broker.

Thousands of readings. Gone. No errors. No warnings. Just gone.

The culprit? paho-mqtt's default behaviour when the broker is unreachable: it silently drops your message and moves on.

After losing enough data I wrote a library to fix it. It's now on PyPI as robmqtt.

pip install robmqtt

But before I show you how it works, let me show you exactly what the problem is — because it's subtler than most people realise.

The Problem with Standard MQTT Clients

When you call client.publish() in paho-mqtt and the broker is unreachable, one of two things happens:

The message is silently discarded (QoS 0)
The message is queued in memory for QoS 1/2 — but that queue is lost on process restart, and there's a second gap that even QoS 1 doesn't close

That second gap is the sneaky one. Here's what happens with QoS 1:

The message was "sent" from your perspective. It was never confirmed from the broker's perspective. And paho has no mechanism to track this gap across reconnections.

On a stable data centre network, this almost never matters. On a Raspberry Pi running on 4G cellular in a field cabinet, it happens constantly.

What a Resilient Edge Client Actually Needs

After losing enough data, I sat down and wrote out what a proper edge MQTT client needs to do:

1. Persist offline messages to disk
If the broker is unreachable when publish() is called, the message should be written to disk and replayed later. Not held in memory — memory is lost on restart.

2. Track in-flight messages separately
Messages that have been handed to the broker but not yet ACK'd need to be tracked. On reconnect, they must be re-sent before any queued messages start draining.

3. Priority-based eviction
When the queue fills up, not all messages are equal. A critical alarm should survive. A routine telemetry reading from 6 hours ago should not block it.

4. Exponential backoff on reconnect
A fleet of 50 devices coming back online after a broker restart should not all hammer the broker at the same second.

5. Thread-safe storage
The MQTT network thread and your application thread are both touching message state. This needs to be safe without forcing the caller to think about locking.

None of this is exotic. All of it is missing from the standard paho-mqtt client when used out of the box.

How robmqtt Solves It

Here's the architecture:

The Offline Queue

When the client detects it's disconnected, publish() routes to an OfflineQueue backed by SQLite:

# Simplified from offline_queue.py
class OfflineQueue:
    def enqueue(self, topic, payload, qos, priority):
        with self._lock:
            self._db.execute("""
                INSERT INTO queue (topic, payload, qos, priority, timestamp)
                VALUES (?, ?, ?, ?, ?)
            """, (topic, payload, qos, priority, time.time()))

    def dequeue_batch(self, batch_size=10):
        with self._lock:
            # Highest priority first, then oldest first within same priority
            return self._db.execute("""
                SELECT id, topic, payload, qos FROM queue
                ORDER BY priority DESC, timestamp ASC
                LIMIT ?
            """, (batch_size,)).fetchall()

SQLite gives you durability without a separate process. It survives power cycles. The threading lock means your application thread and the drain thread never step on each other.

The Inflight Tracker

This closes the gap QoS 1 leaves open:

# Simplified from inflight_tracker.py
class InflightTracker:
    def track(self, mid, topic, payload, qos):
        """Call this when you hand a message to paho."""
        with self._lock:
            self._db.execute("""
                INSERT OR REPLACE INTO inflight (mid, topic, payload, qos)
                VALUES (?, ?, ?, ?)
            """, (mid, topic, payload, qos))

    def acknowledge(self, mid):
        """Call this in on_publish callback."""
        with self._lock:
            self._db.execute("DELETE FROM inflight WHERE mid = ?", (mid,))

    def get_all_pending(self):
        """Call this on reconnect — re-send everything unacknowledged."""
        with self._lock:
            return self._db.execute(
                "SELECT topic, payload, qos FROM inflight"
            ).fetchall()

On reconnect, the client replays all inflight messages first, then starts draining the offline queue. Delivery order is preserved.

Priority Eviction

Each message gets a priority from 1 (lowest) to 10 (highest):

# Routine telemetry — can be evicted when queue is full
client.publish(
    topic="sensors/temperature",
    payload='{"value": 23.5}',
    qos=1,
    priority=3
)

# Critical alert — survives eviction, displaces old telemetry
client.publish(
    topic="alerts/critical",
    payload='{"type": "over_temp", "value": 87.2}',
    qos=2,
    priority=9
)

When the queue hits capacity, the lowest-priority messages are evicted first. Your critical alerts are never blocked by a backlog of stale routine data.

Using robmqtt

Install:

pip install robmqtt

Basic usage — this is everything you need:

from robmqtt import ProductionMQTTClient
import json

client = ProductionMQTTClient(
    client_id="field_device_001",
    broker_host="mqtt.yourdomain.com",
    broker_port=1883,
    max_queue_size=5000,    # holds ~5000 messages during outages
    min_backoff=2,          # start retrying after 2s
    max_backoff=60,         # cap retry interval at 60s
    db_path="./device.db",  # SQLite lives here — survives reboots
)

client.connect()
client.start()

# From here just call publish() — routing is handled internally.
# Connected: sends directly and tracks inflight.
# Disconnected: writes to SQLite, drains automatically on reconnect.

while True:
    reading = read_sensor()
    client.publish(
        topic="sensors/temperature",
        payload=json.dumps(reading),
        qos=1,
        priority=5,
    )
    time.sleep(30)

The application code doesn't need to know whether the broker is reachable. That's the point.

Check what's happening at runtime:

stats = client.get_statistics()
print(stats)
# {
#   'is_connected': True,
#   'offline_queue_size': 0,
#   'inflight_count': 2,
#   'reconnect_count': 4,
#   ...
# }

TLS is supported if your broker requires it:

client = ProductionMQTTClient(
    client_id="secure_device_001",
    broker_host="mqtt.yourdomain.com",
    broker_port=8883,
    use_tls=True,
    ca_certs="/etc/ssl/certs/broker-ca.crt",
    username="device001",
    password="your_password",
)

Seeing It in Action

The repo includes test_13.py, a simulation designed specifically to demo the offline behaviour:

# Terminal 1 — run the simulation (publishes every 5 seconds)
python test_13.py

# Terminal 2 — simulate a network outage
sudo systemctl stop mosquitto

# Watch messages queue up in Terminal 1
# Queue stats print every 10 readings

# Restore connectivity
sudo systemctl start mosquitto

# Watch the offline queue drain automatically — zero messages lost

The queue drain happens in a background daemon thread. Your application code does nothing. It just works.

Real-World Context

I've deployed this pattern on:

Battery management systems — monitoring cell voltages and temperatures in production energy storage systems. A 10-minute broker outage during a network switch should not cause a gap in the battery health record.
Robotics telemetry — ROS2 robots publishing sensor and status data. Process restarts during OTA updates should not lose the last known state.
MQTT edge gateways — aggregating data from downstream sensors over serial or CAN and forwarding to a cloud broker over 4G. The gateway may reconnect dozens of times per day.

In all of these cases, the pattern is the same: treat disconnection as normal, not exceptional. Design the client to buffer, not to fail.

Who This Is For

robmqtt is specifically for edge device deployments where:

Network connectivity is unreliable (cellular, Wi-Fi roaming, VPNs)
Process restarts happen (watchdog resets, power cycles, OTA updates)
Message loss has real consequences (industrial monitoring, remote sensors, fleet telemetry)
You don't want to build and maintain this infrastructure yourself

If you're running MQTT on a stable cloud-to-cloud connection, paho-mqtt alone is probably fine. If you're deploying on Raspberry Pi, industrial gateways, field sensors, or anything running on 4G/LTE — this is for you.

What's Next

The Prometheus metrics endpoint is on the roadmap. The structured logging already writes .jsonl metrics files — exposing them via HTTP is a small step and would make robmqtt slot naturally into standard observability stacks.

If you try it and hit an issue, open a GitHub issue. If you want a feature, open a discussion.

PyPI: pypi.org/project/robmqtt
GitHub: github.com/ranaweerasupun/resilient-edge-mqtt-client

Have you run into MQTT message loss on edge devices? How did you handle it — drop a comment below.