Overview

Sending one queue message per call works fine for occasional traffic, but each call carries its own round trip. At high volume — event ingestion, sensor telemetry, log shipping — that per-call overhead caps your throughput well below what the connection can support.

QueueUpstream opens a persistent, bidirectional gRPC stream you reuse to push any number of batches. upstream.Send(refRequestID, messages) writes a batch and returns immediately; results arrive asynchronously on upstream.Results, correlated by refRequestID. Because the stream stays open, you can pipeline the next batch before the previous one's result comes back.

Gotchas: results are asynchronous — Send returning without error only means the write hit the stream, not that the broker processed it, so you must read upstream.Results (or upstream.Done) to confirm success. A stream error or disconnect ends the whole stream, not just one batch, so production senders need reconnect logic. Always Close() the upstream — an open stream holds server-side resources for the client's lifetime.

Prerequisites

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

Code

main.go

// Example: queues-stream/stream-send
//
// Demonstrates high-throughput queue message publishing using QueueUpstream.
// The bidirectional stream allows sending multiple messages efficiently
// with per-batch result confirmations.
//
// Channel: go-queues-stream.stream-send
// Client ID: go-queues-stream-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-queues-stream-stream-send-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues-stream.stream-send"

	// Open a bidirectional upstream stream for publishing.
	upstream, err := client.QueueUpstream(ctx)
	if err != nil {
		log.Fatalf("QueueUpstream: %v", err)
	}
	defer upstream.Close()

	// Send a batch of messages via the stream.
	msgs := []*kubemq.QueueMessage{
		kubemq.NewQueueMessage().SetChannel(channel).SetBody([]byte("stream-msg-1")),
		kubemq.NewQueueMessage().SetChannel(channel).SetBody([]byte("stream-msg-2")),
		kubemq.NewQueueMessage().SetChannel(channel).SetBody([]byte("stream-msg-3")),
	}
	if err := upstream.Send("req-batch-1", msgs); err != nil {
		log.Fatalf("Send: %v", err)
	}

	// Read the batch result.
	select {
	case res := <-upstream.Results:
		if res != nil {
			if res.IsError {
				log.Printf("Upstream error: %s", res.Error)
			} else {
				fmt.Printf("Stream sent %d messages, RefRequestID=%s\n",
					len(res.Results), res.RefRequestID)
			}
		}
	case <-upstream.Done:
		fmt.Println("Upstream stream done")
	case <-ctx.Done():
		log.Printf("Timed out: %v", ctx.Err())
	}
}

How It Works

client.QueueUpstream(ctx) opens a persistent bidirectional gRPC stream dedicated to high-throughput queue publishing; the returned *QueueUpstream is kept alive by defer upstream.Close().
upstream.Send(refRequestID, messages) sends a batch of *kubemq.QueueMessage values in a single stream frame, identified by refRequestID for correlation with the result.
Results arrive asynchronously on upstream.Results; the select reads the first result and inspects res.RefRequestID and res.IsError to confirm the batch was accepted.
Using QueueUpstream instead of repeated SendQueueMessage calls eliminates per-message connection overhead and allows pipelining multiple batches before waiting for results.

Stream Send

Overview

Prerequisites

Code

How It Works

On this page

Stream Send

Overview

Prerequisites

Code

How It Works

Related

On this page