Reference library

Streaming & messaging

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

4 matches

How to Build a Message Stream Queue in Python

A beginner-friendly MessageStream class built on deque that sends messages one at a time, tracks unread counts, and records sent items.

queue deque streaming

Python

from collections import deque
import time


class MessageStream:
    def __init__(self, messages):
        self._queue = deque(messages)
        self._sent = []

    def send_next(self):
        if not self._queue:
            return None
        message = self._queue.popleft()
        self._sent.append(message)
     …

13 0 Open

Streaming & messaging easy

How to Implement a Tumbling Window Counter in Python

Count events that fall within a fixed-size sliding time window using a deque and pruning logic.

streaming window aggregation

Python

from collections import deque
import time


class TumblingWindowCounter:
    def __init__(self, window_size_seconds):
        self.window_size = window_size_seconds
        self.window = deque()

    def add_event(self, timestamp):
        self.window.append(timestamp)

    def count(self, current_time):
        while…

14 0 Open

Streaming & messaging easy

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

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.