Reference library

Streaming & messaging

Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.

28 matches

Mock NATS queue group load balancing in Python

Simulates a NATS queue group where each message is delivered to exactly one subscriber using random selection with a lightweight mock.

nats queue-group messaging

Python

import random
import time
from collections import defaultdict


class MockQueueGroup:
    """Mock a NATS queue group: each message is delivered to exactly one subscriber."""

    def __init__(self, subscribers):
        self.subscribers = subscribers

    def publish(self, message):
        receiver = random.choice(se…

13 0 Open

Streaming & messaging easy

Redis Pub/Sub Channel Subscribe Mock in Python

A lightweight in-memory mock of Redis pub/sub that lets you subscribe to channels, publish messages, and verify handler behavior in tests without a real Redis server.

redis pubsub testing

Python

class MockRedisPubSub:
    def __init__(self):
        self.channels = {}

    def subscribe(self, channel):
        if channel not in self.channels:
            self.channels[channel] = []
        return self.channels[channel]

    def publish(self, channel, message):
        if channel in self.channels:
            …

11 0 Open

Streaming & messaging easy

Sliding Window Average with Deque in Python

Computes the running average of a sliding window over streaming numbers using a collections.deque for O(1) pop-left operations.

sliding-window deque streaming

Python

from collections import deque

class SlidingAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.window = deque()
        self.total = 0

    def add(self, value):
        self.window.append(value)
        self.total += value
        if len(self.window) > self.window_size:
…

13 0 Open

Streaming & messaging easy

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

Python

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…

14 0 Open

Browse by section

Each section groups closely related Python snippets.