Getting started

This guide installs ruststream-pulsar, starts a local Pulsar and runs a first service against it.

Installation

Cargo.toml
[dependencies]
ruststream = { version = "0.7", features = ["macros", "json", "logging"] }
ruststream-pulsar = "0.7"
serde = { version = "1", features = ["derive"] }
tracing = "0.1"

[dev-dependencies]
ruststream-pulsar = { 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-pulsar is the Pulsar broker. Its testing feature adds the in-process Pulsar that tests run on.

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

Feature	What it gives
`testing`	the in-process Pulsar behind `TestApp`
`asyncapi`	the Pulsar bindings in the generated AsyncAPI document

System requirements

The Pulsar client compiles the Pulsar protocol definitions during the build, so the build needs protoc, the Protocol Buffers compiler. On Debian and Ubuntu it comes with protobuf-compiler, and on macOS with brew install protobuf. A protoc outside the path is named by the PROTOC environment variable.

A local Pulsar

Pulsar in standalone mode runs a broker and its storage in one container:

compose.yaml
services:
  pulsar:
    image: apachepulsar/pulsar:4.1.2
    command: bin/pulsar standalone
    ports:
      - "6650:6650"
      - "8080:8080"
    environment:
      # The address the broker hands to clients outside the container.
      PULSAR_PREFIX_advertisedAddress: 127.0.0.1
      PULSAR_PREFIX_advertisedListeners: pulsar://127.0.0.1:6650
    healthcheck:
      test: ["CMD-SHELL", "curl -fsS http://localhost:8080/admin/v2/brokers/health || exit 1"]
      interval: 10s
      retries: 30

Start it and wait until it is healthy:

docker compose up -d --wait

The broker listens on localhost:6650, and the admin API on localhost:8080. Pulsar creates a topic on its first use, so the service needs no setup.

The first service

Macros
Manual

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[subscriber(PulsarSubscription::new("orders", "billing"))]
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("billing", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(take_order);
        },
    )
}

src/main.rs
#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
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("billing", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(subscriber(PulsarSubscription::new("orders", "billing"), TakeOrder).build());
        },
    )
}

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

take_order reads JSON orders from the orders topic. It acknowledges an order with an amount and drops one without. Pulsar reads a topic only through a named subscription. PulsarSubscription::new names the topic and the subscription, here billing. The subscription is a cursor the server keeps, so the service resumes where it stopped after a restart. PulsarBroker::new takes the service URL and only records it. 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 with the client that ships in the image:

docker compose exec pulsar bin/pulsar-client produce orders \
  -s ';' -m '{"id":1,"amount_cents":1250}'

-s ';' keeps the message whole: by default the client splits it at every comma. 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. PulsarBroker connects to an in-process Pulsar there, so the tests need no server:

cargo test

Testing a Pulsar service walks through these tests and what the in-process Pulsar models.

Coming from the pulsar crate

A service on the pulsar crate builds its own client, consumer loop and producers. Here is where each part goes in RustStream:

`pulsar` crate	RustStream
`Pulsar::builder(url, TokioExecutor).build().await`	`PulsarBroker::new(url)`, connected by the runtime
`.with_auth(Authentication { name: "token", .. })`	`PulsarBroker::token`
`.with_operation_retry_options(..)`	`PulsarBroker::operation_retries`
`client.consumer().with_topic(..).with_subscription(..)` and a loop over the consumer	a `#[subscriber]` handler on a `PulsarSubscription`
`.with_subscription_type(..)`	`subscription_type` on the descriptor
`.with_topics([..])`, `.with_topic_regex(..)`	`PulsarSubscription::topics`, `PulsarSubscription::pattern`
deserializing `msg.payload.data`	the handler's parameter type, decoded by the codec
`consumer.ack(&msg)`, `consumer.nack(&msg)`	the handler's outcome (Acknowledgement)
`.with_dead_letter_policy(..)`	`max_attempts(n)` and `dead_letter(topic)` at the mount site (Retries and dead letters)
`.with_unacked_message_resend_delay(..)`	`ack_timeout` on the descriptor
`consumer.seek(..)`	`start_at` at the mount site, or the seek handle in the handler (Replay and seeking)
`client.producer().with_topic(..).build()` and `send_non_blocking`	a reply or an `Out` parameter, sent through the publish policy (Publishing and replies)
`producer::Message { partition_key, properties, .. }`	the `partition_key` step; headers become properties

Installation​

System requirements​

A local Pulsar​

The first service​

The first test​