PubSubBroker

PubSubBroker connects a service to Google Cloud Pub/Sub; new only records the project, and the runtime connects it at startup.

PubSubBroker::new takes the project id that short topic and subscription names resolve in. It does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. The settings that follow it are builder steps too: a local emulator, explicit credentials and another service endpoint. Without them the broker reaches pubsub.googleapis.com with Application Default Credentials.

At startup the runtime connects the broker and receives a ConnectedPubSubBroker. Every subscription and every publisher of the service opens against that connected form. At shutdown the runtime flushes what the service published and closes the connection.

#[subscriber("orders-workers")] reads the existing subscription orders-workers of the project. A name is a short id or a full resource name, projects/{project}/subscriptions/{id}. Every other setting of a subscription belongs to the GooglePubSub descriptor.

Example

The service reads the subscription orders-workers of the project my-project. cargo run --example pubsub_broker --features testing -- run starts it against Pub/Sub.

Macros
Manual

examples/pubsub_broker.rs
/// Reads the subscription `orders-workers`.
#[subscriber("orders-workers")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

/// Application Default Credentials, against `pubsub.googleapis.com`.
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            b.include(place);
        },
    )
}

examples/manual/pubsub_broker.rs
/// Reads the subscription `orders-workers`.
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(()))
    }
}

/// Application Default Credentials, against `pubsub.googleapis.com`.
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            b.include(subscriber("orders-workers", Place).build());
        },
    )
}

Example​

Example