Reference library

Streaming & messaging

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

53 matches

Mock Watermark Late Event Side Output in Python

Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.

from datetime import datetime, timedelta
from typing import List, Tuple


def watermark_mock(
    events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
    """Simulate watermarking: events arriving on time vs. late by ch…

11 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 medium

Simulate RabbitMQ QoS Prefetch Count in Python

Mocks RabbitMQ QoS prefetch semantics using threading and a queue to cap concurrent unacked message processing per worker.

rabbitmq threading qos

Python

import threading
import time
import queue


class RabbitMQMock:
    def __init__(self, prefetch_count=1):
        self.prefetch_count = prefetch_count
        self.channel_queue = queue.Queue()
        self.currently_processing = 0
        self.lock = threading.Lock()

    def start_consuming(self, messages, worker_co…

13 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.