# Stream Receive (/sdks/rust/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 receive 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.

For manual settlement, `client.new_queue_downstream_receiver()` opens that persistent bidirectional stream; each call to `poll()` fetches a batch that nothing removes from the queue until you explicitly settle it with `ack()`, `nack()`, or `re_queue()` per message. Leaving a message unsettled returns it for redelivery once the visibility timeout expires.

**Gotchas:** a crash between receiving and acknowledging redelivers the batch, so processing must be idempotent; `receive_queue_messages` (the simpler, non-streaming call) auto-acknowledges on delivery, so switch to the downstream receiver whenever a failure mid-processing should put the message back; and `re_queue()` versus `nack()` differ in whether the message goes to the back of the same queue or through the broker's normal redelivery/dead-letter path — pick deliberately.

## Prerequisites [#prerequisites]

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

## Code [#code]

```rust title="main.rs"
use kubemq::prelude::*;
use kubemq::PollRequest;

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

    let channel = "rust-queues.stream-receive";

    let mut receiver = client.new_queue_downstream_receiver().await?;

    let response = receiver
        .poll(PollRequest {
            channel: channel.to_string(),
            max_items: 10,
            wait_timeout_seconds: 5,
            auto_ack: false,
        })
        .await?;

    println!("Polled {} messages", response.messages.len());

    for msg in &response.messages {
        let body = String::from_utf8_lossy(&msg.message.body);
        match process(&body) {
            Outcome::Done => {
                msg.ack().await?;
                println!("Acked: {}", body);
            }
            Outcome::Retry => {
                msg.nack().await?;
                println!("Nacked (redeliver): {}", body);
            }
            Outcome::WrongQueue => {
                msg.re_queue("rust-queues.stream-receive.dead").await?;
                println!("Re-queued: {}", body);
            }
        }
    }

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

enum Outcome {
    Done,
    Retry,
    WrongQueue,
}

fn process(_body: &str) -> Outcome {
    Outcome::Done
}
```

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

* `client.new_queue_downstream_receiver()` opens a persistent bidirectional stream; `poll()` fetches a batch without removing anything from the queue until you settle it.
* `PollRequest { auto_ack: false, .. }` disables auto-acknowledgement, so `response.messages` arrives as `QueueDownstreamMessage`s you settle individually.
* Each message exposes `ack()` (permanently remove it), `nack()` (return it for redelivery), and `re_queue()` (move it to a different channel) — the example routes to each based on how processing turns out.
* The simpler `receive_queue_messages` call auto-acknowledges on delivery and has no per-message settlement; reach for the downstream receiver whenever a failure mid-processing should put the message back.
* Review timeouts, channel names, and client IDs before running against shared environments.
* Run the program while the server from the prerequisites is available.

## Related [#related]

* [Queues overview](/learn/queues)
* [Rust SDK Reference](/sdks/rust/reference/queues)
* [Stream Send](/sdks/rust/how-to/queues/stream-send)