Function shutdown_flushes 

pub async fn shutdown_flushes<B, MkBroker, Src, MkSrc, Pub, MkPub>(
    make_broker: MkBroker,
    make_source: MkSrc,
    make_publisher: MkPub,
    backlog: Backlog,
)
where
    B: Broker,
    MkBroker: Fn() -> B,
    Src: SubscriptionSource<Connected<B>>,
    MkSrc: Fn(&str) -> Src,
    Pub: Publisher,
    MkPub: Fn(&Connected<B>) -> Pub,

Verifies a shutdown finishes what was handed to the broker before it: an acknowledgement and a publish made right before shutdown, with no wait in between.

A client that queues its acknowledgements and publishes for a background task loses both when the shutdown stops that task first: the acknowledged message comes back to the next consumer, and the published one never reaches anyone. Both look fine to the process that shut down.

The check acknowledges one delivery, holds a second one, publishes a third message and shuts down at once; shutdown itself must return within a bound. Then it looks from another connection, and how depends on backlog, the broker’s answer to whether a message waits for a subscription that opens later (TestableBroker::backlog):

• Backlog::Delivered (a queue, a log): a new connection subscribes after the shutdown. The published message must arrive within ten seconds, and for all ten seconds the acknowledged one must not come back, nor the held one where acknowledging it after the shutdown answered Ok.
• Backlog::Missed (publish/subscribe): a subscription on another connection is open before the publish, and the published message must reach it.

make_broker must reach the same broker every time it is called: a server, or in process one world shared by every broker it builds (clones of one MemoryBroker share their bus). An in-process transport that gives every broker a world of its own runs this check live only. Where the transport leases a delivery to the consumer that took it (an SQS visibility timeout, a JetStream ack wait), give the descriptor a lease of a few seconds: a message the first connection prefetched and never settled reaches the second one only once its lease runs out, and the check waits ten seconds.

Examples

use ruststream::conformance::lifecycle;
use ruststream::memory::{MemoryBroker, MemorySource};
use ruststream::testing::Backlog;

let bus = MemoryBroker::new();
lifecycle::shutdown_flushes(
    move || bus.clone(),
    |name| MemorySource::new(name),
    |connected| connected.publisher(),
    Backlog::Missed,
)
.await;

Panics

Panics with a descriptive message when the shutdown hangs or fails, when the published message is lost, or when an acknowledged message comes back.