# Stream Send (/sdks/kotlin/how-to/events-store/stream-send)



## Overview [#overview]

<Callout type="info" title="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](/sdks/kotlin/how-to/events/stream-send).
</Callout>

**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. &#x2A;*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 [#prerequisites]

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

## Code [#code]

```kotlin title="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 [#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.

## Related [#related]

* [Persistent Pub/Sub](/sdks/kotlin/tutorials/persistent-pubsub)
* [Events Stream Send](/sdks/kotlin/how-to/events/stream-send)
* [Events Store Reference](/sdks/kotlin/reference/events-store)