Shutdown

Shutdown flushes every message the service still buffers; a publisher used afterwards reports PubSubError::NotConnected.

The client library sends published messages in batches, so a publish may wait in a buffer for a few milliseconds. At shutdown the runtime stops the subscriptions and closes the connection, then flushes the buffer of every topic the service published to. A message published before the shutdown reaches the topic before the service exits. A subscription that stops ends its streaming pull; the deliveries the service has not settled come back to the subscription once their leases run out.

A publisher taken before the shutdown keeps pointing at the closed connection. A publish through it returns PubSubError::NotConnected and never succeeds against the dead connection.

Example

The handler replies with a shipment, and the shipment reaches its topic before the service exits. Its test publishes through a publisher taken before the start and gets NotConnected after the shutdown.

Macros

examples/pubsub_broker_shutdown.rs

/// Answers every order with a shipment on the topic `shipments`.
#[subscriber("orders-workers", reply)]
async fn ship(order: &Order) -> Shipment {
    Shipment { order_id: order.id }
}

fn service(broker: PubSubBroker) -> impl App<State = ()> {
    // At shutdown the runtime flushes the shipments still buffered for `shipments`.
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(ship).out_reply(Publish::default());
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(PubSubBroker::new("my-project"))
}

Manual

examples/manual/pubsub_broker_shutdown.rs

/// Answers every order with a shipment on the topic `shipments`.
struct Ship;

impl Handle<Order, Shipment> for Ship {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Shipment, HandlerOutcome>> {
        ready(Ok(Shipment { order_id: order.id }))
    }
}

fn service(broker: PubSubBroker) -> impl App<State = ()> {
    // At shutdown the runtime flushes the shipments still buffered for `shipments`.
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders-workers", Ship).reply().build())
            .out_reply(Publish::default());
    })
}

fn app() -> impl App<State = ()> {
    service(PubSubBroker::new("my-project"))
}