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.

client.send_event_stream() opens one bidirectional gRPC stream up front and returns a stream handle. Each subsequent stream.send(event).await writes onto that already-open stream instead of negotiating a new call, so a sender loop isn't blocked waiting on a broker round-trip for every event — sends are buffered internally and flow onto the wire as fast as the connection allows.

Gotchas: because sends don't wait on a per-message round-trip, write failures surface asynchronously on stream.errors() — you must poll or drain that receiver, or failures go unnoticed. Events are still fire-and-forget pub/sub underneath: no subscriber means a streamed event is dropped just like a regular one. The internal send buffer is bounded (256 messages); a stream held open far longer than the broker can drain it will eventually apply backpressure on send().await.

Prerequisites

• KubeMQ server running on localhost:50000
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::EventBuilder;

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

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

    for i in 0..100 {
        let event = EventBuilder::new()
            .channel(channel)
            .metadata(format!("stream-event-{}", i))
            .body(format!("payload-{}", i).into_bytes())
            .build();

        stream.send(event).await?;
    }

    println!("Sent 100 events via stream to channel: {}", channel);

    while let Ok(err) = stream.errors().try_recv() {
        eprintln!("Stream error: {}", err);
    }

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

How It Works

• send_event_stream() opens a bidirectional gRPC stream for high-throughput publishing.
• Events are buffered internally (256-message channel) for non-blocking sends.
• Errors are delivered asynchronously via the errors() receiver.
• 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 overview
• Rust SDK Reference
• Basic Pub/Sub