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.

client.sendQueuesMessage() sends each message over the SDK's internal gRPC streaming transport, reused across calls instead of reconnecting per message — so a sequential loop of sends stays cheap relative to opening a new connection each time. Each call is a suspend function returning the broker-assigned messageId, and it's this per-call round trip the example measures with System.currentTimeMillis() to show the real cost of a synchronous send loop.

Gotchas: because each call suspends until the broker acknowledges it, a sequential loop is bounded by round-trip latency — for genuinely high-throughput ingestion, look at batching or concurrent coroutines instead of a tight one-at-a-time loop. Wrap the client in use { } so the connection is released even if a send throws partway through. Clean up any channel you create explicitly — deleteQueuesChannel() isn't automatic and leftover test channels accumulate in a shared environment.

Prerequisites

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

Code

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-stream-send"
private const val CHANNEL = "kotlin-queues.stream-send"

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

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

            println("Sending messages via stream...\n")
            val start = System.currentTimeMillis()

            // Send messages in a loop via stream
            repeat(10) { i ->
                val result = client.sendQueuesMessage(queueMessage {
                    channel = CHANNEL
                    body = "Stream message #${i + 1}".toByteArray()
                })
                println("  Sent #${i + 1} -> ${result.messageId}")
            }

            val elapsed = System.currentTimeMillis() - start
            println("\n10 messages sent in ${elapsed}ms")

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

How It Works

• Messages are sent sequentially using sendQueuesMessage() inside a loop.
• The SDK internally uses gRPC streaming for efficient transport.
• Throughput is measured with System.currentTimeMillis() for benchmarking.
• Channel is created and cleaned up as part of the example.

Related

• Stream Receive
• Batch Send
• Queues Reference