Send & Receive

Overview

Queue send/receive is the foundational operation for guaranteed-delivery, point-to-point messaging: you reach for it whenever work needs to survive past the moment it's created and be handled by exactly one consumer, not broadcast to every subscriber. Unlike pub/sub, a queued message sits durably on the broker until something pulls it, so the producer and consumer never need to be online at the same time — a slow or offline worker adds latency, it doesn't drop the message.

This tutorial builds the smallest possible version of that round trip, then shows both outcomes of manual settlement: sendQueuesMessage enqueues a message on a channel, and receiveQueuesMessages pulls it back within a bounded waitTimeoutMs. With autoAck = false, each message must be explicitly confirmed with msg.ack() once processing succeeds, or returned to the queue with msg.reject() when it can't be handled.

Gotchas: if your handler crashes before calling ack() or reject(), the message stays in the queue and becomes available for redelivery once its visibility timeout elapses — write handlers that tolerate seeing the same message twice. A rejected message goes back to the queue for redelivery unless a dead-letter policy is configured, so reject() alone doesn't remove a poison message from circulation. And calling receiveQueuesMessages against an empty queue isn't an error; it just blocks up to waitTimeoutMs and returns no messages.

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

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

fun main() = runBlocking {
    val client = KubeMQClient.queues {
        address = "localhost:50000"
        clientId = "kotlin-queues-send-receive"
    }

    client.use {
        val channel = "kotlin-queues.send-receive"

        // Send a message
        val sendResult = client.sendQueuesMessage(queueMessage {
            this.channel = channel
            body = "Hello from queue".toByteArray()
            metadata = "task"
            tags["priority"] = "high"
        })
        println("Sent: id=${sendResult.messageId}, error=${sendResult.error}")

        // Receive with manual ack
        val response = client.receiveQueuesMessages {
            this.channel = channel
            maxItems = 10
            waitTimeoutMs = 5000
            autoAck = false
        }

        if (response.isError) {
            println("Receive error: ${response.error}")
            return@runBlocking
        }

        for (msg in response.messages) {
            println("Received: ${String(msg.body)}, tags=${msg.tags}")
            // Acknowledge the message
            msg.ack()
            println("  Acked message ${msg.id}")
        }

        // Send another and reject it
        client.sendQueuesMessage(queueMessage {
            this.channel = channel
            body = "Will be rejected".toByteArray()
        })
        val resp2 = client.receiveQueuesMessages {
            this.channel = channel
            maxItems = 1
            waitTimeoutMs = 5000
        }
        resp2.messages.forEach { msg ->
            msg.reject()
            println("Rejected message ${msg.id}")
        }

        println("Done.")
    }
}

How It Works

• sendQueuesMessage(queueMessage { }) sends a message with body, metadata, and tags.
• receiveQueuesMessages { } polls for messages with configurable maxItems and waitTimeoutMs.
• With autoAck = false, each message must be explicitly acknowledged via msg.ack() or rejected via msg.reject().
• Rejected messages are returned to the queue for redelivery (unless DLQ policy is configured).

Related

• Queues Reference
• Ack & Reject
• Batch Send