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->SendEventStream() opens one bidirectional gRPC stream up front and returns a handle. Each subsequent handle->Send(event) writes a frame directly 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.

Gotchas: because sends don't wait on a per-message round-trip, write failures surface asynchronously through the error callback passed to SendEventStream — register it up front, 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. Always call handle->Close() explicitly when done; don't rely solely on the handle's destructor in code paths where cleanup order matters.

Prerequisites

KubeMQ server running on localhost:50000
C++ SDK installed (vcpkg or CMake FetchContent)
C++17 compiler (GCC 9+, Clang 9+, MSVC 2019+)

Code

main.cc

// Example: events/stream_send
//
// Demonstrates high-throughput event publishing using SendEventStream.
// A bidirectional stream is opened for sending multiple events efficiently.
//
// Channel: cpp-events.stream-send
// Client ID: cpp-events-stream-send-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).

#include <kubemq/kubemq.h>

#include <chrono>
#include <iostream>
#include <string>
#include <thread>

int main() {
    std::cout << "[1] Connecting to localhost:50000" << std::endl;

    kubemq::ClientOptions options;
    options.set_address("localhost", 50000);
    options.set_client_id("cpp-events-stream-send-client");

    auto client_result = kubemq::Client::Create(options);
    if (!client_result.ok()) {
        std::cerr << "[ERROR] Failed to create client: " << client_result.status().message()
                  << std::endl;
        return 1;
    }
    auto& client = *client_result;

    std::string channel = "cpp-events.stream-send";

    // Subscribe to verify events arrive.
    std::cout << "[2] Subscribing to channel " << channel << std::endl;
    auto sub_result = client->SubscribeToEvents(
        channel, "",
        [](const kubemq::EventReceive& event) {
            std::cout << "[5] Stream received: channel=" << event.channel << " body=" << event.body
                      << std::endl;
        },
        [](const kubemq::Status& err) {
            std::cerr << "[ERROR] Subscription error: " << err.message() << std::endl;
        });
    if (!sub_result.ok()) {
        std::cerr << "[ERROR] SubscribeToEvents: " << sub_result.status().message() << std::endl;
        return 1;
    }
    auto& sub = *sub_result;

    // Open a stream for high-throughput publishing.
    std::cout << "[3] Opening event stream" << std::endl;
    auto handle_result = client->SendEventStream([](const kubemq::Status& err) {
        std::cerr << "[ERROR] Stream send error: " << err.message() << std::endl;
    });
    if (!handle_result.ok()) {
        std::cerr << "[ERROR] SendEventStream: " << handle_result.status().message() << std::endl;
        return 1;
    }
    auto& handle = *handle_result;

    // Send multiple events via the stream.
    for (int i = 0; i < 10; i++) {
        auto ev_result = kubemq::Event::Builder()
                             .SetChannel(channel)
                             .SetBody("stream-msg-" + std::to_string(i))
                             .SetMetadata("stream-demo")
                             .Build();
        if (!ev_result.ok()) {
            std::cerr << "[ERROR] Build event: " << ev_result.status().message() << std::endl;
            return 1;
        }
        auto send_status = handle->Send(*ev_result);
        if (!send_status.ok()) {
            std::cerr << "[ERROR] handle.Send: " << send_status.message() << std::endl;
            return 1;
        }
        std::cout << "[4] Stream sent event " << (i + 1) << std::endl;
    }

    // Wait for at least one event to be received.
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "[6] Stream send demo complete" << std::endl;

    // Close the stream handle explicitly.
    // Note: EventStreamHandle destructor would also handle cleanup (RAII)
    handle->Close();
    sub->Cancel();

    auto close_status = client->Close();
    if (!close_status.ok()) {
        std::cerr << "[ERROR] Close failed: " << close_status.message() << std::endl;
        return 1;
    }
    std::cout << "[7] Client closed" << std::endl;

    return 0;
}

How It Works

Opens a bidirectional stream with SendEventStream() for efficient multi-event publishing.
Sends 10 events through the stream handle using handle->Send(event).
The stream avoids per-message gRPC call overhead, improving throughput.
Closes the stream handle explicitly with handle->Close().

Stream Send

Overview

Prerequisites

Code

How It Works

On this page

Stream Send

Overview

Prerequisites

Code

How It Works

Related

On this page