Acknowledgement

ack and nack report Unsupported on both brokers: nothing records how far a service read, so it resumes from a stored position.

A handler settles a delivery as on any broker, with HandlerOutcome::ack or another outcome. The runtime passes the settlement to the transport, which answers AckError::Unsupported, and the runtime logs a warning naming the subscription. A stream file keeps no record of a reader, and the file client's resumable mode is not implemented. So a subscription opens where its descriptor says on every start, whatever the previous run acknowledged. A service that resumes stores the sequence it reached, read with Ctx<Position>, and opens the next run there with start_at. A pipe has no log to go back to: a line read is gone, whatever the handler answers. What a handler asks for beyond ack still happens, because the runtime does it: retry_after on a stream file, out_retry on a pipe, a cap and a dead-letter key on both.

Example

The service invoices every order the file holds; acknowledged orders are read again on the next start.

Macros
Manual

examples/sea_acknowledgement.rs
/// Invoices every order the file holds, on every start.
#[subscriber(FileStream::new("orders"), start_at(FilePosition::beginning()))]
async fn invoice(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "invoiced");
    // Reported as unsupported: the next start reads this order again.
    HandlerOutcome::ack()
}

/// The invoicing service on the file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("invoicing", "0.1.0")).with_broker(file, |b| {
        b.include(invoice);
    })
}

examples/manual/sea_acknowledgement.rs
/// Invoices every order the file holds, on every start.
struct Invoice;

impl Handle<Order> for Invoice {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "invoiced");
        // Reported as unsupported: the next start reads this order again.
        ready(Ok(()))
    }
}

/// The invoicing service on the file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("invoicing", "0.1.0")).with_broker(file, |b| {
        b.include(
            subscriber(FileStream::new("orders"), Invoice)
                .start_at(FilePosition::beginning())
                .build(),
        );
    })
}

Example​

Example