Overview

The optional kubemq-spring-boot-starter-kotlin module layers idiomatic Kotlin on top of the core starter. It does not change the messaging model — the same KubeMQTemplate, @KubeMQ*Listener annotations, and kubemq.* configuration described in the Concepts page still apply. What it adds is a coroutine-friendly surface: suspend send extensions on KubeMQTemplate, cold Flow subscriptions on the SDK clients, suspend listener method support, and a type-safe kubemq { } configuration DSL.

Add the module alongside the starter:

build.gradle.kts

dependencies {
    implementation("io.kubemq:kubemq-spring-boot-starter:1.0.0")
    implementation("io.kubemq:kubemq-spring-boot-starter-kotlin:1.0.0")
}

Surface

API	What it adds	Backing example
`KubeMQTemplate.send*Suspend(...)`	`suspend` send extensions that bridge the template's `CompletableFuture` async methods to coroutines	`kotlin-coroutine-publish`
`PubSubClient.eventsFlow / eventsStoreFlow`, `QueuesClient.queuesFlow`, `CQClient.commandsFlow / queriesFlow`	cold `Flow` subscriptions; cancelling the collector tears down the gRPC stream	`kotlin-flow-subscribe`
`kubemq { }` DSL (`KubeMQConfigurerDsl`)	programmatic, type-safe configuration as a bean	`kotlin-dsl-config`
`suspend` listener methods	`@KubeMQ*Listener` methods may be `suspend` functions, dispatched on `kubemq.kotlin.dispatcher`	—

Suspend Send Extensions

Each KubeMQTemplate send has a suspend counterpart that delegates to the matching *Async method and suspends until the broker accepts the message (or, for commands and queries, until the response arrives), honoring structured-concurrency cancellation.

Extension	Delegates to	Returns
`sendEventSuspend(channel, data[, tags])`	`sendEventAsync`	`Unit`
`sendEventStoreSuspend(channel, data[, tags])`	`sendEventStoreAsync`	`Unit`
`sendQueueMessageSuspend(channel, data[, tags])`	`sendQueueMessageAsync`	`Unit`
`sendCommandSuspend(channel, data, timeout)`	`sendCommandAsync`	`CommandResponseMessage`
`sendQuerySuspend(channel, data, timeout)`	`sendQueryAsync`	`QueryResponseMessage`

The kotlin-coroutine-publish example publishes from inside runBlocking:

CoroutinePublishApplication.kt

package io.kubemq.spring.boot.examples.kotlin

import io.kubemq.spring.boot.kotlin.sendEventSuspend
import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate
import kotlinx.coroutines.runBlocking
import org.slf4j.LoggerFactory
import org.springframework.boot.ApplicationArguments
import org.springframework.boot.ApplicationRunner
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication
import org.springframework.stereotype.Component

@SpringBootApplication
class CoroutinePublishApplication

fun main(args: Array<String>) {
    runApplication<CoroutinePublishApplication>(*args)
}

@Component
class CoroutinePublishRunner(private val template: KubeMQTemplate) : ApplicationRunner {

    private val log = LoggerFactory.getLogger(CoroutinePublishRunner::class.java)

    override fun run(args: ApplicationArguments) {
        runBlocking {
            for (i in 1..5) {
                template.sendEventSuspend("spring-kotlin.coroutine-publish", "Coroutine event #$i")
                log.info("Published coroutine event #{}", i)
            }
            log.info("Kotlin coroutine publish example completed.")
        }
    }
}

A suspend fun in a @Service reads exactly like the imperative API but composes inside any coroutine scope:

PublishService.kt

@Service
class PublishService(private val template: KubeMQTemplate) {

    suspend fun publishOrder(order: String) {
        // Fire-and-forget event.
        template.sendEventSuspend("spring-events.orders", order)

        // Persisted events-store message with tags.
        template.sendEventStoreSuspend(
            "spring-events-store.orders",
            order,
            mapOf("source" to "order-service"),
        )
    }
}

Flow Subscriptions

The module exposes cold Flow builders on the three SDK clients. Each opens the underlying gRPC subscription when the Flow is collected and tears it down when the collecting coroutine is cancelled.

Extension	Receiver	Emits
`eventsFlow(channel, group = "")`	`PubSubClient`	`EventMessageReceived`
`eventsStoreFlow(channel, group = "", startPosition = StartNewOnly)`	`PubSubClient`	`EventMessageReceived`
`queuesFlow(channel, pollInterval, ...)`	`QueuesClient`	`QueueReceivedMessage`
`commandsFlow(channel, group = "")`	`CQClient`	`CommandReceived`
`queriesFlow(channel, group = "")`	`CQClient`	`QueryReceived`

The kotlin-flow-subscribe example subscribes inside a launched coroutine and cancels the job to stop receiving:

FlowSubscribeApplication.kt

package io.kubemq.spring.boot.examples.kotlin

import io.kubemq.sdk.pubsub.PubSubClient
import io.kubemq.spring.boot.kotlin.sendEventSuspend
import io.kubemq.spring.boot.autoconfigure.template.KubeMQTemplate
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.slf4j.LoggerFactory
import org.springframework.boot.ApplicationArguments
import org.springframework.boot.ApplicationRunner
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication
import org.springframework.stereotype.Component

@SpringBootApplication
class FlowSubscribeApplication

fun main(args: Array<String>) {
    runApplication<FlowSubscribeApplication>(*args)
}

@Component
class FlowSubscribeRunner(
    private val template: KubeMQTemplate,
    private val pubSubClient: PubSubClient
) : ApplicationRunner {

    private val log = LoggerFactory.getLogger(FlowSubscribeRunner::class.java)

    override fun run(args: ApplicationArguments) {
        runBlocking {
            val subJob: Job = launch {
                try {
                    pubSubClient.subscribeToEvents {
                        channel = "spring-kotlin.flow-subscribe"
                    }.collect { event ->
                        val body = String(event.body)
                        log.info("Flow received: channel={} body={}", event.channel, body)
                    }
                } catch (e: CancellationException) {
                    throw e
                } catch (e: Exception) {
                    log.error("Subscription error: {}", e.message, e)
                }
            }
            log.info("Kotlin flow subscription started")

            delay(500)
            for (i in 1..3) {
                template.sendEventSuspend("spring-kotlin.flow-subscribe", "Kotlin flow event #$i")
                log.info("Published flow event #{}", i)
            }
            delay(1000)

            subJob.cancel()
            subJob.join()
            log.info("Kotlin flow subscribe example completed.")
        }
    }
}

The Kotlin SDK uses EventMessageReceived for both events and events store; store messages carry a non-zero sequence. Cancelling the collecting coroutine (here via subJob.cancel()) automatically closes the underlying gRPC subscription stream.

Configuration DSL

Instead of YAML, expose a KubeMQConfigurerDsl bean built with the kubemq { } builder. It maps one-to-one onto KubeMQProperties, so the tls { } and connection { } blocks set the same fields documented in the Configuration Reference.

KubeMQConfig.kt

import io.kubemq.spring.boot.kotlin.KubeMQConfigurerDsl
import io.kubemq.spring.boot.kotlin.kubemq
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import java.time.Duration

@Configuration
class KubeMQConfig {

    @Bean
    fun kubemqConfigurer(): KubeMQConfigurerDsl = kubemq {
        address = "broker.example.com:50000"
        clientId = "my-service"
        tls {
            enabled = true
            certFile = "/certs/client.pem"
            keyFile = "/certs/client-key.pem"
            caCertFile = "/certs/ca.pem"
        }
        connection {
            timeout = Duration.ofSeconds(15)
        }
    }
}

The kotlin-dsl-config example pairs the DSL with property injection — @Value("\${kubemq.address}") reads back the resolved configuration, and a coroutine then publishes through the suspend extension:

DslConfigApplication.kt

@Component
class DslConfigRunner(
    private val template: KubeMQTemplate,
    @Value("\${kubemq.address}") private val address: String,
    @Value("\${kubemq.client-id}") private val clientId: String
) : ApplicationRunner {

    private val log = LoggerFactory.getLogger(DslConfigRunner::class.java)

    override fun run(args: ApplicationArguments) {
        log.info("KubeMQ address: {}", address)
        log.info("KubeMQ client-id: {}", clientId)
        runBlocking {
            template.sendEventSuspend("spring-kotlin.dsl-config", "DSL config event")
            log.info("Sent event via Kotlin DSL config")
        }
    }
}

Suspend Listener Methods

With the Kotlin module on the classpath, a @KubeMQ*Listener / @KubeMQ*Handler method may be declared suspend. The starter detects the trailing Continuation parameter at validation time and invokes the method inside a per-container coroutine scope. The dispatcher is selected by kubemq.kotlin.dispatcher (default, io, or unconfined; default default):

application.yml

kubemq:
  kotlin:
    dispatcher: io   # default | io | unconfined

OrderListener.kt

@Component
class OrderListener {

    @KubeMQEventListener(channels = "spring-events.orders")
    suspend fun onOrder(event: EventMessageReceived) {
        // suspend work runs on the configured dispatcher
        process(String(event.body))
    }
}

Kotlin

Overview

Surface

Suspend Send Extensions

Flow Subscriptions

Configuration DSL

Suspend Listener Methods

Next Steps

Events & Events Store

Concepts

Configuration Reference

On this page