Shutdown

A publisher or requester that outlives the connection reports AmqpError::Closed instead of succeeding against a closed connection.

At shutdown the runtime stops the subscriptions and lets the handlers in flight finish. Then it closes the connection with the AMQP close handshake. A delivery the service received and never settled goes back to its queue when its channel closes, so another consumer gets it.

A publisher handed out before the shutdown can outlive it: a task the service spawned keeps it, or a clone of a requester does. Every use of such a handle after the shutdown returns AmqpError::Closed, which names the routing key, the queue or the exchange the call targeted. The connection is marked closed before the handshake starts, so no message slips into a connection that is going away. A tool that drives the broker by hand receives a ClosedLapinBroker from the shutdown; its handshake tells an orderly close from a connection the server had already dropped.

Example

A heartbeat task takes its publisher at startup and runs past the service's shutdown; its first publish after the shutdown fails with AmqpError::Closed, and the task ends.

Macros
Manual

examples/lapin_broker_shutdown.rs
/// Announces the service every 30 seconds until a publish fails; after shutdown that failure is
/// `AmqpError::Closed`.
async fn heartbeat(publisher: LapinPublisher) {
    let mut ticks = tokio::time::interval(Duration::from_secs(30));
    loop {
        ticks.tick().await;
        let beat = Heartbeat {
            service: "orders".to_owned(),
        };
        if let Err(err) = publisher.message(&beat).publish().await {
            tracing::info!(error = %err, "heartbeat stopped");
            return;
        }
    }
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(place);
            // The task keeps its publisher past the service's shutdown.
            b.after_startup(Publish::default(), async move |publisher| {
                tokio::spawn(heartbeat(publisher));
                Ok::<_, Infallible>(())
            });
        },
    )
}

examples/manual/lapin_broker_shutdown.rs
/// Announces the service every 30 seconds until a publish fails; after shutdown that failure is
/// `AmqpError::Closed`.
async fn heartbeat(publisher: LapinPublisher) {
    let mut ticks = tokio::time::interval(Duration::from_secs(30));
    loop {
        ticks.tick().await;
        let beat = Heartbeat {
            service: "orders".to_owned(),
        };
        if let Err(err) = publisher.message(&beat).publish().await {
            tracing::info!(error = %err, "heartbeat stopped");
            return;
        }
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber("orders", Place).build());
            // The task keeps its publisher past the service's shutdown.
            b.after_startup(Publish::default(), async move |publisher| {
                tokio::spawn(heartbeat(publisher));
                Ok::<_, Infallible>(())
            });
        },
    )
}

Example​

Example