# Events (Pub/Sub) (/integrations/faststream/how-to/events) ## Overview [#overview] Events are KubeMQ's fire-and-forget pub/sub pattern → see [Events](/learn/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="")` and publish with `broker.publish(payload, events="")`; 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="")` | Register an async handler on an events channel. | | Publish | `broker.publish(payload, events="")` | Publish one event. | | Batch publish | `broker.publish_events_batch(*payloads, events="")` | Publish many events in one call. | | Consumer group | `group=""` 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](/integrations/faststream/how-to/events-store) instead. ## Subscribe and Publish [#subscribe-and-publish] A subscriber is any async function decorated with `@broker.subscriber(events="")`. To publish, call `broker.publish(payload, events="")`. 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: ```python title="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: ```text 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 [#consumer-groups] By default every subscriber on a channel receives every event (broadcast fanout). Add `group=""` 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: ```python title="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 [#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: ```python title="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 [#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.*`: ```python title="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 [#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 [#custom-message-id] ```python title="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-and-header-tags] ```python title="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 [#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. ```python title="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`](https://github.com/kubemq-io/kubemq-faststream/tree/main/examples/batch_operations) directory. ## Acknowledgement Is a No-Op [#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](/integrations/faststream/how-to/queues) pattern, which provides transactional settlement. If you need a handler failure to redeliver a message, use Queues, not Events. ## Related [#related]