Stream Send

Overview

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.

send_event_store_stream() opens a single bidirectional stream and hands back a handle: handle.send(event) pushes events onto it as fast as you can call it, while handle.results() is a separate receiver that yields each event's event_id, sent confirmation, and any error, independent of send order. Gotchas: results can arrive out of order relative to sends, so correlate them by event_id rather than assuming a 1:1 positional match; calling handle.close() before draining results() 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
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::EventStoreBuilder;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    let mut handle = client.send_event_store_stream().await?;
    let channel = "rust-events-store.stream-send";

    for i in 0..50 {
        let event = EventStoreBuilder::new()
            .channel(channel)
            .body(format!("streamed-{}", i).into_bytes())
            .build();
        handle.send(event).await?;
    }

    println!("Sent 50 persistent events via stream");

    while let Ok(result) = handle.results().try_recv() {
        if !result.sent {
            eprintln!("Failed: {}", result.error);
        }
    }

    handle.close();
    client.close().await?;
    Ok(())
}

How It Works

• send_event_store_stream() opens a bidirectional stream for persistent event publishing.
• Per-message results are available via the results() receiver.
• Each result includes event_id, sent status, and any error message.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• Events Store overview
• Rust SDK Reference
• Persistent Pub/Sub