What You Will Build

A monitoring system where services publish events to hierarchical channels like orders.created and payments.completed, and wildcard subscribers capture events across categories.

One publish per channel; a single-level (orders.*) and a catch-all (>) subscriber each match a different slice of the stream.

Prerequisites

KubeMQ server running on localhost:50000
SDK installed (Getting Started)

Wildcard subscriptions are supported for Events only. Events Store, Queues, and RPC patterns do not support wildcards.

Wildcard Patterns

Pattern	Matches	Example
`*`	Exactly one token	`orders.*` matches `orders.created` but not `orders.us.created`
`>`	One or more tokens	`orders.>` matches `orders.created` and `orders.us.created`

Tokens are separated by . (dot). A standalone > subscribes to every channel.

Steps

Publish Events to Multiple Channels

multi_publisher.go

package main

import (
    "context"
    "log"
    "time"

    "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()

    events := []struct {
        channel string
        body    string
    }{
        {"orders.created", `{"orderId":"ORD-100","action":"created"}`},
        {"orders.updated", `{"orderId":"ORD-100","action":"updated"}`},
        {"orders.shipped", `{"orderId":"ORD-100","action":"shipped"}`},
        {"payments.completed", `{"paymentId":"PAY-200","status":"completed"}`},
        {"inventory.reserved", `{"sku":"ITEM-300","qty":5}`},
    }

    for _, e := range events {
        err = client.SendEvent(ctx, kubemq.NewEvent().
            SetChannel(e.channel).
            SetBody([]byte(e.body)),
        )
        if err != nil {
            log.Printf("Failed to publish to %s: %v", e.channel, err)
            continue
        }
        log.Printf("Published to %s", e.channel)
        time.Sleep(300 * time.Millisecond)
    }
}

multi_publisher.py

import time
from kubemq.pubsub import Client as PubSubClient
from kubemq.pubsub import EventMessage

events = [
    ("orders.created", '{"orderId":"ORD-100","action":"created"}'),
    ("orders.updated", '{"orderId":"ORD-100","action":"updated"}'),
    ("orders.shipped", '{"orderId":"ORD-100","action":"shipped"}'),
    ("payments.completed", '{"paymentId":"PAY-200","status":"completed"}'),
    ("inventory.reserved", '{"sku":"ITEM-300","qty":5}'),
]

client = PubSubClient(address="localhost:50000")
for channel, body in events:
    client.send_event(
        EventMessage(channel=channel, body=body.encode("utf-8"))
    )
    print(f"Published to {channel}")
    time.sleep(0.3)
client.close()

multi_publisher.js

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

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

const events = [
  { channel: "orders.created", body: '{"orderId":"ORD-100","action":"created"}' },
  { channel: "orders.updated", body: '{"orderId":"ORD-100","action":"updated"}' },
  { channel: "orders.shipped", body: '{"orderId":"ORD-100","action":"shipped"}' },
  { channel: "payments.completed", body: '{"paymentId":"PAY-200","status":"completed"}' },
  { channel: "inventory.reserved", body: '{"sku":"ITEM-300","qty":5}' },
];

for (const e of events) {
  await client.sendEvent({ channel: e.channel, body: Buffer.from(e.body) });
  console.log(`Published to ${e.channel}`);
  await new Promise((r) => setTimeout(r, 300));
}

MultiPublisher.java

PubSubClient client = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("multi-publisher")
    .build();

String[][] events = {
    {"orders.created", "{\"orderId\":\"ORD-100\",\"action\":\"created\"}"},
    {"orders.updated", "{\"orderId\":\"ORD-100\",\"action\":\"updated\"}"},
    {"orders.shipped", "{\"orderId\":\"ORD-100\",\"action\":\"shipped\"}"},
    {"payments.completed", "{\"paymentId\":\"PAY-200\",\"status\":\"completed\"}"},
    {"inventory.reserved", "{\"sku\":\"ITEM-300\",\"qty\":5}"},
};

for (String[] e : events) {
    client.sendEventsMessage(EventMessage.builder()
        .channel(e[0])
        .body(e[1].getBytes())
        .build());
    System.out.println("Published to " + e[0]);
    Thread.sleep(300);
}
client.close();

MultiPublisher.cs

await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

var events = new[]
{
    ("orders.created", "{\"orderId\":\"ORD-100\",\"action\":\"created\"}"),
    ("orders.updated", "{\"orderId\":\"ORD-100\",\"action\":\"updated\"}"),
    ("orders.shipped", "{\"orderId\":\"ORD-100\",\"action\":\"shipped\"}"),
    ("payments.completed", "{\"paymentId\":\"PAY-200\",\"status\":\"completed\"}"),
    ("inventory.reserved", "{\"sku\":\"ITEM-300\",\"qty\":5}"),
};

foreach (var (channel, body) in events)
{
    await client.SendEventAsync(new EventMessage
    {
        Channel = channel,
        Body = Encoding.UTF8.GetBytes(body),
    });
    Console.WriteLine($"Published to {channel}");
    await Task.Delay(300);
}

MultiPublisher.kt

val client = PubSubClient("localhost:50000")

val events = listOf(
    "orders.created" to """{"orderId":"ORD-100","action":"created"}""",
    "orders.updated" to """{"orderId":"ORD-100","action":"updated"}""",
    "orders.shipped" to """{"orderId":"ORD-100","action":"shipped"}""",
    "payments.completed" to """{"paymentId":"PAY-200","status":"completed"}""",
    "inventory.reserved" to """{"sku":"ITEM-300","qty":5}""",
)

for ((channel, body) in events) {
    client.sendEvent(EventMessage(channel = channel, body = body.toByteArray()))
    println("Published to $channel")
    Thread.sleep(300)
}
client.close()

multi_publisher.cpp

auto client = kubemq::PubSubClient("localhost:50000");

std::vector<std::pair<std::string, std::string>> events = {
    {"orders.created", R"({"orderId":"ORD-100","action":"created"})"},
    {"orders.updated", R"({"orderId":"ORD-100","action":"updated"})"},
    {"orders.shipped", R"({"orderId":"ORD-100","action":"shipped"})"},
    {"payments.completed", R"({"paymentId":"PAY-200","status":"completed"})"},
    {"inventory.reserved", R"({"sku":"ITEM-300","qty":5})"},
};

for (const auto& [channel, body] : events) {
    kubemq::EventMessage event;
    event.channel = channel;
    event.body = body;
    client.sendEvent(event);
    std::cout << "Published to " << channel << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
}

multi_publisher.rs

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

let events = [
    ("orders.created", r#"{"orderId":"ORD-100","action":"created"}"#),
    ("orders.updated", r#"{"orderId":"ORD-100","action":"updated"}"#),
    ("orders.shipped", r#"{"orderId":"ORD-100","action":"shipped"}"#),
    ("payments.completed", r#"{"paymentId":"PAY-200","status":"completed"}"#),
    ("inventory.reserved", r#"{"sku":"ITEM-300","qty":5}"#),
];

for (channel, body) in events {
    let event = EventBuilder::new()
        .channel(channel)
        .body(body.as_bytes().to_vec())
        .build();
    client.send_event(event).await?;
    println!("Published to {}", channel);
    tokio::time::sleep(Duration::from_millis(300)).await;
}

multi_publisher.rb

client = KubeMQ::PubSubClient.new(address: "localhost:50000", client_id: "multi-publisher")

events = [
  ["orders.created", '{"orderId":"ORD-100","action":"created"}'],
  ["orders.updated", '{"orderId":"ORD-100","action":"updated"}'],
  ["orders.shipped", '{"orderId":"ORD-100","action":"shipped"}'],
  ["payments.completed", '{"paymentId":"PAY-200","status":"completed"}'],
  ["inventory.reserved", '{"sku":"ITEM-300","qty":5}'],
]

events.each do |channel, body|
  client.send_event(KubeMQ::PubSub::EventMessage.new(channel: channel, body: body))
  puts "Published to #{channel}"
  sleep 0.3
end
client.close

multi_publisher.exs

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

events = [
  {"orders.created", ~s({"orderId":"ORD-100","action":"created"})},
  {"orders.updated", ~s({"orderId":"ORD-100","action":"updated"})},
  {"orders.shipped", ~s({"orderId":"ORD-100","action":"shipped"})},
  {"payments.completed", ~s({"paymentId":"PAY-200","status":"completed"})},
  {"inventory.reserved", ~s({"sku":"ITEM-300","qty":5})}
]

for {channel, body} <- events do
  :ok = KubeMQ.Client.send_event(client, KubeMQ.Event.new(channel: channel, body: body))
  IO.puts("Published to #{channel}")
  Process.sleep(300)
end

KubeMQ.Client.close(client)

Subscribe to orders.* to receive only order-related events at one level of nesting.

orders_monitor.go

sub, err := client.SubscribeToEvents(ctx, "orders.*", "",
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[Orders Monitor] channel=%s body=%s\n",
            event.Channel, string(event.Body))
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("Error:", err)
    }),
)

orders_monitor.py

client.subscribe_to_events(
    subscription=EventsSubscription(
        channel="orders.*",
        on_receive_event_callback=lambda e: print(
            f"[Orders Monitor] channel={e.channel} body={e.body.decode()}"
        ),
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

orders_monitor.js

client.subscribeToEvents({
  channel: "orders.*",
  onEvent: (msg) =>
    console.log(
      `[Orders Monitor] channel=${msg.channel} body=${Buffer.from(msg.body).toString()}`
    ),
  onError: (err) => console.error("Error:", err.message),
});

OrdersMonitor.java

client.subscribeToEvents(EventsSubscription.builder()
    .channel("orders.*")
    .onReceiveEventCallback(event ->
        System.out.printf("[Orders Monitor] channel=%s body=%s%n",
            event.getChannel(), new String(event.getBody())))
    .onErrorCallback(err ->
        System.err.println("Error: " + err.getMessage()))
    .build());

OrdersMonitor.cs

await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "orders.*" }))
{
    Console.WriteLine($"[Orders Monitor] channel={msg.Channel} "
        + $"body={Encoding.UTF8.GetString(msg.Body.Span)}");
}

OrdersMonitor.kt

client.subscribeToEvents(
    channel = "orders.*",
    onEvent = { event ->
        println("[Orders Monitor] channel=${event.channel} body=${String(event.body)}")
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

orders_monitor.cpp

client.subscribeToEvents("orders.*", "",
    [](const kubemq::Event& event) {
        std::cout << "[Orders Monitor] channel=" << event.channel
                  << " body=" << event.body << std::endl;
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);

orders_monitor.rs

let sub = client
    .subscribe_to_events(
        "orders.*",
        "",
        |event| {
            Box::pin(async move {
                println!(
                    "[Orders Monitor] channel={} body={}",
                    event.channel,
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;

orders_monitor.rb

cancel = KubeMQ::CancellationToken.new
sub = KubeMQ::PubSub::EventsSubscription.new(channel: "orders.*")

client.subscribe_to_events(sub, cancellation_token: cancel,
  on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
  puts "[Orders Monitor] channel=#{event.channel} body=#{event.body}"
end

orders_monitor.exs

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events(client, "orders.*",
    on_event: fn event ->
      IO.puts("[Orders Monitor] channel=#{event.channel} body=#{event.body}")
    end
  )

Expected output — receives 3 of 5 events (only orders.*):

[Orders Monitor] channel=orders.created body={"orderId":"ORD-100","action":"created"}
[Orders Monitor] channel=orders.updated body={"orderId":"ORD-100","action":"updated"}
[Orders Monitor] channel=orders.shipped body={"orderId":"ORD-100","action":"shipped"}

Subscribe to > to receive events from all channels.

global_auditor.go

sub, err := client.SubscribeToEvents(ctx, ">", "",
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[Auditor] channel=%s body=%s\n",
            event.Channel, string(event.Body))
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("Error:", err)
    }),
)

global_auditor.py

client.subscribe_to_events(
    subscription=EventsSubscription(
        channel=">",
        on_receive_event_callback=lambda e: print(
            f"[Auditor] channel={e.channel} body={e.body.decode()}"
        ),
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

global_auditor.js

client.subscribeToEvents({
  channel: ">",
  onEvent: (msg) =>
    console.log(
      `[Auditor] channel=${msg.channel} body=${Buffer.from(msg.body).toString()}`
    ),
  onError: (err) => console.error("Error:", err.message),
});

GlobalAuditor.java

client.subscribeToEvents(EventsSubscription.builder()
    .channel(">")
    .onReceiveEventCallback(event ->
        System.out.printf("[Auditor] channel=%s body=%s%n",
            event.getChannel(), new String(event.getBody())))
    .onErrorCallback(err ->
        System.err.println("Error: " + err.getMessage()))
    .build());

GlobalAuditor.cs

await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = ">" }))
{
    Console.WriteLine($"[Auditor] channel={msg.Channel} "
        + $"body={Encoding.UTF8.GetString(msg.Body.Span)}");
}

GlobalAuditor.kt

client.subscribeToEvents(
    channel = ">",
    onEvent = { event ->
        println("[Auditor] channel=${event.channel} body=${String(event.body)}")
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

global_auditor.cpp

client.subscribeToEvents(">", "",
    [](const kubemq::Event& event) {
        std::cout << "[Auditor] channel=" << event.channel
                  << " body=" << event.body << std::endl;
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);

global_auditor.rs

let sub = client
    .subscribe_to_events(
        ">",
        "",
        |event| {
            Box::pin(async move {
                println!(
                    "[Auditor] channel={} body={}",
                    event.channel,
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;

global_auditor.rb

cancel = KubeMQ::CancellationToken.new
sub = KubeMQ::PubSub::EventsSubscription.new(channel: ">")

client.subscribe_to_events(sub, cancellation_token: cancel,
  on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
  puts "[Auditor] channel=#{event.channel} body=#{event.body}"
end

global_auditor.exs

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events(client, ">",
    on_event: fn event ->
      IO.puts("[Auditor] channel=#{event.channel} body=#{event.body}")
    end
  )

Expected output — receives all 5 events:

[Auditor] channel=orders.created body={"orderId":"ORD-100","action":"created"}
[Auditor] channel=orders.updated body={"orderId":"ORD-100","action":"updated"}
[Auditor] channel=orders.shipped body={"orderId":"ORD-100","action":"shipped"}
[Auditor] channel=payments.completed body={"paymentId":"PAY-200","status":"completed"}
[Auditor] channel=inventory.reserved body={"sku":"ITEM-300","qty":5}

Channel Naming Best Practices

Use a hierarchical dot-separated naming convention:

{domain}.{entity}.{action}

orders.created
orders.updated
orders.us-east.created
payments.completed

Subscription	Receives
`orders.created`	Only `orders.created` events
`orders.*`	All single-level order events
`orders.>`	All order events, including nested like `orders.us-east.created`
`>`	Everything across all channels

Channel names used for publishing must not contain * or > characters. Wildcards are only valid in subscription channel patterns.

Wildcard Subscriptions

What You Will Build

Prerequisites

Wildcard Patterns

Steps

Publish Events to Multiple Channels

Channel Naming Best Practices

Next Steps

Multicast Events

Scale Subscribers

On this page