Wildcard Subscription
Subscribe to multiple KubeMQ event channels at once using wildcard patterns with the Python SDK.
Overview
A wildcard subscription lets one subscriber match a whole family of channels with a single call, instead of wiring up a separate subscribe_to_events for every sub-channel and touching code each time a new one appears. It's the natural fit for monitoring, logging, or fan-in aggregation across a channel hierarchy — for example, watching every regional order channel from one place.
KubeMQ matches wildcard tokens against the channel hierarchy server-side at delivery time. * matches exactly one dot-separated segment, and > matches one or more trailing segments, so an EventsSubscription with channel="python-events.wildcard.*" catches any single-segment suffix. Every delivered event still carries its exact channel, so the handler can tell which concrete sub-channel it came from even though the subscription itself only named a pattern.
Gotchas: * matches exactly one segment — it won't reach two levels deep, so orders.* misses orders.us.east; use > for that. Wildcards are only valid on Events subscriptions, not on publish_event/publishes or on events-store, queues, or commands/queries. And an overly broad pattern like > at the root will quietly pull in every channel under that prefix, including ones you didn't intend to monitor.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Wildcard subscription — subscribe to events on channels matching a pattern."""
from __future__ import annotations
import asyncio
from kubemq import AsyncCancellationToken, EventMessage, EventsSubscription
from kubemq import AsyncPubSubClient
def on_event(event) -> None: # type: ignore[no-untyped-def]
"""Handle received event from any matching channel."""
print(f"[{event.channel}] Received: {event.body.decode('utf-8')}")
async def main() -> None:
async with AsyncPubSubClient(
address="localhost:50000",
client_id="python-events-wildcard-subscription-client",
) as client:
# Subscribe using a wildcard pattern
# This will receive events from all channels matching "python-events.wildcard.*"
token = AsyncCancellationToken()
async def subscriber() -> None:
async for event in client.subscribe_to_events(
subscription=EventsSubscription(
channel="python-events.wildcard.*",
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
print(f"Received: {event.body.decode('utf-8')}")
task = asyncio.create_task(subscriber())
await asyncio.sleep(1)
# Publish events to different sub-channels
await client.publish_event(
EventMessage(channel="python-events.wildcard.created", body=b"Order #1001 created")
)
await client.publish_event(
EventMessage(channel="python-events.wildcard.shipped", body=b"Order #1002 shipped")
)
await client.publish_event(
EventMessage(channel="python-events.wildcard.delivered", body=b"Order #1003 delivered")
)
print("Events sent to wildcard sub-channels")
await asyncio.sleep(3)
token.cancel()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
- The subscription channel
python-events.wildcard.*uses a single-level wildcard (*) that matches any single path segment after the prefix. - KubeMQ matches the pattern server-side; the subscriber receives events from
*.created,*.shipped,*.delivered, and any other matching sub-channel. - Each
EventReceivedcarries the originalchannelfield so the handler can distinguish which sub-channel the event came from. - Wildcard patterns do not match multiple segments; use
>for multi-level matching if your channel hierarchy is deeper.
Related
Was this page helpful?