Overview

Events are KubeMQ's fire-and-forget pub/sub pattern → see Events for what an event is, delivery semantics, and consumer groups at the broker level.

In kubemq-faststream, events are first-class FastStream endpoints. Register a handler with @broker.subscriber(events="<channel>") and publish with broker.publish(payload, events="<channel>"); both route through the broker's internal AsyncPubSubClient. The table below summarizes the API this page documents.

API element	Form	Purpose
Subscriber	`@broker.subscriber(events="<channel>")`	Register an async handler on an events channel.
Publish	`broker.publish(payload, events="<channel>")`	Publish one event.
Batch publish	`broker.publish_events_batch(*payloads, events="<channel>")`	Publish many events in one call.
Consumer group	`group="<name>"` on the subscriber	Load-balance each event to one group member.
Metadata	`message_id=`, `metadata=`, `headers=` on `publish`	Per-message id, annotation, and tag headers.

Events are fire-and-forget: no persistence, no acknowledgement. When you need persisted messages that late subscribers can replay, use Events Store instead.

A subscriber is any async function decorated with @broker.subscriber(events="<channel>"). To publish, call broker.publish(payload, events="<channel>"). The payload is a plain Python object — a dict here — which FastStream encodes for you and decodes back to the handler's type hint.

This complete app registers one subscriber and, after the broker starts, publishes three events to the same channel:

basic_pubsub.py

import asyncio
import os

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)


@broker.subscriber(events="example.events.basic")
async def handle_event(msg: dict) -> None:
    print(f"Received event: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(1, 4):
        await broker.publish(
            {"seq": i, "type": "notification"},
            events="example.events.basic",
        )
        print(f"Published event #{i}")
        await asyncio.sleep(0.5)


if __name__ == "__main__":
    asyncio.run(app.run())

Run it with python basic_pubsub.py. Each published event is fanned out to the subscriber, which prints it as it arrives:

Published event #1
Received event: {'seq': 1, 'type': 'notification'}
Published event #2
Received event: {'seq': 2, 'type': 'notification'}
Published event #3
Received event: {'seq': 3, 'type': 'notification'}

Consumer Groups

By default every subscriber on a channel receives every event (broadcast fanout). Add group="<name>" to a subscriber to join a consumer group: KubeMQ then delivers each event to exactly one member of the group, load-balancing across them. This is how you scale event processing horizontally — run several workers in the same group and the broker spreads the load.

Here two handlers share the group workers. The six published events are distributed across the two workers rather than delivered to both:

consumer_group.py

import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)


@broker.subscriber(events="example.events.group", group="workers")
async def worker_a(msg: dict) -> None:
    print(f"Worker-A received: {msg}")


@broker.subscriber(events="example.events.group", group="workers")
async def worker_b(msg: dict) -> None:
    print(f"Worker-B received: {msg}")


@app.after_startup
async def run_demo() -> None:
    print("Publishing 6 events to a consumer group of 2 workers...")
    for i in range(1, 7):
        await broker.publish({"task": i}, events="example.events.group")
        await asyncio.sleep(0.3)


if __name__ == "__main__":
    asyncio.run(app.run())

Subscribers on the same channel without a group still receive the full broadcast. Mix grouped and ungrouped subscribers freely: each group sees each event once, and every standalone subscriber sees every event.

Multiple Channels

A single app can subscribe to as many channels as you need — just register one handler per channel. Each handler is independent, which keeps unrelated concerns cleanly separated within one process:

multiple_channels.py

import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)


@broker.subscriber(events="example.events.orders")
async def on_order(msg: dict) -> None:
    print(f"[Orders]  {msg}")


@broker.subscriber(events="example.events.payments")
async def on_payment(msg: dict) -> None:
    print(f"[Payments] {msg}")


@broker.subscriber(events="example.events.notifications")
async def on_notification(msg: dict) -> None:
    print(f"[Notifs]   {msg}")


@app.after_startup
async def run_demo() -> None:
    await broker.publish({"order_id": 1001}, events="example.events.orders")
    await broker.publish({"payment_id": "pay-42"}, events="example.events.payments")
    await broker.publish({"text": "Welcome!"}, events="example.events.notifications")
    print("Published to 3 different channels")


if __name__ == "__main__":
    asyncio.run(app.run())

Wildcard Subscriptions

KubeMQ channels support wildcard patterns, so one subscriber can match many channels at once. Use prefix.* to match a single trailing segment, or prefix.> to match any number of trailing segments. The subscriber below receives every event published under example.events.wild.*:

wildcard_subscribe.py

import asyncio

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)


@broker.subscriber(events="example.events.wild.*")
async def handle_wildcard(msg: dict) -> None:
    print(f"Wildcard subscriber received: {msg}")


@app.after_startup
async def run_demo() -> None:
    channels = [
        "example.events.wild.orders",
        "example.events.wild.payments",
        "example.events.wild.alerts",
    ]

    for ch in channels:
        await broker.publish({"channel": ch, "data": "test"}, events=ch)
        print(f"Published to {ch}")
        await asyncio.sleep(0.3)


if __name__ == "__main__":
    asyncio.run(app.run())

Message Metadata

broker.publish() accepts optional metadata alongside the payload.

message_id — a custom identifier for the event. Useful for deduplication, idempotency checks, and correlating log lines across services.
metadata — a free-form string annotation that travels with the message.
headers — a string-to-string dict carried as KubeMQ tags. FastStream's gen_cor_id() is a convenient way to generate a correlation id.

Custom Message ID

message_id.py

import asyncio
import os

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

CHANNEL = "example.events.message_id"


@broker.subscriber(events=CHANNEL)
async def handle_event(msg: dict) -> None:
    print(f"Received event: {msg}")


@app.after_startup
async def run_demo() -> None:
    await broker.publish({"order": 1}, events=CHANNEL, message_id="evt-001")
    print("Published event with message_id=evt-001")

    await broker.publish({"order": 2}, events=CHANNEL, message_id="evt-002")
    print("Published event with message_id=evt-002")


if __name__ == "__main__":
    asyncio.run(app.run())

Metadata and Header Tags

metadata_tags.py

import asyncio

from faststream import FastStream
from faststream.message import gen_cor_id

from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)


@broker.subscriber(events="example.events.tags")
async def handle_event(msg: dict) -> None:
    print(f"Body: {msg}")


@app.after_startup
async def run_demo() -> None:
    await broker.publish(
        {"user": "alice", "action": "login"},
        events="example.events.tags",
        metadata="audit-trail",
        headers={
            "source": "auth-service",
            "priority": "high",
            "trace-id": gen_cor_id(),
        },
    )
    print("Published event with metadata='audit-trail' and 3 header tags")


if __name__ == "__main__":
    asyncio.run(app.run())

Batch Publishing

When you need higher throughput, publish many events in a single call with broker.publish_events_batch(). Pass the messages as positional arguments, the target channel via events=, and optionally shared headers and metadata applied to the batch. Sending one batch instead of N individual publish() calls cuts per-message round-trip overhead.

events_batch.py

import asyncio
import os
import time

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

received_count = 0


@broker.subscriber(events="example.batch.events")
async def handle(msg: dict) -> None:
    global received_count
    received_count += 1


@app.after_startup
async def run_demo() -> None:
    # Individual publish
    start = time.monotonic()
    for i in range(20):
        await broker.publish({"order_id": i}, events="example.batch.events")
    individual_time = time.monotonic() - start
    print(f"Published 20 events individually in {individual_time:.3f}s")

    await asyncio.sleep(1)

    # Batch publish
    messages = [{"order_id": i + 20} for i in range(20)]
    start = time.monotonic()
    await broker.publish_events_batch(
        *messages,
        events="example.batch.events",
        headers={"source": "batch"},
        metadata="batch-demo",
    )
    batch_time = time.monotonic() - start
    print(f"Published 20 events in batch in {batch_time:.3f}s")

    await asyncio.sleep(2)
    print(f"Total received: {received_count}")


if __name__ == "__main__":
    asyncio.run(app.run())

Batch publishing is fire-and-forget like single publishing — there is no per-message acknowledgement. For batch operations across the other patterns (queues, commands, queries), see the batch examples in the examples/batch_operations directory.

Acknowledgement Is a No-Op

Events are fire-and-forget by design. There is no settlement step: the broker does not wait for, expect, or track an acknowledgement from subscribers. As a result, any ack or nack operation on an event message is a no-op — it has no effect on delivery.

ack and nack are no-ops for Events and Events Store. The FastStream AckPolicy options only matter for the Queues pattern, which provides transactional settlement. If you need a handler failure to redeliver a message, use Queues, not Events.

Events (Pub/Sub)

Overview

Consumer Groups

Multiple Channels

Wildcard Subscriptions

Message Metadata

Custom Message ID

Metadata and Header Tags

Batch Publishing

Acknowledgement Is a No-Op

Events (core concept)

Events Store

Queues

Composition

On this page