Routers & Publishers
Compose modular KubeMQ FastStream apps with KubeMQRouter and auto-publish handler outputs with @broker.publisher.
Overview
As a FastStream app grows, registering every handler directly on a single KubeMQBroker becomes unwieldy. kubemq-faststream ships two composition primitives that keep large apps organized:
KubeMQRoutergroups related subscribers and publishers into modular units. A router can carry aprefixthat propagates to every channel registered on it, so handler code stays short and channel namespaces stay consistent.@broker.publisher(...)stacks on top of a subscriber to automatically publish the handler's return value to another channel — no manualbroker.publish(...)call inside the handler.
Both are standard FastStream constructs; the KubeMQ adapter implements them so they behave exactly like routers and publishers on any other FastStream broker.
KubeMQRouter
KubeMQRouter is a container for handler registrations. You decorate subscribers (and publishers) on the router instead of the broker, then call broker.include_router(router) to fold those registrations into the broker at startup. This lets you split a large app into focused modules — one router per domain, per pattern, or per bounded context.
import asyncio
import logging
from faststream import FastStream
from kubemq_faststream import KubeMQBroker, KubeMQRouter
logging.basicConfig(level=logging.INFO)
router = KubeMQRouter()
@router.subscriber(events="example.router.basic.orders")
async def handle_order(msg: dict) -> None:
"""Process incoming order events."""
print(f"[Router] Order received: {msg}")
@router.subscriber(events="example.router.basic.payments")
async def handle_payment(msg: dict) -> None:
"""Process incoming payment events."""
print(f"[Router] Payment received: {msg}")
broker = KubeMQBroker("kubemq://localhost:50000")
broker.include_router(router)
app = FastStream(broker)
@app.after_startup
async def run_demo() -> None:
await broker.publish(
{"order_id": "ORD-100", "amount": 49.99},
events="example.router.basic.orders",
)
await broker.publish(
{"payment_id": "PAY-200", "status": "completed"},
events="example.router.basic.payments",
)
await asyncio.sleep(2)
await app.stop()
if __name__ == "__main__":
asyncio.run(app.run())You need a running KubeMQ broker for these examples. Start one locally with Docker — kubemq-faststream connects over the native gRPC port 50000, and port 9090 is the shared HTTP server (REST and connector endpoints, not required here):
docker run -d \ --name kubemq \ -p 50000:50000 \ -p 9090:9090 \ -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \ europe-docker.pkg.dev/kubemq/images/kubemq:nextkubemq-faststream talks to KubeMQ over native gRPC on port 50000 — there is no HTTP connector to enable. The 9090 HTTP port is exposed above only for parity with other KubeMQ tooling; FastStream does not use it.
Channel Namespacing with Prefixes
A router constructed with KubeMQRouter(prefix="orders.") prepends that prefix to every channel registered on it. Inside the handler you decorate with a short, local name; the router resolves it to the full channel at registration time. A subscriber declared with @orders.subscriber(events="created") on a router with prefix="orders." listens on the effective channel orders.created.
This keeps each module's channel names short and guarantees a consistent namespace per domain — the prefix lives in exactly one place.
import asyncio
import logging
import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker, KubeMQRouter
logging.basicConfig(level=logging.INFO)
KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)
# Router with prefix -- all channels get "orders." prepended
orders_router = KubeMQRouter(prefix="orders.")
@orders_router.subscriber(queues="new") # effective channel: orders.new
async def handle_order(msg: dict) -> None:
print(f"[orders] Received on orders.new: {msg}")
# Second router with a different prefix
payments_router = KubeMQRouter(prefix="payments.")
@payments_router.subscriber(events="received") # effective channel: payments.received
async def handle_payment(msg: dict) -> None:
print(f"[payments] Received on payments.received: {msg}")
broker.include_router(orders_router)
broker.include_router(payments_router)
@app.after_startup
async def run_demo() -> None:
await broker.publish({"order_id": 1}, queues="orders.new")
await broker.publish({"payment_id": "p-001"}, events="payments.received")
await asyncio.sleep(2)
await app.stop()
if __name__ == "__main__":
asyncio.run(app.run())The prefix applies regardless of pattern: the orders. router above namespaces a queues= subscriber, and the payments. router namespaces an events= subscriber. When you publish, you use the full resolved channel name (orders.new, payments.received) — the prefix only rewrites the subscriber registration, not your publish calls.
Nested Routers and Multi-File Apps
Routers compose. A router can include another router, and prefixes stack: a child subscriber on "placed" under a parent prefix "example.nested.orders." resolves to example.nested.orders.placed. This lets you build deep, modular hierarchies where each level adds a namespace segment.
import asyncio
import logging
from faststream import FastStream
from kubemq_faststream import KubeMQBroker, KubeMQRouter
logging.basicConfig(level=logging.INFO)
child_router = KubeMQRouter(prefix="example.nested.orders.")
@child_router.subscriber(events="placed")
async def on_order_placed(msg: dict) -> None:
"""Handle order-placed events (channel: example.nested.orders.placed)."""
print(f"[Child] Order placed: {msg}")
@child_router.subscriber(events="shipped")
async def on_order_shipped(msg: dict) -> None:
"""Handle order-shipped events (channel: example.nested.orders.shipped)."""
print(f"[Child] Order shipped: {msg}")
parent_router = KubeMQRouter()
parent_router.include_router(child_router)
broker = KubeMQBroker("kubemq://localhost:50000")
broker.include_router(parent_router)
app = FastStream(broker)
@app.after_startup
async def run_demo() -> None:
await broker.publish(
{"order_id": "N-001", "items": 3},
events="example.nested.orders.placed",
)
await broker.publish(
{"order_id": "N-001", "carrier": "DHL"},
events="example.nested.orders.shipped",
)
await asyncio.sleep(2)
await app.stop()
if __name__ == "__main__":
asyncio.run(app.run())The most common production layout puts each router in its own module and composes them from a single entry point. Define a router per concern in dedicated files:
from kubemq_faststream import KubeMQRouter
events_router = KubeMQRouter(prefix="example.multifile.events.")
@events_router.subscriber(events="notifications")
async def on_notification(msg: dict) -> None:
"""Handle notification events."""
print(f"[Events] Notification: {msg}")
@events_router.subscriber(events="alerts")
async def on_alert(msg: dict) -> None:
"""Handle alert events."""
print(f"[Events] Alert: {msg}")from kubemq_faststream import KubeMQRouter
queues_router = KubeMQRouter(prefix="example.multifile.queues.")
@queues_router.subscriber(queues="tasks")
async def process_task(msg: dict) -> None:
"""Process queued tasks with auto-ack."""
print(f"[Queues] Task processed: {msg}")
@queues_router.subscriber(queues="emails")
async def send_email(msg: dict) -> None:
"""Process email send requests."""
print(f"[Queues] Email sent to: {msg.get('to', 'unknown')}")Then main.py imports each router module and includes them on the broker:
import asyncio
import logging
from faststream import FastStream
from kubemq_faststream import KubeMQBroker
from events_router import events_router
from queues_router import queues_router
logging.basicConfig(level=logging.INFO)
broker = KubeMQBroker("kubemq://localhost:50000")
broker.include_router(events_router)
broker.include_router(queues_router)
app = FastStream(broker)
@app.after_startup
async def run_demo() -> None:
await broker.publish(
{"type": "info", "text": "System started"},
events="example.multifile.events.notifications",
)
await broker.publish(
{"task": "generate-report", "params": {"format": "pdf"}},
queues="example.multifile.queues.tasks",
)
await asyncio.sleep(3)
await app.stop()
if __name__ == "__main__":
asyncio.run(app.run())Run the multi-file app through its entry point — python examples/router/multi_file_app/main.py — so the router modules are imported and included before the broker connects.
Publisher Decorator
Stacking @broker.publisher(...) on top of a @broker.subscriber(...) turns a handler into a transform stage: whatever the handler returns is automatically published to the publisher's channel. There is no broker.publish(...) call inside the handler — the return value is the message.
The decorator order matters. The subscriber must be the innermost decorator (closest to the function); the publisher wraps it. In the example below, transform consumes from the input channel and its return value is auto-published to OUTPUT_CHANNEL, where a second subscriber picks it up.
import asyncio
import logging
from faststream import FastStream
from kubemq_faststream import KubeMQBroker
logging.basicConfig(level=logging.INFO)
broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)
INPUT_CHANNEL = "example.publisher.basic.input"
OUTPUT_CHANNEL = "example.publisher.basic.output"
@broker.subscriber(events=OUTPUT_CHANNEL)
async def receive_processed(msg: dict) -> None:
"""Consume the auto-published output."""
print(f"[Output] Processed message received: {msg}")
@broker.publisher(events=OUTPUT_CHANNEL)
@broker.subscriber(events=INPUT_CHANNEL)
async def transform(msg: dict) -> dict:
"""Transform input and auto-publish the return value to the output channel."""
print(f"[Transform] Input: {msg}")
return {"original": msg, "processed": True, "uppercase_name": msg.get("name", "").upper()}
@app.after_startup
async def run_demo() -> None:
await broker.publish(
{"name": "widget", "quantity": 5},
events=INPUT_CHANNEL,
)
await asyncio.sleep(2)
await app.stop()
if __name__ == "__main__":
asyncio.run(app.run())Publisher Headers and Chaining
Create a publisher object up front with broker.publisher(...) and assign its headers attribute to attach a default headers dict to every message it auto-publishes. This is the place to stamp metadata — source, version, content-type — onto all outgoing messages from that stage. Apply the publisher to a handler by using it as a decorator (@output_publisher).
import asyncio
import logging
import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker
logging.basicConfig(level=logging.INFO)
KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)
INPUT_CHANNEL = "example.publisher.headers.input"
OUTPUT_CHANNEL = "example.publisher.headers.output"
# Create the publisher and set default headers
output_publisher = broker.publisher(events=OUTPUT_CHANNEL)
output_publisher.headers = {
"x-source": "header-demo",
"x-version": "1.0",
"x-content-type": "application/json",
}
@broker.subscriber(events=OUTPUT_CHANNEL)
async def receive_output(msg: dict) -> None:
"""Consume messages that were auto-published with headers."""
print(f"[output] Received message: {msg}")
@output_publisher
@broker.subscriber(events=INPUT_CHANNEL)
async def transform(msg: dict) -> dict:
"""Transform input and auto-publish to the output channel with headers."""
print(f"[transform] Processing: {msg}")
return {**msg, "processed": True}To fan a single handler's output out to several destinations, stack multiple @broker.publisher(...) decorators. The return value is published once to each publisher's channel, so two downstream consumers — analytics and archive, say — each receive a copy.
ANALYTICS_CHANNEL = "example.publisher.chain.analytics"
ARCHIVE_CHANNEL = "example.publisher.chain.archive"
@broker.publisher(events=ANALYTICS_CHANNEL)
@broker.publisher(events=ARCHIVE_CHANNEL)
@broker.subscriber(events=INPUT_CHANNEL)
async def transform(msg: dict) -> dict:
"""Process input and auto-publish to both analytics and archive channels."""
print(f"[transform] Processing: {msg}")
return {**msg, "processed": True}Cross-Pattern and Dynamic Publishing
A single app can publish to any KubeMQ pattern — the keyword on broker.publish(...) selects the target. Use events= for fire-and-forget broadcast and queues= for reliable point-to-point delivery, side by side, with the same payload.
EVENT_CHANNEL = "example.publisher.pattern.notifications"
QUEUE_CHANNEL = "example.publisher.pattern.tasks"
@app.after_startup
async def run_demo() -> None:
payload = {"action": "user_signup", "user_id": "U-42"}
await broker.publish(payload, events=EVENT_CHANNEL)
print("Published to events pattern (fire-and-forget)")
await broker.publish(payload, queues=QUEUE_CHANNEL)
print("Published to queues pattern (reliable delivery)")Channel names do not have to be static. Compute the target channel at runtime from message content — for example, route an order to a region-specific channel based on a field in the payload.
@app.after_startup
async def run_demo() -> None:
orders = [
{"order_id": "ORD-1", "region": "us", "total": 29.99},
{"order_id": "ORD-2", "region": "eu", "total": 45.00},
{"order_id": "ORD-3", "region": "us", "total": 12.50},
]
for order in orders:
channel = f"example.publisher.dynamic.{order['region']}"
await broker.publish(order, events=channel)
print(f"Routed order {order['order_id']} to {channel}")For full control over routing metadata, build a KubeMQPublishCommand directly. It is the wire-level alternative to broker.publish(...), exposing pattern, headers, a metadata string, correlation ID, and a custom message ID. Publish it through the broker's producer.
from kubemq_faststream import KubeMQBroker, KubeMQPattern, KubeMQPublishCommand
CHANNEL = "example.publisher.command"
@app.after_startup
async def run_demo() -> None:
cmd = KubeMQPublishCommand(
{"action": "deploy", "version": "2.1.0"},
destination=CHANNEL,
pattern=KubeMQPattern.EVENTS,
metadata="deployment-event",
headers={"env": "staging", "region": "us-east-1"},
correlation_id="deploy-corr-001",
message_id="custom-msg-001",
)
if broker.config.producer is None:
raise RuntimeError("Broker not connected")
await broker.config.producer.publish(cmd)
print("Published via KubeMQPublishCommand with custom metadata and headers")KubeMQPublishCommand is a low-level escape hatch. For everyday publishing — including auto-publish via @broker.publisher — broker.publish(...) and the publisher decorators are the idiomatic API.
AsyncAPI Metadata
FastStream generates an AsyncAPI schema from your subscribers and publishers. Both @broker.subscriber(...) and @broker.publisher(...) accept three documentation parameters:
| Parameter | Purpose |
|---|---|
title | Human-readable name for the channel in the generated schema |
description | Longer explanation shown in the AsyncAPI docs |
include_in_schema | Set False to hide an internal channel from the public schema |
The example below documents the public output channel and the input intake, while hiding an internal audit channel from the schema with include_in_schema=False.
INPUT_CHANNEL = "example.config.asyncapi.orders"
OUTPUT_CHANNEL = "example.config.asyncapi.processed"
AUDIT_CHANNEL = "example.config.asyncapi.audit"
# Subscriber with full AsyncAPI metadata -- visible in docs
@broker.subscriber(
events=OUTPUT_CHANNEL,
title="Processed Orders",
description="Receives enriched/processed order data",
include_in_schema=True,
)
async def processed_handler(msg: dict) -> None:
"""Consume processed orders (documented in AsyncAPI schema)."""
print(f"[processed] Enriched order: {msg}")
# Publisher with AsyncAPI metadata, publishing to documented and internal channels
@broker.publisher(
events=OUTPUT_CHANNEL,
title="Order Processing Output",
description="Publishes enriched orders after processing",
include_in_schema=True,
)
@broker.publisher(
events=AUDIT_CHANNEL,
title="Audit Publisher",
description="Internal audit log publisher",
include_in_schema=False,
)
@broker.subscriber(
events=INPUT_CHANNEL,
title="Order Intake",
description="Receives raw orders for processing",
include_in_schema=True,
)
async def process_order(msg: dict) -> dict:
"""Process the order, auto-publish to output and audit channels."""
print(f"[public] Received order: {msg}")
return {**msg, "processed": True}Setting include_in_schema=False on the audit publisher keeps the runtime behavior identical — the message is still published — but omits the channel from the generated documentation, which is the right default for internal-only channels.
Related
Events (Pub/Sub)
Fire-and-forget broadcast to all subscribers, with optional group load balancing.
Events Store
Persistent pub/sub with replay from first, at a sequence, or by time.
Queues
Point-to-point messaging with AckPolicy-controlled settlement and batch publishing.
Commands & Queries
Request-reply RPC: void-response commands and cacheable data-response queries.
Reference
The KubeMQRouter, KubeMQPublisher, and KubeMQPublishCommand API, plus broker constructor options.
Was this page helpful?