# Ack & Reject (/sdks/rust/how-to/queues/ack-reject)



## Overview [#overview]

Ack and reject give you per-message control over queue delivery instead of an all-or-nothing batch outcome. Unlike the simple `receive_queue_messages` call, which auto-acknowledges as soon as messages are delivered, the queue stream downstream API lets a consumer hold a batch in an open transaction on the broker — invisible to other consumers — until it explicitly settles each message. That's what you need when one bad record in a batch shouldn't take the rest down with it.

Settlement happens through two calls on the received message: an ack, which permanently removes it from the queue, and a nack (reject), which returns it to the queue for redelivery. Internally the broker tracks this against a receive count, which a dead-letter policy can use to stop retrying a poison message forever.

**Gotchas:** an unsettled message isn't gone — it snaps back to the queue once the visibility timeout expires, so a slow consumer looks identical to a rejecting one; settle every message before that deadline, and don't confuse `is_peek`-based auto-ack with manual settlement — only the downstream stream API gives you a real per-message accept/reject decision.

## 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, QueueMessageBuilder};

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

    let channel = "rust-queues.ack-reject";

    for label in ["accept", "reject"] {
        let msg = QueueMessageBuilder::new()
            .channel(channel)
            .body(label.as_bytes().to_vec())
            .build();
        client.send_queue_message(msg).await?;
    }

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

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

    for msg in &response.messages {
        let body = String::from_utf8_lossy(&msg.message.body);
        if body.starts_with("accept") {
            msg.ack().await?;
            println!("Acknowledged: id={}", msg.message.id);
        } else {
            msg.nack().await?;
            println!("Rejected: id={}", msg.message.id);
        }
    }

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

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

* `client.new_queue_downstream_receiver()` opens the queue stream downstream API and holds the polled batch in an open transaction until you settle it.
* `PollRequest { auto_ack: false, .. }` disables auto-acknowledgement, so each `QueueDownstreamMessage` carries its own `ack()` and `nack()`.
* `ack()` permanently removes the message; `nack()` returns it to the queue for redelivery — the example picks one or the other per message based on its content.
* By contrast, the simple `receive_queue_messages` call auto-acknowledges as soon as messages are delivered and gives you no per-message accept/reject decision.
* 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)
* [Auto Ack](/sdks/rust/how-to/queues/auto-ack)