Close a KubeMQ Python Client

Overview

Closing a client isn't an afterthought — it tells the broker and your own process that this connection is done, so both sides release what they were holding for it. A KubeMQ client is more than a socket: it's a gRPC channel plus whatever subscriptions and in-flight sends it's servicing. Skip the close and those linger — subscriptions keep streaming, the channel stays open — and in short-lived processes (CLI tools, batch jobs, test suites) you leak connections until the process is killed.

Awaiting client.close() drains operations already in flight, then tears down the underlying gRPC channel. Once it returns, the client is in a terminal closed state — every call after that fails fast instead of hanging silently.

Gotchas: the drain window is bounded, not unlimited, so a slow consumer can still lose the tail of a burst if you close mid-stream; a closed client is dead forever — no reconnect on the same instance, build a new one; and if an exception is possible between connect() and close(), put close() in a finally block (or use async with) so shutdown runs on every exit path, not just the happy one.

Prerequisites

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

Code

"""Example: Close — demonstrate graceful client close."""

from __future__ import annotations

import asyncio

from kubemq import AsyncPubSubClient, EventMessage


async def main() -> None:
    client = AsyncPubSubClient(
        address="localhost:50000",
        client_id="python-connection-close-client",
    )
    try:
        await client.connect()
        await client.publish_event(
            EventMessage(channel="python-connection.close", body=b"Hello before close")
        )
        print("Event sent successfully")
    finally:
        await client.close()
        print("Client closed gracefully")


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

How It Works

AsyncPubSubClient is instantiated without a context manager so that the finally block controls the close lifecycle explicitly. connect() is awaited to establish the gRPC connection, an event is published, then close() is awaited in the finally block to drain any in-flight operations and release the channel. Using async with achieves the same result automatically; the explicit form shows the underlying lifecycle for clarity.

Related

• Python SDK Reference
• Connect
• Ping