Cancel Subscription
Unsubscribe from a KubeMQ Events Store channel and cancel an active subscription with the Python SDK.
Overview
Every events store subscription opens a long-lived stream to the broker — an async for loop pulling delivered events until you tell it to stop. Signaling AsyncCancellationToken.cancel() is how you release that loop deliberately: shutting down a worker, rotating consumers, or tearing down a task without leaking connections or leaving a dangling stream on the server.
Internally, cancelling the token stops the subscribe_to_events_store generator and ends the async for loop; the enclosing asyncio task is typically cancelled too, to interrupt any pending await inside the subscriber coroutine immediately rather than waiting for the next event.
Gotchas: cancelling only stops this subscriber — the channel keeps storing every event published afterward, so nothing is lost, and a fresh subscription with a replay start position picks up exactly where this one left off. Cancelling the token and cancelling the task are two different signals; forgetting the task cancellation can leave a coroutine blocked on an in-flight await longer than expected.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Cancel subscription — demonstrate cancelling an events store subscription."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncPubSubClient,
EventStoreMessage,
EventsStoreSubscription,
)
from kubemq.pubsub import EventStoreStartPosition
async def main() -> None:
async with AsyncPubSubClient(
address="localhost:50000",
client_id="python-events-store-cancel-subscription-client",
) as client:
token = AsyncCancellationToken()
async def subscriber() -> None:
async for event in client.subscribe_to_events_store(
subscription=EventsStoreSubscription(
channel="python-events-store.cancel-subscription",
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
events_store_type=EventStoreStartPosition.StartFromNew,
),
cancellation_token=token,
):
print(f"Received: {event.body.decode('utf-8')}")
task = asyncio.create_task(subscriber())
await asyncio.sleep(1)
await client.send_event_store(
EventStoreMessage(
channel="python-events-store.cancel-subscription",
body=b"before cancel",
)
)
await asyncio.sleep(1)
token.cancel()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
print("Events store subscription cancelled")
await client.send_event_store(
EventStoreMessage(
channel="python-events-store.cancel-subscription",
body=b"after cancel",
)
)
print("Message sent after cancel — subscriber will NOT receive it")
await asyncio.sleep(1)
if __name__ == "__main__":
asyncio.run(main())
How It Works
AsyncCancellationToken.cancel()signals thesubscribe_to_events_storegenerator to stop, ending theasync forloop.- The task is also explicitly cancelled with
task.cancel()to interrupt any pendingawaitinside the subscriber coroutine. - Events sent to the channel after cancellation are still persisted by the broker; the subscriber simply no longer receives them.
- A new subscription with a replay start position can resume from where processing stopped, because the events remain stored.
Related
Was this page helpful?