ruststream-gcp-pubsub 0.6.0

ruststream-gcp-pubsub implements the RustStream broker contract over the official google-cloud-pubsub client. Handlers, routers, codecs, and middleware come from the framework; this crate supplies the transport - and nothing broker-specific leaks back into the framework.

Features

• Lazy startup contract. PubSubBroker::new(project) is synchronous and does no I/O (Application Default Credentials by default; explicit credentials, a regional endpoint, or a local emulator as builder options); the runtime connects once at startup, so the broker composes with #[ruststream::app].
• Streaming pull as the message stream. Each subscription is a Stream of deliveries; the client extends ack deadlines in the background while a handler runs, so a slow handler does not cause redelivery.
• Native acknowledgement. ack and nack(requeue = true) map onto the product directly (with the confirmed forms on exactly-once subscriptions). nack(requeue = false) acknowledges: Pub/Sub has no drop-without-redelivery verb - poison routing belongs to the subscription's dead-letter policy, and the delivery-attempt count is surfaced as a header.
• Ordering keys as the partition key. A partition-key header becomes the message's ordering key on publish and comes back as the same header (feeding Partitioned) on delivery.
• Attributes carry headers directly - no envelope format is invented; non-Rust peers see plain Pub/Sub messages.
• Emulator as a first-class target. PubSubBroker::new(p).emulator("localhost:8085") wires the plaintext endpoint and anonymous credentials (the client does not honour PUBSUB_EMULATOR_HOST on its own), and PubSubSubscription::create_with_topic creates the resources on subscribe for local development.
• In-process test broker (feature testing). PubSubTestBroker reproduces core routing with no server, implements ruststream::testing::TestableBroker, and passes the framework's conformance suite in process.

Status

Implemented and verified against the Pub/Sub emulator (the framework's conformance lifecycle suite and the integration tests run in CI against it). Built on the ruststream 0.6 line, which is on crates.io; this crate itself is not published yet. Design and scope are tracked in powersemmi/ruststream#188.

MSRV is 1.88, tracking the official client (the core stays at 1.85; a dependent may exceed its dependency's floor).

Write a service

use std::time::Duration;

use ruststream::runtime::{App, AppInfo, HandlerResult, RustStream};
use ruststream::subscriber;
use ruststream_gcp_pubsub::{PubSubBroker, PubSubSubscription};
use serde::Deserialize;

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

#[subscriber(PubSubSubscription::new("orders-workers").max_outstanding(1_000))]
async fn handle(order: &Order) -> HandlerResult {
    println!("got order {}", order.id);
    HandlerResult::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))
}

The descriptor names an existing subscription; create_with_topic("orders") opts into creating the subscription (and topic) on subscribe, which is what the emulator workflow wants.

Test it

The testing feature runs handlers against an in-process Pub/Sub stand-in - no server, same routing. Product behaviour (deadline extension, redelivery, ordered delivery) is covered by the env-gated live suite instead: just test-brokers starts the emulator and runs the integration tests plus the framework conformance lifecycle against it.

Layout

ruststream-gcp-pubsub/
├── crates/
│   └── ruststream-gcp-pubsub/  the published crate
│       └── examples/           runnable pubsub_* examples
├── docker-compose.test.yml     the Pub/Sub emulator for the live suite
└── Cargo.toml                  workspace

Contributing

just check          # fmt, clippy, feature checks
just test           # handler-stub tests, no server
just test-brokers   # live integration + conformance against the emulator

License

Licensed under the Apache-2.0 license.