ruststream-gcp-pubsub 0.7.0

ruststream-gcp-pubsub connects a RustStream service to Google Cloud Pub/Sub over the official google-cloud-pubsub client. Handlers, routing, codecs and middleware come from the framework; this crate is the transport.

Features

• Streaming pull with ack deadlines extended in the background while a handler runs.
• Native acknowledgement, with the confirmed forms on exactly-once subscriptions.
• Retry caps as the subscription's own dead-letter policy, and delayed retries held under the delivery's lease.
• Ordering keys as a per-message setting.
• Plain Pub/Sub messages: headers ride attributes, with no envelope.
• Batches assembled on the client.
• The emulator as a target, with topics and subscriptions created on subscribe for local development.
• AsyncAPI under the googlepubsub protocol, behind the asyncapi feature.
• Tests on the production app: TestApp runs it with PubSubBroker connected in process.

Install

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json"] }
ruststream-gcp-pubsub = "0.7"
serde = { version = "1", features = ["derive"] }

[dev-dependencies]
ruststream-gcp-pubsub = { version = "0.7", features = ["testing"] }

Write a service

use ruststream_gcp_pubsub::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Outgoing, Serialize)]
struct Order {
    id: u64,
}

#[derive(Debug, Deserialize, PartialEq, Serialize, Outgoing)]
struct Confirmation {
    order_id: u64,
}

#[subscriber(GooglePubSub::new("orders-workers").create_with_topic("orders"))]
async fn handle(order: &Order, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    if out
        .message(&Confirmation { order_id: order.id })
        .to("confirmations")
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            b.include(handle).out(DefaultSlot, Publish::default()).build();
        },
    )
}

#[ruststream::app] generates main, so the binary understands run and asyncapi gen. A plain name subscribes to a subscription that already exists; GooglePubSub describes one with options.

Test it

TestApp runs the app main runs with PubSubBroker connected in process, with no server.

use ruststream::testing::TestApp;

let tb = TestApp::start(app()).await?;

tb.broker::<PubSubBroker>()
    .message(&Order { id: 42 })
    .to("orders")
    .publish()
    .await?;

tb.out::<DefaultSlot>()
    .assert_called_once()
    .decoded_as::<Confirmation>()
    .with(&Confirmation { order_id: 42 });

Documentation

• This crate: https://docs.rs/ruststream-gcp-pubsub
• The framework: https://powersemmi.github.io/ruststream/latest

Minimum supported Rust version

The MSRV is 1.95, edition 2024.

Contributing

See CONTRIBUTING.md.

License

Licensed under the Apache-2.0 license.