Overview

Which to use

This page covers high-throughput Events Store (persistent, replayable) streaming via SendEventStoreStream. For the fire-and-forget equivalent, see Events Stream Send.

Stream send decouples publishing a persistent event from waiting for its storage confirmation, so you can push a large batch onto the wire without stalling on a round trip per message. The one-at-a-time SendEventStore call is fine for occasional writes, but if you're bulk-loading history, replicating a firehose of records, or backfilling an Events Store channel, a request/response call per event turns network latency into your throughput ceiling.

SendEventStoreStream opens a single bidirectional gRPC stream and hands back a handle: handle.Send pushes events onto it as fast as you can call it, while a separate handle.Results channel delivers each event's broker-assigned EventID and Sent confirmation as storage completes, independent of send order. Gotchas: results can arrive out of order relative to sends, so correlate them by EventID rather than assuming a 1:1 positional match; closing the handle before draining Results can silently drop confirmations for events still in flight; and for low-volume or one-off publishing, the extra bookkeeping isn't worth it — reach for the plain SendEventStore call instead.

Prerequisites

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

Code

main.go

// Example: events-store/stream-send
//
// Demonstrates high-throughput event store publishing using SendEventStoreStream.
// Each sent event receives a confirmation result via the Results channel.
//
// Channel: go-events-store.stream-send
// Client ID: go-events-store-stream-send-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-stream-send-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events-store.stream-send"

	// Open a bidirectional stream for event store publishing.
	handle, err := client.SendEventStoreStream(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer handle.Close()

	// Read confirmation results in the background.
	go func() {
		for r := range handle.Results {
			fmt.Printf("Stream result: eventId=%s sent=%v err=%s\n",
				r.EventID, r.Sent, r.Error)
		}
	}()

	// Send multiple events via the stream.
	for i := range 5 {
		ev := kubemq.NewEventStore().
			SetChannel(channel).
			SetBody(fmt.Appendf(nil, "stream-data-%d", i)).
			SetMetadata("stream-meta")
		if err := handle.Send(ev); err != nil {
			log.Fatal(err)
		}
		fmt.Printf("Stream sent event %d\n", i+1)
	}

	// Allow time for results to arrive.
	time.Sleep(2 * time.Second)
	fmt.Println("Event store stream send demo complete")
}

How It Works

client.SendEventStoreStream(ctx) opens a bidirectional gRPC stream; unlike SendEventStore, it does not block for each individual acknowledgement.
A background goroutine drains handle.Results — each result contains the broker-assigned EventID, a Sent boolean, and an error string if storage failed.
The send loop calls handle.Send(ev) five times without waiting for individual results, maximising write throughput while still receiving confirmations asynchronously.
defer handle.Close() flushes the send buffer and waits for the stream to close cleanly before the function returns.

Stream Send

Overview

Prerequisites

Code

How It Works

Related

On this page