Struct HandlerOutcome 

pub struct HandlerOutcome { /* private fields */ }

The settlement of one dispatched message: the broker status the dispatcher acts on, plus an optional post-settle continuation.

A handler’s Err side carries one (Result<(), HandlerOutcome>, or per element Vec<HandlerOutcome> on a batch); Ok acks. Build one with the short constructors - ack, retry, retry_after, drop - and attach follow-up work with and_after.

Cancel safety

The continuation runs after the message is already settled, so it is at-most-once: a panic or a crash before it completes never redelivers the message. See and_after.

Examples

use ruststream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Payment {
    id: u64,
    settled: bool,
    valid: bool,
}

async fn release_hold(_id: u64) {}

#[subscriber("payments")]
async fn reconcile(payment: &Payment) -> HandlerOutcome {
    if !payment.valid {
        // Never coming right: off the queue for good.
        return HandlerOutcome::drop();
    }
    if !payment.settled {
        return HandlerOutcome::retry();
    }
    // Settled, and the hold is released once the ack is through.
    let id = payment.id;
    HandlerOutcome::ack().and_after(async move { release_hold(id).await })
}

A batch handler settles each element with its own outcome:

use ruststream::memory::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

/// One database round-trip per batch; entries that are not ready come back alone.
#[subscriber("orders")]
async fn reconcile(orders: &[Order]) -> Vec<HandlerOutcome> {
    orders
        .iter()
        .map(|order| {
            if order.id == 0 {
                HandlerOutcome::retry()
            } else {
                HandlerOutcome::ack()
            }
        })
        .collect()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(reconcile.batch(nonzero!(64)));
    })
}

Implementations

impl HandlerOutcome

pub const fn ack() -> HandlerOutcome

Acknowledge the message; the broker removes it from the queue.

Examples

use ruststream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

pub const fn retry() -> HandlerOutcome

Negatively acknowledge the message, asking the broker to redeliver it.

Under a registration’s max_attempts cap this obeys the cap as retry_after does. Where the transport counts its own redeliveries (redelivery_count) it stays the broker’s requeue and that count is read; where it counts none the runtime republishes the delivery at once so the framework’s count travels with it, and on that path an immediate retry is no longer a broker requeue.

pub const fn retry_after(delay: Duration) -> HandlerOutcome

Redeliver, but not before delay has passed - the not-ready-yet case (a dependency has not arrived, an upstream is rate-limited), where an immediate redelivery would just spin.

A broker with native delayed redelivery (JetStream NAK with delay) honours the delay itself, through IncomingMessage::nack_after; on one without, the runtime drops the delivery and re-publishes a copy of it to its own source after the delay, with the RETRY_COUNT_HEADER incremented. The publisher that copy leaves through is on every registration already, from the broker’s own DefaultPublish policy, and .out_retry(policy) replaces it (see Retry). That copy is at-most-once over the delay window: it rides a detached task, so a process that exits before the timer fires loses it.

How many times a message comes back, and where it goes when the attempts run out, is the registration’s own declaration: see max_attempts.

Examples

A payment not settled yet comes back after five seconds, at most five times, then goes to a dead-letter destination:

use std::time::Duration;

use ruststream::memory::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Payment {
    settled: bool,
}

#[subscriber("payments")]
async fn reconcile(payment: &Payment) -> HandlerOutcome {
    if !payment.settled {
        return HandlerOutcome::retry_after(Duration::from_secs(5));
    }
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(reconcile)
            .max_attempts(nonzero!(5u32))
            .dead_letter("payments.dead");
    })
}

pub const fn drop() -> HandlerOutcome

Negatively acknowledge the message without asking for redelivery.

pub fn and_after<F>(self, fut: F) -> HandlerOutcome

where F: Future<Output = ()> + Send + 'static,

Attaches a post-settle continuation to this outcome.

The dispatcher first settles the message by this outcome (ack / nack), then runs fut on a tracked task that graceful shutdown drains. Use it for a non-critical side effect that must not gate the settlement decision or affect redelivery: a notification, slow follow-up work, a cache warm-up. The continuation runs after any settle, so drop().and_after(..) is valid. Attaching a second continuation replaces the first.

Cancel safety

At-most-once: the message is already settled when fut runs, so a continuation that panics or is lost on a crash never triggers redelivery. Do not put work whose loss must redeliver the message in here; settle by outcome and let the broker retry instead.

Examples

use ruststream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

async fn notify_downstream(_id: u64) {}

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    let id = order.id;
    // Runs once the ack is through; its loss never brings the order back.
    HandlerOutcome::ack().and_after(async move { notify_downstream(id).await })
}

pub const fn is_ack(&self) -> bool

Whether this outcome acknowledges the message.

pub const fn is_retry(&self) -> bool

Whether this outcome asks for redelivery (an immediate or a delayed retry).

pub const fn is_drop(&self) -> bool

Whether this outcome drops the message (a negative acknowledgement without redelivery).

pub const fn retry_delay(&self) -> Option<Duration>

The redelivery delay of a retry_after outcome; None for every other status.

Trait Implementations

impl Debug for HandlerOutcome

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl From<!> for HandlerOutcome

fn from(never: !) -> HandlerOutcome

Converts to this type from the input type.

impl From<HandlerResult> for HandlerOutcome

fn from(outcome: HandlerResult) -> HandlerOutcome

Converts to this type from the input type.