# Kotlin (/integrations/spring-boot/how-to/kotlin) ## Overview [#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](/integrations/spring-boot/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: ```kotlin title="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 [#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 [#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`: ```kotlin title="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) { runApplication(*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: ```kotlin title="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 [#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: ```kotlin title="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) { runApplication(*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 [#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](/integrations/spring-boot/reference/configuration). ```kotlin title="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: ```kotlin title="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 [#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`): ```yaml title="application.yml" kubemq: kotlin: dispatcher: io # default | io | unconfined ``` ```kotlin title="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)) } } ``` ## Next Steps [#next-steps]