Struct GooglePubSub 

pub struct GooglePubSub { /* private fields */ }

A subscription descriptor for one Pub/Sub subscription.

Implements SubscriptionSource for the connected broker, so it sits inline in the #[subscriber(..)] decorator, and the declaration a service ships is the one its tests run:

use std::time::Duration;

use ruststream_gcp_pubsub::prelude::*;
use serde::Deserialize;

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

#[subscriber(
    GooglePubSub::new("orders-workers")
        .max_outstanding(1_000)
        .ack_extension(Duration::from_secs(60))
)]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    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);
        })
}

Implementations

impl GooglePubSub

pub fn new(name: impl Into<String>) -> Self

Names an existing subscription (short name or full projects/{p}/subscriptions/{s} resource name).

The name is taken on trust: nothing is created, and the subscription is infrastructure this crate does not own. Where it answers to nothing, the streaming pull is what says so, and the stream yields PubSubError::Receive naming the subscription instead of delivering.

pub fn create_with_topic(self, topic: impl Into<String>) -> Self

Creates the subscription bound to topic on subscribe when it does not exist yet (the topic is created too). Meant for local development and tests against the emulator; production subscriptions are usually managed as infrastructure.

The subscription it creates enables message ordering, so a run of messages sharing an ordering key reaches the handler in publish order. A subscription that already exists keeps whatever its own configuration says: ordered delivery is a field of the subscription resource, and the crate does not change it on a subscription it did not create.

pub fn max_outstanding(self, messages: i64) -> Self

Flow control: how many received messages may be outstanding (unacked) at once. Defaults to the client’s 1000.

The limit travels with the streaming pull and the service is what holds to it, pausing the stream while this many deliveries are unsettled and resuming below it.

pub fn ack_extension(self, extension: Duration) -> Self

How far each background ack-deadline extension reaches while a handler runs. The client clamps it to the protocol’s 10s..=600s range; defaults to 60s.

pub fn max_lease(self, lease: Duration) -> Self

How long one delivery may stay unsettled before the client stops extending its ack deadline. Defaults to the client’s own hour.

This is the budget a delayed retry spends: HandlerOutcome::retry_after(delay) holds the delivery in the process for delay, and a delay longer than this is refused at the call, because the subscription would redeliver before it elapsed.

use std::time::Duration;

use ruststream_gcp_pubsub::prelude::*;
use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct Invoice {
    id: u64,
    ready: bool,
}

/// An invoice that is not ready yet comes back in ninety minutes, inside the lease.
#[subscriber(
    GooglePubSub::new("invoices-workers").max_lease(Duration::from_hours(2))
)]
async fn send(invoice: &Invoice) -> HandlerOutcome {
    if !invoice.ready {
        return HandlerOutcome::retry_after(Duration::from_mins(90));
    }
    tracing::info!(invoice.id, "sent");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("invoices", "0.1.0"))
        .with_broker(PubSubBroker::new("my-project"), |b| {
            b.include(send);
        })
}

pub fn batch_wait(self, wait: Duration) -> Self

How long a partial batch waits for more deliveries after its first one, for a handler that takes a slice. Defaults to 50ms.

How large a batch may be is not named here: that is the registration’s batch(n), which reaches the subscription on its own. This is the other half - how long the subscription is willing to wait for a batch that size, before handing over what it has.

use std::time::Duration;

use ruststream_gcp_pubsub::prelude::*;
use serde::Deserialize;

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

#[subscriber(GooglePubSub::new("orders-workers").batch_wait(Duration::from_millis(200)))]
async fn index(orders: &[Order]) -> HandlerOutcome {
    let ids: Vec<u64> = orders.iter().map(|order| order.id).collect();
    tracing::info!(?ids, "indexed");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(PubSubBroker::new("my-project"), |b| {
            // Up to 100 orders per call, or whatever arrived within 200ms of the first.
            b.include(index.batch(nonzero!(100)));
        })
}

pub fn subscription(&self) -> &str

The subscription name this descriptor resolves.

Trait Implementations

impl Clone for GooglePubSub

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for GooglePubSub

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Eq for GooglePubSub

impl FromName for GooglePubSub

A subscription is identified by its name and nothing else, every other setting having a default, so the mount site may supply the name instead of the declaration: #[subscriber(GooglePubSub)] on the handler and .name("orders-workers") where it is mounted.

Examples

use ruststream_gcp_pubsub::prelude::*;
use serde::Deserialize;

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

#[subscriber(GooglePubSub)]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(PubSubBroker::new("my-project"), |b| {
            // The deployment names the subscription; every other setting is the default.
            b.include(handle.name("orders-workers"));
        })
}

fn from_name(name: impl Into<Cow<'static, str>>) -> Self

Builds the source bound to name.

impl IntoSource for GooglePubSub

type Source = GooglePubSub

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl PartialEq for GooglePubSub

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for GooglePubSub

impl SubscriptionSource<ConnectedPubSubBroker> for GooglePubSub

A Pub/Sub subscription moves a spent delivery itself, so the copy path is BrokerMoves: the subscription’s dead-letter policy is the mechanism, and nothing is published from the service to retry a message. .out_retry(..) over this descriptor is therefore a compile error, and the error names it.

Examples

.max_attempts(n).dead_letter(topic) at the mount site becomes the subscription’s own dead-letter policy, written when the service starts:

use std::time::Duration;

use ruststream_gcp_pubsub::prelude::*;
use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct Payment {
    id: u64,
    settled: bool,
}

#[subscriber(GooglePubSub::new("payments-workers").batch_wait(Duration::from_millis(200)))]
async fn reconcile(payments: &[Payment]) -> HandlerOutcome {
    if payments.iter().any(|payment| !payment.settled) {
        return HandlerOutcome::retry_after(Duration::from_secs(5));
    }
    tracing::info!(count = payments.len(), "settled");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // Pub/Sub gives one payment five deliveries, then publishes it to `payments-dead`.
            b.include(reconcile.batch(nonzero!(50)))
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments-dead");
        },
    )
}

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

Takes in the registration’s cap and dead-letter destination, which become the subscription’s dead-letter policy when subscribe opens it: the topic is the policy’s deadLetterTopic and the cap its maxDeliveryAttempts.

Only recorded here. The declaration turns into topology where the connection exists, and a half declaration is refused there rather than opening a subscription that honours neither half.

type Subscriber = PubSubSubscriber

The subscriber type this source opens.

type Copies = BrokerMoves

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, connected: &ConnectedPubSubBroker, ) -> Result<PubSubSubscriber, PubSubError>

Opens the subscription against the connected broker. Called once at startup.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.