Stream Receive

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

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

Code

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

• 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 QueueDownstreamMessages 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

• Queues overview
• Rust SDK Reference
• Stream Send