Create Channel
Programmatically create a KubeMQ messaging channel by name and type using the Python SDK admin API.
Overview
KubeMQ auto-creates a channel the first time a client publishes or subscribes to it — convenient for prototyping, but a liability once channels are infrastructure you need to reason about. Pre-creating channels with the management API lets you provision topology before any producer or consumer connects: enforce naming conventions in a startup script, stand up the channels a service depends on as part of deployment, or fail fast if a required channel is missing instead of it silently springing into existence.
Each client exposes typed create_*_channel_async methods for the patterns it owns — events and events-store on PubSubClient, queues on QueuesClient, commands and queries on CQClient — registering the channel directly with the broker.
Gotchas: the call is idempotent for a matching name and type, so it's safe to run on every startup — but a channel's type is fixed at creation, and reusing the name with a different type fails rather than migrating it. Creation only registers the channel; it does not start a consumer, so a freshly created queue or events channel happily accepts messages with nothing yet reading them.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Create channel — create channels for events, events store, queues, and CQ."""
from __future__ import annotations
import asyncio
from kubemq import CQClient
from kubemq import PubSubClient
from kubemq import QueuesClient
async def main() -> None:
# Create events channel
async with PubSubClient(
address="localhost:50000",
client_id="python-management-create-channel-client",
) as client:
try:
await client.create_events_channel_async("python-management.create-events")
print("Events channel created: python-management.create-events")
except Exception as e:
print(f"Error creating events channel: {e}")
try:
await client.create_events_store_channel_async("python-management.create-events-store")
print("Events store channel created: python-management.create-events-store")
except Exception as e:
print(f"Error creating events store channel: {e}")
# Create queues channel
async with QueuesClient(
address="localhost:50000",
client_id="python-management-create-channel-client",
) as client:
try:
await client.create_queues_channel_async("python-management.create-queues")
print("Queues channel created: python-management.create-queues")
except Exception as e:
print(f"Error creating queues channel: {e}")
# Create CQ channels
async with CQClient(
address="localhost:50000",
client_id="python-management-create-channel-client",
) as client:
try:
await client.create_commands_channel_async("python-management.create-commands")
print("Commands channel created: python-management.create-commands")
except Exception as e:
print(f"Error creating commands channel: {e}")
try:
await client.create_queries_channel_async("python-management.create-queries")
print("Queries channel created: python-management.create-queries")
except Exception as e:
print(f"Error creating queries channel: {e}")
if __name__ == "__main__":
asyncio.run(main())
How It Works
Channel management operations use the synchronous-style async wrappers (create_*_channel_async) on the sync clients (PubSubClient, QueuesClient, CQClient). Each client type covers the channels it owns: PubSubClient manages events and events-store channels, QueuesClient manages queues channels, and CQClient manages commands and queries channels. Creating a channel that already exists is idempotent — the broker returns success rather than an error.
Related
Was this page helpful?