Streaming &amp; messaging

import random
import threading
import time
from collections import deque


class MockBatchProcessor:
    def __init__(self, process_func, commit_func, batch_size=5):
        self.queue = deque()
        self.batch_size = batch_size
        self.process_func = process_func
        self.commit_func = commit_func

    de…

14 0 Open

How to Implement At-Least-Once Delivery with Acknowledgment in Python

This code demonstrates a mock message broker with at-least-once delivery, including retry logic and acknowledgment after successful processing.

messaging queue retry

import time
import uuid
from collections import deque


class MockMessageBroker:
    def __init__(self):
        self.queue = deque()
        self.acked = set()

    def publish(self, payload: str) -> str:
        msg_id = str(uuid.uuid4())
        self.queue.append((msg_id, payload))
        return msg_id

    def po…

session-window streaming timeout

How to Track Session Windows with Gap Timeout in Python

A Python class that groups events into sessions, closing a session when the gap between events exceeds a timeout threshold.

import time

class SessionWindow:
    """Track sessions with a gap timeout (mock)."""
    
    def __init__(self, timeout_seconds=5):
        self.timeout = timeout_seconds
        self.session_start = None
        self.last_event_time = None
        self.event_count = 0
        self.events = []
    
    def add_event…

cqrs projector read-model

How to mock a CQRS projector read model update in Python

Build a CQRS projector class that maintains denormalized read models by applying domain events in a mock order-processing service.

from dataclasses import dataclass, field
from typing import Dict, List, Optional


@dataclass
class OrderReadModel:
    order_id: str
    customer_name: str
    total: float
    status: str = "pending"
    items: List[Dict] = field(default_factory=list)

    def apply_event(self, event_type: str, payload: Dict) -> Non…

11 0 Open

Kafka Consumer Poll Loop Mock in Python

Simulate a Kafka consumer poll loop with a mock class, process messages in batches, and commit offsets to understand streaming consumption patterns.

kafka streaming mock

import time

class MockKafkaConsumer:
    def __init__(self, topic, messages):
        self.topic = topic
        self.messages = list(messages)
        self.position = 0

    def poll(self, timeout_ms=100):
        if self.position >= len(self.messages):
            time.sleep(timeout_ms / 1000)
            return []…

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

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…