Startup

At startup the broker builds its clients, then each subscription creates or updates its topology and opens a streaming pull.

The runtime connects the broker once, after the synchronous builder has run. The connect builds four clients of the Google client library: a subscriber, a publisher, and the topic and subscription admin clients. The clients open their connections on first use.

Then each subscription opens in turn. A descriptor the service cannot express fails the startup before any call: an empty name, or a dead-letter declaration with one half missing or a cap outside 5 to 100. A descriptor with create_with_topic creates its topic, its dead-letter topic and the subscription where they are missing. A subscription that already exists and carries a dead-letter declaration gets the declared policy written onto it. Each of these calls needs the matching Pub/Sub permission, and a refusal fails the startup with PubSubError::Admin. Last, the subscription opens its streaming pull. A subscription that does not exist fails there, on its stream, with PubSubError::Receive.

Example

The subscription orders-workers creates itself on the topic orders with a dead-letter policy of five attempts. Its test also starts the same service with a cap of three, which the startup refuses.

Macros
Manual

examples/pubsub_broker_startup.rs
/// The subscription `orders-workers` on the topic `orders`, both created when they are missing.
#[subscriber(GooglePubSub::new("orders-workers").create_with_topic("orders"))]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // Before the pull opens: the topics `orders` and `orders-dead`, then the
            // subscription with its dead-letter policy.
            b.include(place)
                .max_attempts(nonzero!(5u32))
                .dead_letter("orders-dead");
        },
    )
}

examples/manual/pubsub_broker_startup.rs
struct Place;

impl Handle<Order> for Place {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order placed");
        ready(Ok(()))
    }
}

/// The subscription `orders-workers` on the topic `orders`, both created when they are missing.
fn orders() -> GooglePubSub {
    GooglePubSub::new("orders-workers").create_with_topic("orders")
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // Before the pull opens: the topics `orders` and `orders-dead`, then the
            // subscription with its dead-letter policy.
            b.include(subscriber(orders(), Place).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("orders-dead");
        },
    )
}

Example​

Example