# Stream Receive (/sdks/go/how-to/queues/stream-receive)



## Overview [#overview]

A **downstream receiver** is the persistent-connection way to pull queue messages: instead of opening and tearing down a request for every batch, you open one gRPC stream and reuse it across many poll cycles. That matters for any consumer that runs continuously — a worker loop, a background processor — where reconnecting per batch would add latency and churn on both the client and the broker.

The receiver is created once with `NewQueueDownstreamReceiver`, then each call to `Poll` fetches a batch under a transaction, with `AutoAck: false` so nothing is removed from the queue until you explicitly settle it. Every message carries a `TransactionID`; acknowledging it — individually, or as a batch with `resp.AckAll()` — permanently removes it, while leaving it unacknowledged returns it for redelivery once the visibility timeout expires.

**Gotchas:** an unclosed receiver holds server-side state — always `Close()` it when the consumer shuts down; a crash between receiving and acknowledging redelivers the batch, so processing must be idempotent; and forgetting to set `AutoAck: false` silently drops the manual-settlement guarantee this pattern exists for.

## 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-receive
//
// Demonstrates receiving queue messages using NewQueueDownstreamReceiver + Poll.
// The receiver manages a persistent downstream stream with automatic reconnection.
//
// Channel: go-queues-stream.stream-receive
// Client ID: go-queues-stream-stream-receive-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-receive-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

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

	// Send a message to receive.
	_, err = client.SendQueueMessage(ctx, kubemq.NewQueueMessage().
		SetChannel(channel).
		SetBody([]byte("message to receive")))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("Message sent")

	// Allow message to be committed to the queue.
	time.Sleep(time.Second)

	// Open a downstream receiver.
	receiver, err := client.NewQueueDownstreamReceiver(ctx)
	if err != nil {
		log.Fatalf("NewQueueDownstreamReceiver: %v", err)
	}
	defer receiver.Close()

	// Poll for messages (manual ack).
	resp, err := receiver.Poll(ctx, &kubemq.PollRequest{
		Channel:            channel,
		MaxItems:           10,
		WaitTimeoutSeconds: 5,
		AutoAck:            false,
	})
	if err != nil {
		log.Fatalf("Poll: %v", err)
	}

	for _, dm := range resp.Messages {
		if dm.Message != nil {
			fmt.Printf("Received: body=%s tx=%s\n", dm.Message.Body, dm.TransactionID)
		}
	}

	// Ack all received messages.
	if len(resp.Messages) > 0 {
		if err := resp.AckAll(); err != nil {
			log.Printf("AckAll: %v", err)
		}
		fmt.Println("Messages acknowledged")
	}
}

```

## How It Works [#how-it-works]

1. `client.NewQueueDownstreamReceiver(ctx)` opens a persistent gRPC downstream stream that can handle multiple sequential `Poll` calls without reconnecting.
2. `receiver.Poll(ctx, &kubemq.PollRequest{AutoAck: false})` fetches messages into an open transaction; the `dm.TransactionID` field identifies the transaction.
3. `resp.AckAll()` sends a single bulk-acknowledge that permanently removes all fetched messages from the queue in one broker operation.
4. Reusing a `QueueDownstreamReceiver` across multiple poll cycles is more efficient than calling `PollQueue` repeatedly, because the underlying gRPC connection and server-side state are preserved.

## 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)