Reference library

Streaming & messaging

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

30 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 hard

Mock Protobuf Binary Encoding in Python

Demonstrates a minimal protobuf-like binary encoding and decoding of an event dataclass using varints and length-delimited fields in pure Python.

protobuf binary-encoding varint

Python

import struct
from dataclasses import dataclass


@dataclass
class Event:
    id: int
    user_id: int
    action: str

    def encode(self) -> bytes:
        # Mock protobuf-like binary encoding using varint and length-delimited fields
        buf = bytearray()
        # field 1: varint id (tag = (1 << 3) | 0 = 8)
  …

11 0 Open

Streaming & messaging medium

Mock Redis Streams XADD and XREAD in Python

A pure-Python mock of Redis streams that implements basic XADD, XREAD, and XLEN behavior for local testing without a real Redis server.

redis streams mocking

Python

import redis
import time
import threading


class MockRedisStreams:
    def __init__(self):
        self.streams = {}

    def xadd(self, stream_name, fields):
        if stream_name not in self.streams:
            self.streams[stream_name] = []
        entry_id = f"{time.time_ns()}-{len(self.streams[stream_name])}"
…

13 0 Open

Streaming & messaging medium

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.

watermark streaming side output

Python

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

Browse by section

Each section groups closely related Python snippets.