Wildcard Subscription
Subscribe to KubeMQ Events using wildcard channel patterns with the Go SDK to match many channels at once.
Overview
A wildcard subscription lets one subscriber match a whole family of channels with a single call, instead of wiring up a separate SubscribeToEvents for every sub-channel and touching code each time a new one appears. It's the natural fit for monitoring, logging, or fan-in aggregation across a channel hierarchy — for example, watching every regional order channel from one place.
KubeMQ matches wildcard tokens against the channel hierarchy server-side at delivery time. * matches exactly one dot-separated segment, and > matches one or more trailing segments, so SubscribeToEvents(ctx, "go-events.wildcard.*", ...) catches any single-segment suffix. Every delivered Event still carries its exact Channel, so the callback can tell which concrete sub-channel it came from even though the subscription itself only named a pattern.
Gotchas: * matches exactly one segment — it won't reach two levels deep, so orders.* misses orders.us.east; use > for that. Wildcards are only valid on Events subscriptions, not on SendEvent/publishes or on events-store, queues, or commands/queries. And an overly broad pattern like > at the root will quietly pull in every channel under that prefix, including ones you didn't intend to monitor.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: events/wildcard-subscription
//
// Demonstrates subscribing to events using a wildcard pattern.
// Wildcard "go-events.wildcard.*" matches channels like
// "go-events.wildcard.a" and "go-events.wildcard.b".
//
// Channel: go-events.wildcard-subscription
// Client ID: go-events-wildcard-subscription-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/kubemq-io/kubemq-go/v2"
)
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
client, err := kubemq.NewClient(ctx,
kubemq.WithAddress("localhost", 50000),
kubemq.WithClientId("go-events-wildcard-subscription-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
received := make(chan struct{}, 2)
// Subscribe with a wildcard pattern to match multiple channels.
sub, err := client.SubscribeToEvents(ctx, "go-events.wildcard.*", "",
kubemq.WithOnEvent(func(event *kubemq.Event) {
fmt.Printf("Wildcard received: channel=%s body=%s\n",
event.Channel, event.Body)
select {
case received <- struct{}{}:
default:
}
}),
kubemq.WithOnError(func(err error) {
log.Println("Subscription error:", err)
}),
)
if err != nil {
log.Fatal(err)
}
defer sub.Unsubscribe()
// Publish to two channels that match the wildcard pattern.
for _, ch := range []string{"go-events.wildcard.a", "go-events.wildcard.b"} {
err = client.SendEvent(ctx, kubemq.NewEvent().
SetChannel(ch).
SetBody([]byte("wildcard-msg")).
SetMetadata("wildcard-demo"))
if err != nil {
log.Fatalf("SendEvent to %s: %v", ch, err)
}
fmt.Printf("Published to %s\n", ch)
}
// Wait for at least one event.
select {
case <-received:
fmt.Println("Wildcard subscription demo complete")
case <-ctx.Done():
log.Fatal("Timed out waiting for wildcard event")
}
}
How It Works
SubscribeToEventsis called with the pattern"go-events.wildcard.*"— the*wildcard matches any single path segment on the broker.- Events published to
go-events.wildcard.aandgo-events.wildcard.bboth match the pattern and are delivered to the same subscription callback. - The
event.Channelfield in the callback contains the exact channel the publisher used, not the wildcard pattern, so you can distinguish the source. - A buffered
receivedchannel (capacity 2) prevents the callback from blocking the gRPC stream if the main goroutine is slow to read.
Related
Was this page helpful?