Stream Receive

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

receiveQueueMessages() fetches a batch under manual settlement — messages come back locked, not auto-removed — so nothing leaves the queue until you explicitly settle it. Each returned QueueMessageReceived is settled on its own: calling ack() removes it from the queue immediately, while an unacknowledged message is redelivered to the next poller once the visibility window expires.

Gotchas: a crash between receiving and acknowledging redelivers the whole batch, so processing must be idempotent; forgetting to call ack() doesn't lose the message, it just delays redelivery until the timeout; and a large pollMaxMessages with a slow per-message handler can hold the batch past the visibility window, triggering duplicate delivery mid-processing.

Prerequisites

• KubeMQ server running on localhost:50000
• Java SDK installed (implementation 'io.kubemq.sdk:kubemq-sdk-Java:3.1.1' (Gradle) or Maven dependency from Getting Started)

Code

package io.kubemq.example.queuesstream;

import io.kubemq.sdk.queues.*;
import java.util.UUID;

/**
 * StreamReceiveExample for Queues Stream
 */
public class StreamReceiveExample {
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-queues-stream-receive-client";
    private static final String CHANNEL = "java-queues.stream-receive";

    public static void main(String[] args) {
        // Create a client connected to the KubeMQ server
        try (QueuesClient client = QueuesClient.builder().address(ADDRESS).clientId(CLIENT_ID).build()) {
            // Create the queue channel
            client.createQueuesChannel(CHANNEL);

            // Send messages to the queue
            for (int i = 1; i <= 5; i++) {
                client.sendQueueMessage(QueueMessage.builder()
                        .channel(CHANNEL).body(("Message " + i).getBytes()).build());
            }

            // Poll for messages via stream
            System.out.println("Receiving messages via stream poll...\n");
            QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
                    .channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(5).build());

            // Process each message and acknowledge it
            response.getMessages().forEach(msg -> {
                System.out.println("  Received: " + new String(msg.getBody()));
                msg.ack();
            });
            System.out.println("\nReceived and acked " + response.getMessages().size() + " messages.");

            // Clean up resources
            client.deleteQueuesChannel(CHANNEL);
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }
    }
}

How It Works

• receiveQueueMessages() with pollMaxMessages(10) opens a downstream gRPC stream to the broker and returns up to 10 messages in one batch; the SDK closes the stream internally once the batch is delivered or pollWaitTimeoutInSeconds elapses.
• Each QueueMessageReceived.ack() sends a per-message acknowledgement; the broker removes acknowledged messages from the queue immediately.
• Unacknowledged messages (not acked or rejected before the visibility window expires) are redelivered to the next poller.
• Five messages are pre-sent, then all are received in a single receiveQueueMessages() call — this demonstrates the batch receive pattern that reduces round-trip overhead compared to polling one message at a time.

Related

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