Getting started

This guide installs ruststream-nats, starts a local NATS server and runs a first service against it.

Installation

Cargo.toml

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

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

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

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

The client underneath is async-nats, written in Rust with TLS through rustls, so the build needs no system libraries.

A local NATS

One NATS server with JetStream is enough to develop against:

compose.yaml

services:
  nats:
    image: nats:2-alpine
    command: ["-js"]
    ports:
      - "4222:4222"

docker compose up -d

The server listens on localhost:4222. -js turns on JetStream, the streams that the guides from Streams and consumers on read and write. The examples publish with the NATS command-line tool, nats.

The first service

Macros

src/main.rs

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

#[subscriber("orders.created")]
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(
        NatsBroker::new("nats://localhost:4222"),
        |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(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(subscriber("orders.created", TakeOrder).build());
        },
    )
}

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

take_order reads JSON orders from the Core NATS subject orders.created. It takes an order with an amount and drops one without. NatsBroker::new takes the server address 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:

nats pub orders.created '{"id":1,"amount_cents":1250}'

The service logs order taken with the order's id. Core NATS has no acknowledgement, so the log also shows ack / nack failed for each order: the runtime reports that the delivery could not be settled. Acknowledgement and delivery guarantees explains when that matters and how a stream settles each message. Ctrl+C stops the service.

cargo generate scaffolds the same kind of service with a router:

cargo generate --git https://github.com/powersemmi/ruststream-nats templates/nats --name my-service

templates/nats-js scaffolds a service that reads a stream through a durable consumer.

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. NatsBroker connects to an in-process NATS there, so the tests need no server:

cargo test

Testing a NATS service walks through such tests and what the in-process NATS models.

Coming from async-nats

A service on async-nats writes its own subscription loops and publishes by hand. Here is where each part goes in RustStream:

async-nats	RustStream
async_nats::connect or ConnectOptions::connect	NatsBroker::new, with with_options for the ConnectOptions
client.subscribe and a while let Some(message) loop	a #[subscriber("subject")] handler
client.queue_subscribe	CoreSubject::queue_group (Subjects, wildcards and queue groups)
serde_json::from_slice(&message.payload)	the handler's parameter type, decoded by the codec
client.publish	a reply or an Out parameter, sent through the Publish policy
client.request	request on an Out<impl RequestReply, _> slot (Request-reply)
create_consumer on a stream and a messages() loop	JetStreamSubject (Streams and consumers)
message.ack(), AckKind::Nak, AckKind::Term	the handler's outcome: ack, retry, retry_after, drop (Acknowledgement and delivery guarantees)
jetstream.publish and its acknowledgement	JetStreamPublish (Publishing to streams)