Reference library

Streaming & messaging

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

5 matches

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

Python

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…

13 0 Open

Streaming & messaging medium

How to Mock Offset Commit Auto vs Manual in Python

Demonstrates a Kafka-style offset commit function with auto/manual modes and tests it using unittest.mock.patch.

unittest mocking kafka

Python

from unittest.mock import Mock, patch

def commit_offsets(topic_partition_offsets, auto_commit=False):
    """Manually commit offsets or simulate auto-commit."""
    if auto_commit:
        print(f"Auto-committing offsets: {topic_partition_offsets}")
        return {"status": "auto_committed"}
    
    print(f"Manuall…

15 0 Open

Streaming & messaging easy

How to mock RabbitMQ queue binding with routing keys in Python

A mock demonstration of binding a queue to an exchange with multiple routing keys in RabbitMQ using Python and pika, without a real broker connection.

rabbitmq messaging pika

Python

import pika
import sys


def bind_queue_with_routing(channel, queue_name, exchange_name, routing_keys):
    """
    Mock RabbitMQ queue binding with routing keys.
    Prints the binding configuration instead of connecting to a real broker.
    """
    for routing_key in routing_keys:
        binding = {
            "q…

14 0 Open

Streaming & messaging hard

Mock Protobuf Binary Encoding in Python

Demonstrates a minimal protobuf-like binary encoding and decoding of an event dataclass using varints and length-delimited fields in pure Python.

protobuf binary-encoding varint

Python

import struct
from dataclasses import dataclass


@dataclass
class Event:
    id: int
    user_id: int
    action: str

    def encode(self) -> bytes:
        # Mock protobuf-like binary encoding using varint and length-delimited fields
        buf = bytearray()
        # field 1: varint id (tag = (1 << 3) | 0 = 8)
  …

11 0 Open

Streaming & messaging easy

Using the retained message flag in MQTT with Python

This script subscribes to an MQTT topic and prints the retained flag for each received message, demonstrating how to distinguish retained messages from normal ones.

mqtt paho-mqtt iot

Python

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    # Subscribe to a topic and check retained flag
    client.subscribe("test/retained")
    print("Subscribed to test/retained")

def on_message(client, userdata, msg):
    # msg.retain is the M…

14 0 Open

Browse by section

Each section groups closely related Python snippets.