Basic Pub/Sub

Overview

This tutorial builds the "hello world" of KubeMQ messaging: a publisher and a subscriber talking over the Events pattern. Events are fire-and-forget — the broker fans a message out to every subscriber currently listening on the channel and moves on. There's no persistence, no acknowledgment, and no replay, which makes this the pattern to reach for when you need low-latency, high-throughput broadcast (metrics ticks, live status updates, cache-invalidation signals) and can tolerate losing a message if nobody is listening at the moment it's sent.

You'll wire up subscribeToEvents { }, which returns a cold Flow you collect in a background coroutine, give the subscription a moment to register with the server, then call publishEvent(eventMessage { }) to publish. Every connected subscriber on the channel gets its own copy, as opposed to a consumer group where only one member would receive it. Gotchas: if the subscriber isn't fully established before you publish, the event is simply gone — there's no queue catching it, which is why the sample delays briefly before sending; and because delivery isn't acknowledged, a crashed or disconnected subscriber never knows it missed anything.

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

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

private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-events-basic-pubsub"
private const val CHANNEL = "kotlin-events.basic-pubsub"

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

    client.use {
        // Subscribe in background
        val job = launch {
            client.subscribeToEvents {
                channel = CHANNEL
            }.take(3).collect { msg ->
                println("Received: ${String(msg.body)} on ${msg.channel}")
            }
        }

        delay(500) // let subscription establish

        // Publish 3 events
        repeat(3) { i ->
            client.publishEvent(eventMessage {
                channel = CHANNEL
                body = "Hello #$i".toByteArray()
                metadata = "greeting"
            })
            println("Published event #$i")
        }

        job.join()
        println("Done.")
    }
}

How It Works

• subscribeToEvents { } returns a cold Flow that connects when collected.
• The subscription runs in a background coroutine via launch.
• take(3) limits collection to 3 events, then the Flow completes.
• publishEvent(eventMessage { }) uses the DSL builder to construct and send events.
• Events are fire-and-forget -- no delivery acknowledgement from subscribers.

Related

• Events Reference
• Consumer Group
• Stream Send