Basic Pub/Sub

Overview

This tutorial builds the "hello world" of KubeMQ messaging: a publisher and a subscriber talking over the Events pattern. Events are fire-and-forget — the broker fans a message out to every subscriber currently listening on the channel and moves on. There's no persistence, no acknowledgment, and no replay, which makes this the pattern to reach for when you need low-latency, high-throughput broadcast (metrics ticks, live status updates, cache-invalidation signals) and can tolerate losing a message if nobody is listening at the moment it's sent.

You'll wire up SubscribeToEvents with a WithOnEvent callback, give the subscription a moment to register with the server, then call SendEvent to publish. The empty consumer-group argument means fan-out delivery: every connected subscriber gets its own copy, as opposed to a consumer group where only one member would receive it. Gotchas: if the subscriber isn't fully established before you publish, the event is simply gone — there's no queue catching it, which is why the sample sleeps briefly before sending; and because delivery isn't acknowledged, a crashed or disconnected subscriber never knows it missed anything.

Prerequisites

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

Code

// Example: events/basic-pubsub
//
// Demonstrates basic fire-and-forget event publish/subscribe.
// A subscriber listens on a channel, then a publisher sends an event.
//
// Channel: go-events.basic-pubsub
// Client ID: go-events-basic-pubsub-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), // TODO: Replace with your KubeMQ server address
		kubemq.WithClientId("go-events-basic-pubsub-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

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

	// Subscribe to events on the channel.
	sub, err := client.SubscribeToEvents(ctx, channel, "",
		kubemq.WithOnEvent(func(event *kubemq.Event) {
			fmt.Printf("Received: channel=%s body=%s metadata=%s\n",
				event.Channel, event.Body, event.Metadata)
			close(received)
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Subscription error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Unsubscribe()

	// Allow subscription to fully establish before publishing.
	time.Sleep(time.Second)

	// Publish an event to the channel.
	err = client.SendEvent(ctx, kubemq.NewEvent().
		SetChannel(channel).
		SetBody([]byte("hello from Go SDK")).
		SetMetadata("greeting"))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("Event published")

	// Wait for the event to be received.
	select {
	case <-received:
		fmt.Println("Event received successfully")
	case <-ctx.Done():
		log.Fatal("Timed out waiting for event")
	}
}

// Expected output:
// Event published
// Received: channel=go-events.basic-pubsub body=hello from Go SDK metadata=greeting
// Event received successfully

How It Works

1. NewClient opens a gRPC connection using WithClientId("go-events-basic-pubsub-client").
2. SubscribeToEvents registers a callback via WithOnEvent; the empty group argument means every subscriber on the channel receives every event (fan-out). A WithOnError callback surfaces transport errors.
3. A one-second sleep lets the subscription register on the server before the publisher sends.
4. SendEvent publishes to go-events.basic-pubsub; the broker delivers it to all active subscribers.
5. The channel received synchronises the program — close(received) is called inside the callback, and the main goroutine unblocks via the select.
6. defer sub.Unsubscribe() and defer client.Close() clean up on exit.

Related

• Pattern overview
• Go SDK Reference
• Cancel Subscription
• Consumer Group