Peek Messages

Overview

Peeking lets you look at what's sitting in a queue without touching it — the messages stay exactly where they are, still waiting for whichever consumer eventually receives them. It's the tool you reach for when you need visibility into queue state — checking backlog depth, inspecting payloads while debugging a stuck pipeline, or building an operational dashboard — without risking a collision with real consumers competing for the same work.

The Go SDK has no dedicated peek call; you get the same effect through receiver.Poll(ctx, &kubemq.PollRequest{AutoAck: false}). The broker hands back a snapshot of messages inside an open transaction but never marks them as delivered — they stay invisible to other consumers only for the life of that transaction. Let the receiver close without calling AckAll() or NackAll() and the transaction rolls back on the server-side timeout, making the messages visible again, unconsumed.

Gotchas: this isn't a true non-locking peek — while the transaction is open, other consumers can't see those messages, so a long-lived unsettled poll can quietly stall a queue. Calling AckAll() or NackAll() settles the transaction and ends the peek — AckAll() consumes the messages, not merely peeks at them. For dashboards or depth checks where you never intend to settle the transaction, keep the timeout short so messages don't sit invisible longer than necessary.

Prerequisites

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

Code

// Example: queues/peek-messages
//
// Demonstrates peeking at queue messages without consuming them (transactional).
// Messages are received with AutoAck=false and the transaction is never
// settled — the receiver closes without calling AckAll() or NackAll(), so
// the broker rolls the transaction back and the messages stay in the queue.
//
// Channel: go-queues.peek-messages
// Client ID: go-queues-peek-messages-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-peek-messages-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues.peek-messages"

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

	// Create a downstream receiver for transactional (non-auto-ack) polling.
	receiver, err := client.NewQueueDownstreamReceiver(ctx)
	if err != nil {
		log.Fatal(err)
	}

	// Poll with AutoAck=false — messages are held in a transaction until settled.
	resp, err := receiver.Poll(ctx, &kubemq.PollRequest{
		Channel:            channel,
		MaxItems:           10,
		WaitTimeoutSeconds: 5,
		AutoAck:            false,
	})
	if err != nil {
		log.Fatal(err)
	}
	if resp.IsError {
		log.Fatalf("Poll failed: %s", resp.Error)
	}
	fmt.Printf("Peeked: %d messages (not settled)\n", len(resp.Messages))
	for _, dsMsg := range resp.Messages {
		fmt.Printf("  body=%s\n", dsMsg.Message.Body)
	}

	// Close without calling AckAll() or NackAll() — the open transaction is
	// never settled, so the broker rolls it back and the messages remain in
	// the queue for the next consumer, unconsumed.
	if err := receiver.Close(); err != nil {
		log.Fatalf("Close failed: %s", err)
	}
	fmt.Println("Receiver closed without settling; messages remain queued")
}

How It Works

1. client.NewQueueDownstreamReceiver(ctx) creates a persistent downstream stream; receiver.Poll(ctx, &kubemq.PollRequest{AutoAck: false}) fetches messages into an open transaction without immediately removing them.
2. With AutoAck: false the messages are held invisibly by the broker for the duration of the transaction; other consumers cannot receive them during this window.
3. This example never calls resp.AckAll() or resp.NackAll() — settling the transaction either way would consume or explicitly requeue the messages, which is not a peek.
4. receiver.Close() tears down the stream without settling the open transaction, so the broker rolls it back once the server-side timeout elapses: the peeked messages become visible to other consumers again, unconsumed. This is the genuinely non-destructive path — if you need to consume the batch instead, settle it explicitly with AckAll() (see Ack All).

Related

• Pattern overview
• Go SDK Reference
• Send & Receive
• Ack All