Reference library

Streaming & messaging

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

5 matches

At Most Once Fire-and-Forget Mock in Python

A Python mock that enforces send() is called at most once and records the arguments for verification.

class FireForgetMock:
    def __init__(self):
        self._calls = 0
        self._last_args = None
        self._last_kwargs = None

    def send(self, *args, **kwargs):
        if self._calls > 0:
            raise RuntimeError("send() called more than once")
        self._calls += 1
        self._last_args = args
…

15 0 Open

Streaming & messaging easy

Build a Streaming Messaging Helper in Python

Create a simple message stream class that stores recent messages, sends user messages, and retrieves history or latest messages with timestamps.

streaming deque dataclass

Python

from collections import deque
from dataclasses import dataclass
from datetime import datetime
import time


@dataclass
class Message:
    user: str
    text: str
    timestamp: str = ""

    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.now().strftime("%H:%M:%S")


class…

13 0 Open

Streaming & messaging easy

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 medium

How to Implement Backpressure Pause Producer with a Bounded Queue in Python

Places a Producer thread that sends items into a bounded queue with backpressure: on Full, it pauses to let the consumer catch up.

queues backpressure threading

Python

import threading
import time
import queue
import random


class Producer:
    def __init__(self, q):
        self.q = q
        self.running = True

    def produce(self):
        while self.running:
            item = random.randint(1, 100)
            try:
                self.q.put(item, timeout=0.5)
              …

14 0 Open

Streaming & messaging medium

How to Mock a Kafka Producer Batch Send in Python

Simulate a Kafka producer in Python that sends batched JSON events with mock partitions and latency for testing streaming pipelines without a real broker.

kafka mock streaming

Python

import json
import random
import time
from datetime import datetime


class MockKafkaProducer:
    def __init__(self, topic):
        self.topic = topic
        self.sent_messages = []

    def send(self, value, key=None):
        message = {
            "topic": self.topic,
            "key": key,
            "value"…

13 0 Open

Browse by section

Each section groups closely related Python snippets.