Shutdown

Shutdown closes the producers the crate opened, and a publisher or a seek used afterwards reports PulsarError::NotConnected.

The crate opens one producer per topic the service publishes to, on the first publish there. At shutdown it closes each of them, so the server releases them at once. A producer that fails to close is logged and does not fail the shutdown. The client's connection itself ends with the client. The shutdown therefore returns (), with no report of its own.

A publisher taken before the shutdown keeps pointing at the closed connection. That includes one taken off the broker before startup with PulsarBroker::publisher. A publish through it returns PulsarError::NotConnected and never succeeds against the dead connection. A seek through a handle that outlived the connection is refused the same way.

Example

The service replies to each order with a shipment, which opens a producer on shipments; shutdown closes it.

Macros
Manual

examples/pulsar_broker_shutdown.rs
/// The service on `broker`. Its replies open a producer on `shipments`, which shutdown closes.
fn service(broker: PulsarBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(ship).out_reply(Publish);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(PulsarBroker::new("pulsar://localhost:6650"))
}

examples/manual/pulsar_broker_shutdown.rs
/// The service on `broker`. Its replies open a producer on `shipments`, which shutdown closes.
fn service(broker: PulsarBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(PulsarSubscription::new("orders", "shipping"), Ship)
                .reply()
                .build(),
        )
        .out_reply(Publish);
    })
}

fn app() -> impl App<State = ()> {
    service(PulsarBroker::new("pulsar://localhost:6650"))
}

Example​

Example