Stream Send

Overview

Sending one queue message per call works fine for occasional traffic, but each call carries its own round trip. At high volume — event ingestion, sensor telemetry, log shipping — that per-call overhead caps your throughput well below what the connection can support.

QueuesClient.sendQueueMessage() sends each message over the SDK's internal upstream gRPC stream, shared across every send on the client — so the connection cost is paid once, not per message. Each call is synchronous and returns a QueueSendResult with the broker-assigned message ID; per-message failures surface via result.isError() rather than an exception. createQueuesChannel() / deleteQueuesChannel() are separate management calls — sending doesn't require the channel to pre-exist, but creating it first ensures channel-level configuration like DLQ or TTL is applied before messages arrive.

Gotchas: because each call is synchronous, a sequential loop is bounded by round-trip latency — that's the tradeoff the example measures, and why high-volume producers should look at batch or async sending instead. Always check result.isError(); a broker-side rejection doesn't throw, so an ignored result can mean a message never reached the queue. The client is AutoCloseable — use try-with-resources so the stream and connection are released even on error.

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;

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

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

            System.out.println("Sending messages via stream...\n");
            long start = System.currentTimeMillis();

            // Send messages in a loop via stream
            for (int i = 1; i <= 10; i++) {
                QueueSendResult result = client.sendQueueMessage(QueueMessage.builder()
                        .id(UUID.randomUUID().toString()).channel(CHANNEL)
                        .body(("Stream message #" + i).getBytes()).build());
                System.out.println("  Sent #" + i + " -> " + result.getId());
            }

            long elapsed = System.currentTimeMillis() - start;
            System.out.println("\n10 messages sent in " + elapsed + "ms");

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

How It Works

• sendQueueMessage() sends each message individually via the SDK's internal upstream gRPC stream; the stream is shared across all sends on the same client, avoiding per-message connection overhead.
• The loop sends 10 messages sequentially and measures total elapsed time, demonstrating the throughput advantage of the persistent gRPC stream over reconnect-per-message approaches.
• Each sendQueueMessage() call is synchronous and returns a QueueSendResult with the broker-assigned message ID; errors per message are surfaced via result.isError().
• createQueuesChannel() and deleteQueuesChannel() are management calls — use them during setup/teardown; the channel is not required to exist before sending but creating it first ensures the channel configuration (DLQ, TTL) is applied.

Related

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