Stream Send

Overview

Publishing events one at a time means each call pays its own round-trip: write the request, wait on the connection, then move to the next event. That's fine for occasional notifications, but it caps throughput when you need to push hundreds or thousands of events per second — log forwarding, sensor telemetry, change-data-capture feeds — where per-call overhead dominates.

publishEventStream() accepts a Flow<EventMessage> and opens one bidirectional gRPC stream to send it, returning a Flow<EventSendResult> you can collect to see per-event outcomes. Because the whole batch rides one open connection instead of one call per event, a cold flow { } producer can emit events as fast as the stream can take them without waiting on a broker round-trip for each one.

Gotchas: unlike a plain exception, a failed send here doesn't throw — it shows up as a result.sent == false entry in the result flow, so you must collect and check it or failures pass silently. Events are still fire-and-forget pub/sub underneath: no subscriber means a streamed event is dropped just like a regular one. Because the flow is cold, nothing is sent until something collects publishEventStream(eventFlow) — building the flow alone doesn't publish 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.flow
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

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

    client.use {
        // Subscribe to collect results
        val subJob = launch {
            client.subscribeToEvents {
                channel = "kotlin-events.stream-send"
            }.take(100).collect { msg ->
                println("Received: ${String(msg.body)}")
            }
        }

        delay(500)

        // High-throughput stream of events
        val eventFlow = flow {
            repeat(100) { i ->
                emit(eventMessage {
                    channel = "kotlin-events.stream-send"
                    body = "Stream message #$i".toByteArray()
                    metadata = "batch"
                })
            }
        }

        // publishEventStream returns a Flow of send results
        client.publishEventStream(eventFlow).collect { result ->
            if (!result.sent) {
                println("Failed to send ${result.id}: ${result.error}")
            }
        }

        println("All 100 events streamed.")
        delay(1000)
        subJob.cancel()
    }
}

How It Works

• publishEventStream() accepts a Flow<EventMessage> and returns a Flow<EventSendResult>.
• The gRPC bidirectional stream provides higher throughput than individual publishEvent() calls.
• Failed sends are reported in the result flow with sent = false and an error message.
• The flow { } builder creates a cold flow that emits messages on demand.
• This pattern is ideal for batch ingestion, log forwarding, or high-frequency sensor data.

Related

• Basic Pub/Sub
• Events Store Stream Send
• Events Reference