Reference library

Streaming & messaging

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

4 matches

How to Mock Kafka Topic Partitions with a Python dict of lists

Mocks a Kafka topic and its partitions using a defaultdict of lists to simulate message production, consumption, and per-partition counts.

kafka mock partitions

Python

from collections import defaultdict

class KafkaTopicPartitionMock:
    """A simple mock for Kafka topic-partition assignment using dict of lists."""

    def __init__(self, topic):
        self.topic = topic
        self.partitions = defaultdict(list)  # partition_id -> list of messages

    def produce(self, message…

15 0 Open

Streaming & messaging medium

How to Simulate RabbitMQ Exchange Routing in Python

Simulate RabbitMQ exchange routing using a nested dict, matching routing keys against patterns like error.* and info.# to return bound queues.

rabbitmq routing messaging

Python

from collections import defaultdict

def route_message(exchanges, exchange_name, routing_key):
    """
    Simulate RabbitMQ exchange routing using a nested dict structure.
    Returns list of queue names that match the routing key.
    """
    queues = exchanges.get(exchange_name, {})
    matched = []
    
    for pa…

14 0 Open

Streaming & messaging medium

Implement a retry queue with visibility timeout in Python

This code simulates a message queue with a visibility timeout, allowing messages to be retried if not deleted before the timeout expires.

queue retry visibility-timeout

Python

import time
from collections import deque


class SimpleQueue:
    def __init__(self, visibility_timeout=2):
        self.queue = deque()
        self.in_flight = {}
        self.visibility_timeout = visibility_timeout

    def send(self, message):
        self.queue.append(message)

    def receive(self):
        if …

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

Browse by section

Each section groups closely related Python snippets.