Getting started

This guide installs ruststream-gcp-pubsub, starts the Pub/Sub emulator and runs a first service against it.

Installation

Cargo.toml

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

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

ruststream is the framework: macros gives the attributes, json the default codec, and logging prints the service's log to the terminal. ruststream-gcp-pubsub is the Pub/Sub broker. Its testing feature adds the in-process Pub/Sub that tests run on.

The broker crate has two features, both off by default:

Feature	What it gives
testing	the in-process Pub/Sub behind TestApp
asyncapi	the Pub/Sub bindings in the generated AsyncAPI document

The client underneath is google-cloud-pubsub, written in Rust, so the build needs no system libraries. The crate needs Rust 1.95 or newer.

A local emulator

Google ships a Pub/Sub emulator in the Cloud SDK image:

compose.yaml

services:
  pubsub:
    image: gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
    command: gcloud beta emulators pubsub start --host-port=0.0.0.0:8085 --project=my-project
    ports:
      - "8085:8085"

docker compose up -d

The emulator listens on localhost:8085. It starts empty and accepts any project id.

The first service

Macros

src/main.rs

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

#[subscriber(GooglePubSub::new("orders-workers").create_with_topic("orders"))]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

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

Manual

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    amount_cents: u64,
}

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(
                subscriber(
                    GooglePubSub::new("orders-workers").create_with_topic("orders"),
                    TakeOrder,
                )
                .build(),
            );
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

take_order reads JSON orders from the subscription orders-workers. It takes an order with an amount and drops one without. A Pub/Sub service reads a subscription, never a topic. GooglePubSub describes the subscription, and create_with_topic creates it at startup on the topic orders, with the topic itself. PubSubBroker::new takes the project id and only records it. emulator points the broker at the local emulator, because the Rust client does not read PUBSUB_EMULATOR_HOST. The connection opens when the service starts. #[ruststream::app] turns app into main; in the manual form, run_main does the same.

Run the service:

cargo run -- run

Publish an order from another terminal, through the emulator's REST API:

curl -X POST localhost:8085/v1/projects/my-project/topics/orders:publish \
  -H 'content-type: application/json' \
  -d "{\"messages\": [{\"data\": \"$(printf '{"id":1,"amount_cents":1250}' | base64)\"}]}"

Pub/Sub carries the message body as base64 in its REST API, so the order is encoded first. The service logs order taken with the order's id. Ctrl+C stops it.

The first test

The example file also holds two tests. Each starts the same app with TestApp::start, publishes an order and checks how take_order settled it. PubSubBroker connects to an in-process Pub/Sub there, so the tests need no emulator:

cargo test

Testing a Pub/Sub service walks through such tests and what the in-process Pub/Sub models.

Coming from google-cloud-pubsub

A service on google-cloud-pubsub builds its clients, pulls in a loop and publishes by hand. Here is where each part goes in RustStream:

google-cloud-pubsub	RustStream
Subscriber::builder() with with_endpoint and with_credentials	PubSubBroker::new, with endpoint, credentials or emulator
subscriber.subscribe(name) and a loop over the message stream	a #[subscriber(..)] handler
set_max_outstanding_messages, set_max_lease_extension, set_max_lease	max_outstanding, ack_extension, max_lease on the descriptor (Flow control and leases)
create_topic and create_subscription on the admin clients	create_with_topic (Topics and subscriptions)
serde_json::from_slice over the message data	the handler's parameter type, decoded by the codec
handler.ack(), handler.nack()	the handler's outcome: ack, retry, retry_after, drop (Acknowledgement and retries)
Publisher::publish	a reply or an Out parameter, sent through the Publish policy (Publishing and replies)
Message::set_ordering_key	the ordering_key step (Ordering keys)