# Stream Receive (/sdks/cpp/how-to/queues/stream-receive)



## Overview [#overview]

A **downstream receiver** is the persistent-connection way to pull queue messages: instead of opening and tearing down a request for every batch, you open one gRPC stream and reuse it across many poll cycles. That matters for any consumer that runs continuously — a worker loop, a background processor — where reconnecting per batch would add latency and churn on both the client and the broker.

The receiver is created once with `NewQueueDownstreamReceiver()`, then each call to `Poll()` fetches a batch with `auto_ack = false` so nothing is removed from the queue until you explicitly settle it. Every message carries a `transaction_id`; acknowledging it — individually, or as a batch with `AckAll()` — permanently removes it, while leaving it unacknowledged returns it for redelivery once the visibility timeout expires.

**Gotchas:** an unclosed receiver holds server-side state — always call `Close()` on both the receiver and the client when done; a crash between receiving and acknowledging redelivers the batch, so processing must be idempotent; and forgetting to set `auto_ack = false` silently drops the manual-settlement guarantee this pattern exists for.

## 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_receive
//
// Demonstrates receiving queue messages using NewQueueDownstreamReceiver + Poll.
// The receiver manages a persistent downstream stream with automatic reconnection.
//
// Channel: cpp-queues.stream-receive
// Client ID: cpp-queues-stream-receive-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).

#include <kubemq/kubemq.h>

#include <chrono>
#include <iostream>
#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-queues-stream-receive-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-receive";

    // Send a message to the queue so we have something to receive.
    std::cout << "[2] Sending message to queue " << channel << std::endl;
    auto msg_result =
        kubemq::QueueMessage::Builder().SetChannel(channel).SetBody("message to receive").Build();
    if (!msg_result.ok()) {
        std::cerr << "[ERROR] Build message: " << msg_result.status().message() << std::endl;
        return 1;
    }
    auto send_result = client->SendQueueMessage(*msg_result);
    if (!send_result.ok()) {
        std::cerr << "[ERROR] SendQueueMessage: " << send_result.status().message() << std::endl;
        return 1;
    }
    std::cout << "[3] Message sent" << std::endl;

    // Allow message to be committed to the queue.
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // Open a downstream receiver.
    std::cout << "[4] Opening downstream receiver" << std::endl;
    auto receiver_result = client->NewQueueDownstreamReceiver();
    if (!receiver_result.ok()) {
        std::cerr << "[ERROR] NewQueueDownstreamReceiver: " << receiver_result.status().message()
                  << std::endl;
        return 1;
    }
    auto& receiver = *receiver_result;

    // Poll for messages (manual ack).
    kubemq::PollRequest poll_req;
    poll_req.channel = channel;
    poll_req.max_items = 10;
    poll_req.wait_timeout_seconds = 5;
    poll_req.auto_ack = false;

    auto poll_result = receiver->Poll(poll_req);
    if (!poll_result.ok()) {
        std::cerr << "[ERROR] Poll: " << poll_result.status().message() << std::endl;
        return 1;
    }
    if (poll_result->is_error()) {
        std::cerr << "[ERROR] Poll response error: " << poll_result->error() << std::endl;
        return 1;
    }

    std::cout << "[5] Received " << poll_result->messages().size() << " messages" << std::endl;
    for (const auto& dm : poll_result->messages()) {
        std::cout << "  body=" << dm.message().body() << " tx=" << dm.transaction_id() << std::endl;
    }

    // Ack all received messages.
    if (!poll_result->messages().empty()) {
        auto ack_status = poll_result->AckAll();
        if (!ack_status.ok()) {
            std::cerr << "[ERROR] AckAll: " << ack_status.message() << std::endl;
            return 1;
        }
        std::cout << "[6] Messages acknowledged" << std::endl;
    }

    // Close the receiver and client explicitly.
    // Note: Destructors also handle cleanup, but explicit calls are shown
    // for clarity and to match Go's defer pattern.
    receiver->Close();
    std::cout << "[7] Downstream receiver 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 << "[8] Client closed" << std::endl;

    return 0;
}
```

## How It Works [#how-it-works]

* Creates a downstream receiver with `NewQueueDownstreamReceiver()`.
* Polls for messages using `receiver->Poll(poll_req)` with configurable timeout and max items.
* Messages can be individually settled with `Ack()`, `Nack()`, or `ReQueue()`.
* The receiver maintains a persistent connection for efficient polling.

## Related [#related]

* [Pattern overview](/learn/queues/getting-started)
* [C++ SDK Reference](/sdks/cpp/reference/queues)
* [Stream Send](/sdks/cpp/how-to/queues/stream-send)
* [Auto Ack](/sdks/cpp/how-to/queues/auto-ack)