Overview

Which to use

This page covers high-throughput Events Store (persistent, replayable) streaming via publishEventStoreStream(). For the fire-and-forget equivalent, see Events Stream Send.

Stream send decouples publishing a persistent event from waiting for its storage confirmation, so you can push a large batch onto the wire without stalling on a round trip per message. A single-shot publish call is fine for occasional writes, but if you're bulk-loading history, replicating a firehose of records, or backfilling an Events Store channel, a request/response call per event turns network latency into your throughput ceiling.

publishEventStoreStream() accepts a Flow<EventStoreMessage> and opens one long-lived bidirectional gRPC connection instead of one RPC per message, returning a Flow<EventSendResult> that emits each event's id, sent confirmation, and error as storage completes, independent of send order. Gotchas: results can arrive out of order relative to sends, so correlate them by id rather than assuming a 1:1 positional match; cancelling collection of the results flow before the send flow finishes can drop confirmations for events still in flight; and for low-volume or one-off publishing, the extra bookkeeping isn't worth it — reach for a plain publish call instead.

Prerequisites

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

Code

StreamSendExample.kt

package io.kubemq.sdk.examples.eventsstore

import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.pubsub.StartPosition
import io.kubemq.sdk.pubsub.eventStoreMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flow
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-store-stream-send"
private const val CHANNEL = "kotlin-events-store.stream-send"

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

    client.use {
        // Subscribe to collect results
        val subJob = launch {
            client.subscribeToEventsStore {
                channel = CHANNEL
                startPosition = StartPosition.StartNewOnly
            }.take(100).collect { msg ->
                println("Received: seq=${msg.sequence} ${String(msg.body)}")
            }
        }

        delay(500)

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

        println("Streaming 100 persistent events...")
        val start = System.currentTimeMillis()
        client.publishEventStoreStream(eventFlow).collect { result ->
            if (!result.sent) {
                println("Failed to send ${result.id}: ${result.error}")
            }
        }
        val elapsed = System.currentTimeMillis() - start
        println("100 events streamed in ${elapsed}ms (${100 * 1000.0 / elapsed} msg/s)")

        delay(2000)
        subJob.cancel()
        println("Done.")
    }
}

How It Works

publishEventStoreStream() accepts a Flow<EventStoreMessage> and returns a Flow<EventSendResult>.
The gRPC bidirectional stream keeps a single long-lived connection open rather than one RPC per message.
All streamed events are persisted and can be replayed later.
Throughput metrics are measured with System.currentTimeMillis() for benchmarking.

Stream Send

Overview

Prerequisites

Code

How It Works

Related

On this page