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 receiveQueuesMessages sends a channel, maxItems, and waitTimeoutMs; 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. With autoAck = false each message is acknowledged individually via ack(), giving you a chance to skip one you can't process.

Gotchas: un-acked messages return to the queue only after the broker's visibility timeout, so a crashed or slow consumer can leave messages invisible to others for a while; the timeout bounds latency, not throughput, so a small maxItems on a busy queue means many round trips; and messages beyond your requested batch size simply wait for the next poll — they aren't dropped.

Prerequisites

KubeMQ server running on localhost:50000
Kotlin SDK installed (implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))

Code

PollModeExample.kt

package io.kubemq.sdk.examples.queuesstream

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.queues.queueMessage
import kotlinx.coroutines.runBlocking

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-queues-poll-mode"
private const val CHANNEL = "kotlin-queues.poll-mode"

fun main() = runBlocking {
    val client = KubeMQClient.queues {
        address = ADDRESS
        clientId = CLIENT_ID
    }

    client.use {
        try {
            client.createQueuesChannel(CHANNEL)

            // Send messages
            repeat(5) { i ->
                client.sendQueuesMessage(queueMessage {
                    channel = CHANNEL
                    body = "Poll msg ${i + 1}".toByteArray()
                })
            }

            // Poll in pull mode with maxItems=3
            println("=== Waiting Pull Mode ===\n")
            val response = client.receiveQueuesMessages {
                channel = CHANNEL
                maxItems = 3
                waitTimeoutMs = 10000
                autoAck = false
            }

            println("Received ${response.messages.size} messages:")
            response.messages.forEach { msg ->
                println("  ${String(msg.body)}")
                msg.ack()
            }

            // Cleanup remaining
            val cleanup = client.receiveQueuesMessages {
                channel = CHANNEL
                maxItems = 10
                waitTimeoutMs = 2000
                autoAck = true
            }
            println("\nCleanup: consumed ${cleanup.messages.size} remaining messages.")
        } finally {
            try { client.deleteQueuesChannel(CHANNEL) } catch (_: Exception) {}
        }
        println("Done.")
    }
}

How It Works

maxItems = 3 limits the batch to 3 messages per poll, even if more are available.
waitTimeoutMs controls the long-poll wait time -- the call returns when either maxItems messages are available or the timeout expires.
This pull-based pattern gives the consumer control over processing rate and batch size.
Remaining messages stay in the queue for subsequent polls.

Poll Mode

Overview

Prerequisites

Code

How It Works

On this page

Poll Mode

Overview

Prerequisites

Code

How It Works

Related

On this page