Cancel Subscription

Overview

A live Events subscription holds an async generator and its underlying gRPC stream open indefinitely, so a long-running service needs an explicit way to tear one down without closing the whole client connection — for example when a feature flag disables a channel, a worker is draining before shutdown, or a subscription needs to be re-created with different options. Signaling an AsyncCancellationToken stops delivery cleanly and lets the subscriber task exit on its own terms.

subscribe_to_events yields events through an async for loop as long as the AsyncCancellationToken you pass to it stays uncancelled. Calling token.cancel() tells the generator to stop yielding and end the loop; calling task.cancel() on the surrounding asyncio.Task additionally interrupts any pending await inside the subscriber coroutine, so both should be used together for a clean, immediate exit.

Gotchas: cancelling the token doesn't kill the client or the asyncio event loop — only this one subscription. Events already in flight when you cancel may still arrive before the loop notices; there's no atomic cutoff point. And because Events are fire-and-forget, anything published after cancellation reaches the server but is simply dropped for this subscriber — there's no queue to catch up from later.

Prerequisites

• KubeMQ server running on localhost:50000
• Python SDK installed (pip install kubemq)

Code

"""Example: Cancel subscription — demonstrate cancelling an active subscription."""

from __future__ import annotations

import asyncio

from kubemq import (
    AsyncCancellationToken,
    AsyncPubSubClient,
    EventMessage,
    EventsSubscription,
)


async def main() -> None:
    async with AsyncPubSubClient(
        address="localhost:50000",
        client_id="python-events-cancel-subscription-client",
    ) as client:

        token = AsyncCancellationToken()

        async def subscriber() -> None:
            async for event in client.subscribe_to_events(
                subscription=EventsSubscription(
                    channel="python-events.cancel-subscription",
                    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)

        await client.publish_event(
            EventMessage(channel="python-events.cancel-subscription", body=b"before cancel")
        )
        await asyncio.sleep(1)

        token.cancel()
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        print("Subscription cancelled")

        await asyncio.sleep(1)
        await client.publish_event(
            EventMessage(channel="python-events.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 the subscribe_to_events generator to stop yielding, ending the async for loop.
• The task is also cancelled with task.cancel() to interrupt any pending await inside the subscriber coroutine.
• Events published after cancel() are sent to the server but the subscriber's loop has already exited, so they are never received.
• This pattern cleanly terminates a long-lived subscription without killing the entire asyncio event loop.

Related

• Pattern overview
• Python SDK Reference
• Basic Pub/Sub
• Consumer Group