Cancel Subscription
Cancel and unsubscribe from a KubeMQ Events Store channel cleanly using the Go SDK.
Overview
Every events store subscription opens a long-lived stream to the broker — a background goroutine that keeps pulling delivered events until you tell it to stop. Calling Unsubscribe on the subscription handle is how you release that goroutine deliberately: shutting down a worker, rotating consumers, or tearing down a test without leaking connections or leaving a dangling stream on the server.
Internally, Unsubscribe sends a close request to the broker and terminates the background receive loop; sub.IsDone() then reports once that goroutine has fully exited, which is useful as a lightweight synchronization point rather than guessing with a sleep.
Gotchas: cancelling only stops this subscriber — the channel keeps storing every event published afterward, so nothing is lost, and a fresh subscription with StartFromFirstEvent() or a specific sequence picks up exactly where this one left off. Unsubscribe() doesn't guarantee the goroutine has exited the instant it returns; poll IsDone() (or synchronize some other way) before assuming no more callbacks can fire.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: events-store/cancel-subscription
//
// Demonstrates cancelling an event store subscription.
// After Unsubscribe is called, no more events are delivered.
//
// Channel: go-events-store.cancel-subscription
// Client ID: go-events-store-cancel-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-store-cancel-subscription-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
channel := "go-events-store.cancel-subscription"
received := make(chan struct{})
// Create an event store subscription.
sub, err := client.SubscribeToEventsStore(ctx, channel, "",
kubemq.StartFromNewEvents(),
kubemq.WithOnEventStoreReceive(func(e *kubemq.EventStoreReceive) {
fmt.Printf("Received: seq=%d body=%s\n", e.Sequence, string(e.Body))
close(received)
}),
kubemq.WithOnError(func(err error) {
log.Println("Subscription error:", err)
}),
)
if err != nil {
log.Fatal(err)
}
// Allow time for subscription to register on server
time.Sleep(1 * time.Second)
// Send an event and wait for it.
_, err = client.SendEventStore(ctx, kubemq.NewEventStore().
SetChannel(channel).
SetBody([]byte("before cancel")))
if err != nil {
log.Fatal(err)
}
select {
case <-received:
fmt.Println("Event received before cancellation")
case <-ctx.Done():
log.Fatal("Timed out")
}
// Cancel the subscription.
sub.Unsubscribe()
fmt.Println("Subscription cancelled")
if sub.IsDone() {
fmt.Println("Subscription confirmed done")
}
}
How It Works
SubscribeToEventsStoreopens a persistent subscription stream;StartFromNewEvents()means only events published after this call are delivered.- After one event is received,
sub.Unsubscribe()sends a close request to the broker and terminates the background stream goroutine. sub.IsDone()returnstrueonce the stream goroutine has fully exited — useful as a lightweight synchronisation point in tests.- Events stored on the channel while the subscription is inactive are not lost; re-subscribing with
StartFromFirstEvent()or a specific sequence would replay them.
Related
Was this page helpful?