Ack & Reject

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

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

Code

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

• 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

• Queues overview
• Rust SDK Reference
• Auto Ack