Overview

Production connections drop for reasons that have nothing to do with your application code: the broker restarts during a rolling upgrade, a load balancer fails over, a network blip severs the connection. Without built-in reconnection, every client in your fleet needs its own hand-rolled retry-connect loop, and it's easy to get the backoff wrong — too aggressive and you hammer a recovering broker, too slow and you leave the application dark longer than necessary. Automatic reconnection moves that logic into the client itself, so the connection self-heals without any code the application has to write or maintain.

It works by configuring the reconnection { } DSL on KubeMQClient.pubSub { } — initialBackoffMs, maxBackoffMs, multiplier, and maxRetries shape the backoff curve. client.connectionState is a Flow<ConnectionState> that emits every transition (Idle, Connecting, Ready, Reconnecting(attempt), Closed) in real time, so the application can collect it for diagnostics instead of polling. Gotchas: during a Reconnecting state, outgoing calls like publishEvent block until the connection is re-established rather than failing fast, which can back up a hot publish loop if the outage is long; maxRetries = 0 means infinite retries, so a permanently dead broker will be retried forever unless you add your own circuit breaker; and collecting connectionState on the same coroutine that also does the publishing can deadlock — run the monitor in its own launch block as shown.

Prerequisites

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

Code

ReconnectionExample.kt

package io.kubemq.sdk.examples.errorhandling

import io.kubemq.sdk.client.ConnectionState
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.eventMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.takeWhile
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val client = KubeMQClient.pubSub {
        address = "localhost:50000"
        clientId = "kotlin-error-handling-reconnection"
        reconnection {
            initialBackoffMs = 1000
            maxBackoffMs = 30_000
            multiplier = 2.0
            maxRetries = 10 // 0 = infinite retries
        }
    }

    client.use {
        // Monitor connection state changes
        val monitorJob = launch {
            client.connectionState
                .takeWhile { it !is ConnectionState.Closed }
                .collect { state ->
                    when (state) {
                        is ConnectionState.Idle ->
                            println("[state] Idle")
                        is ConnectionState.Connecting ->
                            println("[state] Connecting...")
                        is ConnectionState.Ready ->
                            println("[state] Connected and ready")
                        is ConnectionState.Reconnecting ->
                            println("[state] Reconnecting, attempt #${state.attempt}")
                        is ConnectionState.Closed ->
                            println("[state] Closed")
                    }
                }
        }

        // Initial connection
        val info = client.ping()
        println("Connected to ${info.host} v${info.version}")

        // Publish events periodically -- if broker goes down, reconnection
        // will kick in automatically. Outgoing calls will block until
        // the connection is re-established.
        println("\nPublishing events (stop broker to see reconnection)...")
        repeat(10) { i ->
            try {
                client.publishEvent(eventMessage {
                    channel = "kotlin-error-handling.reconnection"
                    body = "Message #$i".toByteArray()
                })
                println("Published #$i")
            } catch (e: Exception) {
                println("Publish #$i failed: ${e.message}")
            }
            delay(2000)
        }

        monitorJob.cancel()
        println("Done.")
    }
}

How It Works

The reconnection { } DSL configures exponential backoff: initial delay, max delay, multiplier, and max retries.
client.connectionState is a Flow<ConnectionState> that emits state changes in real time.
State transitions: Idle -> Connecting -> Ready -> Reconnecting(attempt) -> Ready or Closed.
During reconnection, outgoing calls block until the connection is re-established.
Set maxRetries = 0 for infinite retry attempts.
The backoff sequence with the example config: 1s, 2s, 4s, 8s, 16s, 30s (capped).

Reconnection

Overview

Prerequisites

Code

How It Works

Related

On this page