For the conceptual model behind channels and how messages reach subscribers, see Channels & Routing in Fundamentals.

Overview

KubeMQ channels are created implicitly when a message is first published or subscribed to. However, you can also manage channels explicitly — creating, listing, deleting, and purging them through the SDK.

Create Channel

Create a channel before publishing to pre-register it with the server. The channel type must be specified: events, events_store, or queues.

create_channel.go

package main

import (
    "context"
    "log"

    "github.com/kubemq-io/kubemq-go/v2"
)

func main() {
    ctx := context.Background()
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    err = client.CreateChannel(ctx, "queues", "order-processing")
    if err != nil {
        log.Fatal(err)
    }
    log.Println("Channel created: order-processing")
}

create_channel.py

from kubemq.queues import Client as QueuesClient

client = QueuesClient(address="localhost:50000")

client.create_channel("order-processing")
print("Channel created: order-processing")
client.close()

create_channel.js

const { KubeMQClient } = require("kubemq-js");

const client = new KubeMQClient({ address: "localhost:50000" });

await client.createChannel("queues", "order-processing");
console.log("Channel created: order-processing");

CreateChannel.java

QueuesClient client = QueuesClient.builder()
    .address("localhost:50000")
    .clientId("admin-client")
    .build();

client.createChannel("order-processing");
System.out.println("Channel created: order-processing");
client.close();

CreateChannel.cs

await using var client = new KubeMQClient(new KubeMQClientOptions
{
    Address = "localhost:50000",
});
await client.ConnectAsync();

await client.CreateChannelAsync("queues", "order-processing");
Console.WriteLine("Channel created: order-processing");

CreateChannel.kt

val client = KubeMQClient.queues {
    address = "localhost:50000"
    clientId = "admin-client"
}

client.createChannel("order-processing")
println("Channel created: order-processing")
client.close()

create_channel.cpp

#include <kubemq/client.h>
#include <iostream>

kubemq::QueuesClient client("localhost:50000");

client.createChannel("order-processing");
std::cout << "Channel created: order-processing" << std::endl;

create_channel.rs

use kubemq::prelude::*;

let client = KubemqClient::builder()
    .host("localhost")
    .port(50000)
    .build()
    .await?;

client.create_queues_channel("order-processing").await?;
println!("Channel created: order-processing");

create_channel.rb

require "kubemq"

client = KubeMQ::QueuesClient.new(address: "localhost:50000", client_id: "admin-client")

client.create_queues_channel(channel_name: "order-processing")
puts "Channel created: order-processing"
client.close

create_channel.exs

{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "admin-client")

:ok = KubeMQ.Client.create_channel(client, "order-processing", :queues)
IO.puts("Channel created: order-processing")
KubeMQ.Client.close(client)

Delete Channel

Remove a channel and its configuration from the server. Messages in the channel are discarded.

Deleting a channel is irreversible. All pending messages in a queue channel are lost. Active subscriptions on the channel will receive an error.

delete_channel.go

err = client.DeleteChannel(ctx, "queues", "order-processing")
if err != nil {
    log.Fatal(err)
}
log.Println("Channel deleted: order-processing")

delete_channel.py

client.delete_channel("order-processing")
print("Channel deleted: order-processing")

delete_channel.js

await client.deleteChannel("queues", "order-processing");
console.log("Channel deleted: order-processing");

DeleteChannel.java

client.deleteChannel("order-processing");
System.out.println("Channel deleted: order-processing");

DeleteChannel.cs

await client.DeleteChannelAsync("queues", "order-processing");
Console.WriteLine("Channel deleted: order-processing");

DeleteChannel.kt

client.deleteChannel("order-processing")
println("Channel deleted: order-processing")

delete_channel.cpp

client.deleteChannel("order-processing");
std::cout << "Channel deleted: order-processing" << std::endl;

delete_channel.rs

client.delete_queues_channel("order-processing").await?;
println!("Channel deleted: order-processing");

delete_channel.rb

client.delete_queues_channel(channel_name: "order-processing")
puts "Channel deleted: order-processing"

delete_channel.exs

:ok = KubeMQ.Client.delete_channel(client, "order-processing", :queues)
IO.puts("Channel deleted: order-processing")

List Channels

Query the server for all active channels, optionally filtering by channel type or name pattern.

list_channels.go

channels, err := client.ListChannels(ctx, "queues", "order")
if err != nil {
    log.Fatal(err)
}
for _, ch := range channels {
    log.Printf("Channel: %s | Type: %s | Subscribers: %d | Messages: %d",
        ch.Name, ch.Type, ch.LastActivity, ch.Incoming)
}

list_channels.py

channels = client.list_channels("order")

for ch in channels:
    print(f"Channel: {ch.name} | Type: {ch.type} | Messages: {ch.incoming}")

list_channels.js

const channels = await client.listChannels("queues", "order");

for (const ch of channels) {
  console.log(`Channel: ${ch.name} | Type: ${ch.type} | Messages: ${ch.incoming}`);
}

ListChannels.java

List<ChannelInfo> channels = client.listChannels("order");

for (ChannelInfo ch : channels) {
    System.out.printf("Channel: %s | Type: %s | Messages: %d%n",
        ch.getName(), ch.getType(), ch.getIncoming());
}

ListChannels.cs

var channels = await client.ListChannelsAsync("queues", "order");

foreach (var ch in channels)
{
    Console.WriteLine($"Channel: {ch.Name} | Type: {ch.Type} | Messages: {ch.Incoming}");
}

ListChannels.kt

val channels = client.listChannels("order")

for (ch in channels) {
    println("Channel: ${ch.name} | Type: ${ch.type} | Messages: ${ch.incoming}")
}

list_channels.cpp

auto channels = client.listChannels("order");

for (const auto& ch : channels) {
    std::cout << "Channel: " << ch.name
              << " | Type: " << ch.type
              << " | Messages: " << ch.incoming << std::endl;
}

list_channels.rs

use kubemq::channel_type;

let channels = client.list_channels(channel_type::QUEUES, "order").await?;

for ch in &channels {
    println!("Channel: {} | Active: {} | Last activity: {}",
        ch.name, ch.is_active, ch.last_activity);
}

list_channels.rb

channels = client.list_queues_channels(search: "order")

channels.each do |ch|
  puts "Channel: #{ch.name} | Active: #{ch.is_active} | Messages: #{ch.incoming}"
end

list_channels.exs

{:ok, channels} = KubeMQ.Client.list_queues_channels(client, "order")

Enum.each(channels, fn ch ->
  IO.puts("Channel: #{ch.name} | Active: #{ch.is_active} | Messages: #{ch.incoming}")
end)

Purge Queue

Remove all pending messages from a queue channel without deleting the channel itself. This is useful for clearing a backlog during development or after recovering from a failure.

Purge permanently removes all messages from the queue. This cannot be undone.

purge_queue.go

resp, err := client.PurgeQueue(ctx, "order-processing")
if err != nil {
    log.Fatal(err)
}
log.Printf("Purged %d messages from order-processing", resp.MessagesCount)

purge_queue.py

result = client.purge_queue("order-processing")
print(f"Purged {result.messages_count} messages from order-processing")

purge_queue.js

const result = await client.purgeQueue("order-processing");
console.log(`Purged ${result.messagesCount} messages from order-processing`);

PurgeQueue.java

PurgeResult result = client.purgeQueue("order-processing");
System.out.printf("Purged %d messages from order-processing%n", result.getMessagesCount());

PurgeQueue.cs

var result = await client.PurgeQueueAsync("order-processing");
Console.WriteLine($"Purged {result.MessagesCount} messages from order-processing");

PurgeQueue.kt

val result = client.purgeQueue("order-processing")
println("Purged ${result.messagesCount} messages from order-processing")

purge_queue.cpp

auto result = client.purgeQueue("order-processing");
std::cout << "Purged " << result.messagesCount << " messages from order-processing"
          << std::endl;

purge_queue.rs

// The Rust SDK purges a queue by acking all pending messages.
client.ack_all_queue_messages("order-processing").await?;
println!("Purged all messages from order-processing");

purge_queue.rb

client.purge_queue_channel(channel_name: "order-processing")
puts "Purged all messages from order-processing"

purge_queue.exs

:ok = KubeMQ.Client.purge_queue_channel(client, "order-processing")
IO.puts("Purged all messages from order-processing")

Key Points

Implicit creation — channels are automatically created on first use (publish or subscribe)
Explicit management — use the SDK for administrative operations like cleanup or provisioning
Type-scoped — create and list operations require specifying the channel type (events, events_store, or queues)
Purge is queue-only — only queue channels support purging; events and events_store channels have different lifecycle semantics

Channel Management

Overview

Create Channel

Delete Channel

List Channels

Purge Queue

Key Points

Next Steps

Channel Routing

Production Checklist

On this page