Cancel Subscription

Overview

A live Events subscription holds a client-side goroutine and its underlying gRPC stream open indefinitely, so a long-running service needs an explicit way to tear one down without closing the whole client connection — for example when a feature flag disables a channel, a worker is draining before shutdown, or a subscription needs to be re-created with different options. Calling Unsubscribe() on the returned handle stops delivery cleanly and frees those resources on both sides.

SubscribeToEvents returns a *Subscription handle rather than blocking, so the handler keeps running in the background until you cancel it. Unsubscribe() sends an unsubscribe request to the broker and terminates the stream goroutine; IsDone() gives you a synchronous check you can poll to confirm the teardown actually completed, which matters in tests and graceful-shutdown paths.

Gotchas: Unsubscribe() only affects this one handle — a consumer group with multiple subscribers keeps delivering to the others. Events already in flight when you call it may still land in the handler; there's a brief window where the server hasn't yet processed the unsubscribe. And because Events are fire-and-forget, anything published after cancellation is simply dropped for this subscriber — there's no queue to catch up from later.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: events/cancel-subscription
//
// Demonstrates how to cancel (unsubscribe from) an event subscription.
// After cancellation, no more events are delivered to the handler.
//
// Channel: go-events.cancel-subscription
// Client ID: go-events-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-cancel-subscription-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events.cancel-subscription"
	received := make(chan struct{})

	// Create a subscription.
	sub, err := client.SubscribeToEvents(ctx, channel, "",
		kubemq.WithOnEvent(func(event *kubemq.Event) {
			fmt.Printf("Received: body=%s\n", event.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.SendEvent(ctx, kubemq.NewEvent().
		SetChannel(channel).
		SetBody([]byte("before cancel")).
		SetMetadata("test"))
	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. After this, no more events are delivered.
	sub.Unsubscribe()
	fmt.Println("Subscription cancelled")

	// Check that the subscription is done.
	if sub.IsDone() {
		fmt.Println("Subscription confirmed done")
	}
}

How It Works

1. SubscribeToEvents returns a *Subscription handle that represents the live gRPC stream.
2. After one event is received and acknowledged via the received channel, sub.Unsubscribe() sends an unsubscribe request to the broker and terminates the stream goroutine.
3. sub.IsDone() is a synchronous check that returns true once the stream has fully closed — useful to confirm teardown in tests or cleanup logic.
4. Any events published to go-events.cancel-subscription after Unsubscribe() returns will not be delivered to this handler.

Related

• Pattern overview
• Go SDK Reference
• Basic Pub/Sub
• Consumer Group