# Stream Send (/sdks/cpp/how-to/queues/stream-send)



## Overview [#overview]

Sending one queue message per call works fine for occasional traffic, but each call carries its own round trip. At high volume — event ingestion, sensor telemetry, log shipping — that per-call overhead caps your throughput well below what the connection can support.

`client->QueueUpstream(on_result, on_error)` opens a persistent, bidirectional gRPC stream you reuse to push any number of batches. `upstream->Send(request_id, messages)` writes a batch and returns immediately; results — including per-message IDs and error status — arrive asynchronously through `on_result`, correlated by request ID. Because the stream stays open, you can queue the next batch before the previous callback fires.

**Gotchas:** results are delivered on the callback thread, not the thread that called `Send` — shared state your callback touches (like the `batches_confirmed` counter) needs its own synchronization. A stream error surfaces through `on_error` and ends the whole stream, not just one batch, so production code needs to detect that and reopen `QueueUpstream`. Always call `upstream->Close()` before closing the client — closing the client under an open stream can hide errors from in-flight batches.

## Prerequisites [#prerequisites]

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

## Code [#code]

```cpp title="main.cc"
// Example: queues_stream/stream_send
//
// Demonstrates high-throughput queue message publishing using QueueUpstream.
// The bidirectional stream allows sending multiple messages efficiently
// with per-batch result confirmations via callback.
//
// Channel: cpp-queues.stream-send
// Client ID: cpp-queues-stream-send-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).

#include <kubemq/kubemq.h>

#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>
#include <vector>

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

    kubemq::ClientOptions options;
    options.set_address("localhost", 50000);
    options.set_client_id("cpp-queues-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-queues.stream-send";
    std::atomic<int> batches_confirmed{0};

    // Open a bidirectional upstream stream for publishing with callbacks.
    std::cout << "[2] Opening upstream stream" << std::endl;
    auto upstream_result = client->QueueUpstream(
        [&batches_confirmed](const kubemq::QueueUpstreamResult& res) {
            if (res.is_error) {
                std::cerr << "[ERROR] Upstream result error: " << res.error << std::endl;
            } else {
                std::cout << "[4] Batch confirmed: ref_request_id=" << res.ref_request_id
                          << " messages=" << res.results.size() << std::endl;
            }
            batches_confirmed.fetch_add(1);
        },
        [](const kubemq::Status& err) {
            std::cerr << "[ERROR] Upstream stream error: " << err.message() << std::endl;
        });
    if (!upstream_result.ok()) {
        std::cerr << "[ERROR] QueueUpstream: " << upstream_result.status().message() << std::endl;
        return 1;
    }
    auto& upstream = *upstream_result;

    // Send 5 batches of 3 messages each.
    std::cout << "[3] Sending 5 batches of 3 messages each" << std::endl;
    for (int batch = 1; batch <= 5; batch++) {
        std::vector<kubemq::QueueMessage> messages;
        for (int i = 1; i <= 3; i++) {
            auto msg_result =
                kubemq::QueueMessage::Builder()
                    .SetChannel(channel)
                    .SetBody("batch-" + std::to_string(batch) + "-msg-" + std::to_string(i))
                    .Build();
            if (!msg_result.ok()) {
                std::cerr << "[ERROR] Build message batch=" << batch << " msg=" << i << ": "
                          << msg_result.status().message() << std::endl;
                return 1;
            }
            messages.push_back(std::move(*msg_result));
        }

        auto send_status = upstream->Send("req-batch-" + std::to_string(batch), messages);
        if (!send_status.ok()) {
            std::cerr << "[ERROR] Upstream Send batch " << batch << ": " << send_status.message()
                      << std::endl;
            return 1;
        }
    }

    // Wait for all batch result callbacks.
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "[5] Batches confirmed: " << batches_confirmed.load() << " of 5" << std::endl;

    // Close the upstream stream and client explicitly.
    // Note: Destructors also handle cleanup, but explicit calls are shown
    // for clarity and to match Go's defer pattern.
    upstream->Close();
    std::cout << "[6] Upstream stream closed" << std::endl;

    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 [#how-it-works]

* Opens an upstream stream with `QueueUpstream()` for efficient batch sending.
* Sends messages using `upstream->Send(request_id, messages)` with a batch per call.
* Results arrive via the `on_result` callback with message IDs and error status.
* Stream sending avoids per-message gRPC call overhead.

## Related [#related]

* [Pattern overview](/learn/queues/getting-started)
* [C++ SDK Reference](/sdks/cpp/reference/queues)
* [Stream Receive](/sdks/cpp/how-to/queues/stream-receive)
* [Batch Send](/sdks/cpp/how-to/queues/batch-send)