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

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

client.NewQueueDownstreamReceiver(ctx) opens a persistent gRPC downstream stream that can handle multiple sequential Poll calls without reconnecting.
receiver.Poll(ctx, &kubemq.PollRequest{AutoAck: false}) fetches messages into an open transaction; the dm.TransactionID field identifies the transaction.
resp.AckAll() sends a single bulk-acknowledge that permanently removes all fetched messages from the queue in one broker operation.
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.

Stream Receive

Overview

Prerequisites

Code

How It Works

On this page

Stream Receive

Overview

Prerequisites

Code

How It Works

Related

On this page