Streaming &amp; messaging

import json
from collections import deque


class Message:
    def __init__(self, message_id, payload, attempts=0):
        self.message_id = message_id
        self.payload = payload
        self.attempts = attempts

    def __repr__(self):
        return f"Message(id={self.message_id}, attempts={self.attempts})"


c…

16 0 Open

How to Build a Mock Change Data Capture Event Stream in Python

Generate a deterministic list of mock CDC events with event IDs, stream positions, payloads, and timestamps for testing streaming pipelines.

cdc mock event-stream

from itertools import count
from random import choice, randint, seed
from datetime import datetime, timedelta

seed(42)  # Make output deterministic
event_types = ["INSERT", "UPDATE", "DELETE"]
table_names = ["users", "orders", "products", "payments"]
counter = count(1)

def mock_cdc_event(stream_index: int) -> dict:
…

12 0 Open

How to Implement Publish-Subscribe Fanout with Multiple Subscribers in Python

Create a simple publish-subscribe system in Python that broadcasts messages to multiple subscriber callbacks for a given topic.

pubsub messaging events

import time

class PubSub:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, topic, callback):
        if topic not in self.subscribers:
            self.subscribers[topic] = []
        self.subscribers[topic].append(callback)

    def publish(self, topic, message):
        if topic in sel…

How to Mock MQTT Topic Subscriptions with QoS in Python

Build a lightweight MQTT client mock that tracks topic subscriptions with QoS levels and simulates wildcard message delivery.

mqtt mock qos

import time
from collections import defaultdict

class MockMQTTClient:
    def __init__(self):
        self.subscriptions = defaultdict(list)
        self.messages = []
    
    def subscribe(self, topic, qos=0):
        self.subscriptions[topic].append(qos)
        print(f"Subscribed to '{topic}' with QoS {qos}")
   …

deduplication set messaging

How to deduplicate messages by ID in Python

Track seen message IDs in a set to skip duplicate messages and store unique content in a dict, with exact output showing which messages were added or skipped.

import time

class MessageStore:
    def __init__(self):
        self.seen_ids = set()
        self.messages = {}
    
    def add(self, message_id, content, timestamp=None):
        timestamp = timestamp or time.time()
        if message_id in self.seen_ids:
            return False
        self.seen_ids.add(message_…

12 0 Open

How to mock RabbitMQ queue binding with routing keys in Python

A mock demonstration of binding a queue to an exchange with multiple routing keys in RabbitMQ using Python and pika, without a real broker connection.

rabbitmq messaging pika

import pika
import sys


def bind_queue_with_routing(channel, queue_name, exchange_name, routing_keys):
    """
    Mock RabbitMQ queue binding with routing keys.
    Prints the binding configuration instead of connecting to a real broker.
    """
    for routing_key in routing_keys:
        binding = {
            "q…

Using the retained message flag in MQTT with Python

This script subscribes to an MQTT topic and prints the retained flag for each received message, demonstrating how to distinguish retained messages from normal ones.

mqtt paho-mqtt iot

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    # Subscribe to a topic and check retained flag
    client.subscribe("test/retained")
    print("Subscribed to test/retained")

def on_message(client, userdata, msg):
    # msg.retain is the M…