# Stream Send (/sdks/go/how-to/queues/stream-send)



## Overview [#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 [#prerequisites]

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

## Code [#code]

```go title="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 [#how-it-works]

1. `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()`.
2. `upstream.Send(refRequestID, messages)` sends a batch of `*kubemq.QueueMessage` values in a single stream frame, identified by `refRequestID` for correlation with the result.
3. 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.
4. Using `QueueUpstream` instead of repeated `SendQueueMessage` calls eliminates per-message connection overhead and allows pipelining multiple batches before waiting for results.

## Related [#related]

* [Pattern overview](/learn/queues/getting-started)
* [Go SDK Reference](/sdks/go/reference)
* [Send & Receive](/sdks/go/tutorials/send-receive)
* [Ack All](/sdks/go/how-to/queues/ack-all)