Poll Mode

Overview

Poll mode is a pull-based way to consume queue messages: the consumer decides exactly when to ask for work and how much, instead of holding an open stream the broker pushes into. That control matters for batch jobs, cron-triggered workers, and any consumer that only runs intermittently and would rather ask "is there anything for me?" than keep a subscription alive.

A single call to PollQueue sends a channel, MaxItems, and WaitTimeoutSeconds; the broker holds the request open as a long poll and returns once enough messages are available or the timeout elapses, so the call never spins on an empty queue. Auto-ack (the PollRequest default) settles the whole batch on delivery, with no separate acknowledgment step.

Gotchas: auto-ack removes messages the instant they're delivered — a crash mid-processing loses them, so switch to NewQueueDownstreamReceiver with manual ack when work can fail; the timeout bounds latency, not throughput, so a small MaxItems on a busy queue means many round trips; and PollQueue wraps the same receiver machinery as streaming — use a persistent stream instead for continuous, low-latency consumption.

Prerequisites

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

Code

// Example: queues-stream/poll-mode
//
// Demonstrates PollQueue for simple single-shot queue polling.
// PollQueue is a high-level abstraction that handles the receiver lifecycle
// automatically with auto-ack.
//
// Channel: go-queues-stream.poll-mode
// Client ID: go-queues-stream-poll-mode-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-poll-mode-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues-stream.poll-mode"

	// Send some messages to poll.
	for i := 1; i <= 3; i++ {
		_, err := client.SendQueueMessage(ctx, kubemq.NewQueueMessage().
			SetChannel(channel).
			SetBody(fmt.Appendf(nil, "poll-msg-%d", i)))
		if err != nil {
			log.Fatal(err)
		}
	}
	fmt.Println("Sent 3 messages")

	// PollQueue: single-shot poll with auto-ack.
	pollResp, err := client.PollQueue(ctx, &kubemq.PollRequest{
		Channel:            channel,
		MaxItems:           10,
		WaitTimeoutSeconds: 3,
	})
	if err != nil {
		log.Fatalf("PollQueue: %v", err)
	}
	fmt.Printf("PollQueue: %d messages\n", len(pollResp.Messages))
	for _, dm := range pollResp.Messages {
		if dm.Message != nil {
			fmt.Printf("  body=%s\n", dm.Message.Body)
		}
	}
}

How It Works

1. client.PollQueue(ctx, &kubemq.PollRequest{...}) is a one-shot convenience wrapper: it creates a temporary QueueDownstreamReceiver internally, polls, auto-acknowledges, and tears down the receiver — all in one call.
2. Without an explicit AutoAck field, PollRequest defaults to AutoAck: true, so messages are automatically removed from the queue upon delivery without a separate ack step.
3. MaxItems: 10 and WaitTimeoutSeconds: 3 bound the call: it returns as soon as 10 messages are available, or after 3 seconds if fewer arrive.
4. Use NewQueueDownstreamReceiver directly when you need manual ack/nack control or want to reuse the stream across multiple polls (see Ack & Reject, Ack All).

Related

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