Single-node · In-memory · gRPC

Heads up: I ended up switching to NATS + PostgreSQL. If you need consumer groups, bare NATS won't help — you need JetStream, which requires persistence. tinybroker fills the gap when you want in-memory consumer groups with no persistence overhead. The code works and is free to use; just know why you're reaching for it.

Python

Dependencies

pip install grpcio grpcio-tools

Generate stubs:

python -m grpc_tools.protoc \
  -I proto \
  --python_out=gen \
  --grpc_python_out=gen \
  tinybroker/v1/service.proto \
  tinybroker/v1/client.proto \
  tinybroker/v1/server.proto \
  tinybroker/v1/shared.proto

Connect

import grpc
from gen.tinybroker.v1 import service_pb2_grpc

channel = grpc.insecure_channel("tinybroker:50051")
stub = service_pb2_grpc.BrokerStub(channel)

Publish

from gen.tinybroker.v1 import client_pb2

stub.Publish(client_pb2.PublishRequest(
    topic="events.user.signup",
    payload=b'{"user_id":"abc123"}',
))

Subscribe (fan-out)

Subscriptions use a request iterator — a generator that yields SubscribeCommand messages:

from gen.tinybroker.v1 import client_pb2, shared_pb2

def subscribe_commands():
    yield client_pb2.SubscribeCommand(
        open=client_pb2.OpenSubscription(
            topic_patterns=["events.user.*", "events.order.*"],
            mode=shared_pb2.DELIVERY_MODE_MULTI,
        )
    )
    # The generator stays alive; yield more commands as needed.
    # Block here until the subscription should end:
    import threading
    threading.Event().wait()

subscription_id = None
for event in stub.Subscribe(subscribe_commands()):
    if event.HasField("subscription_id"):
        subscription_id = event.subscription_id
    elif event.HasField("message"):
        msg = event.message
        print(f"topic={msg.topic} payload={msg.payload}")

Subscribe (consumer group)

from gen.tinybroker.v1 import server_pb2

def worker_commands():
    yield client_pb2.SubscribeCommand(
        open=client_pb2.OpenSubscription(
            subscription_id="job-workers",
            topic_patterns=["jobs.*"],
            mode=shared_pb2.DELIVERY_MODE_BALANCED,
            consumer_group="job-workers",
        )
    )
    threading.Event().wait()

subscription_id = None
for event in stub.Subscribe(worker_commands()):
    if event.HasField("subscription_id"):
        subscription_id = event.subscription_id
    elif event.HasField("message"):
        msg = event.message
        process(msg.payload)
        stub.Ack(client_pb2.AckRequest(
            subscription_id=subscription_id,
            system_id=msg.system_id,
        ))

Dynamic pattern management with a queue

Use a queue.Queue to push additional commands from other threads:

import queue, threading

cmd_queue = queue.Queue()

def commands():
    # Initial open
    yield client_pb2.SubscribeCommand(
        open=client_pb2.OpenSubscription(topic_patterns=["events.*"]))
    # Subsequent commands from the queue
    while True:
        yield cmd_queue.get()

# From another thread, add a pattern:
cmd_queue.put(client_pb2.SubscribeCommand(
    add_patterns=client_pb2.AddTopicPatterns(topic_patterns=["alerts.*"])
))

Async (asyncio + grpcio-aio)

import grpc.aio

async def main():
    async with grpc.aio.insecure_channel("tinybroker:50051") as channel:
        stub = service_pb2_grpc.BrokerStub(channel)

        async def commands():
            yield client_pb2.SubscribeCommand(
                open=client_pb2.OpenSubscription(topic_patterns=["events.*"]))
            await asyncio.sleep(3600)

        async for event in stub.Subscribe(commands()):
            if event.HasField("message"):
                print(event.message.topic, event.message.payload)

Reconnect pattern

import time

def subscribe_with_reconnect():
    while True:
        try:
            for event in stub.Subscribe(subscribe_commands()):
                handle(event)
        except grpc.RpcError as e:
            print(f"disconnected: {e.code()} — retrying in 2s")
            time.sleep(2)