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…

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…

queues backpressure threading

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.

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

rabbitmq routing messaging

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.

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

queue retry visibility-timeout

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.

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 …

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…