Skip to main content

Getting started

This guide installs ruststream-sea-file, runs a first service on a stream file and moves the same handler into a shell pipeline. Neither transport needs a server.

Installation​

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

[dev-dependencies]
ruststream-sea-file = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

ruststream is the framework: macros gives the attributes, json the default codec, and logging prints the service's log to the terminal. ruststream-sea-file holds both transports. Its testing feature adds the in-process transports that tests run on, and asyncapi adds what the crate contributes to the AsyncAPI document. Both features are off by default. The stream file client builds on Unix systems only, so the crate does not build on Windows.

The first service​

src/main.rs
use ruststream_sea_file::file::prelude::*;
use serde::Deserialize;

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

#[subscriber(FileStream::new("orders"), start_at(FilePosition::beginning()))]
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(
        FileBroker::new("/tmp/orders.ss"),
        |b| {
            b.include(take_order);
        },
    )
}

take_order reads JSON orders under the stream key orders of the file /tmp/orders.ss. It acknowledges an order with an amount and drops one without. FileBroker::new only records the path. The file opens when the service starts, and startup creates it when it is missing. FileStream::new names the stream key, and start_at(FilePosition::beginning()) reads the key from the start of the file instead of from its end. #[ruststream::app] turns app into main; in the manual form, run_main does the same.

Run the service:

cargo run -- run

It reads every order the file holds, then waits for the next one. Another service appends an order by publishing it under orders on the same path, as An event log on disk shows. Ctrl+C stops the service.

Inside the file​

The file client ships a decoder that prints a stream file one message per line:

cargo install sea-streamer-file --features=executables --bin ss-decode
ss-decode --file /tmp/orders.ss

Each line is [timestamp | stream key | sequence | shard] payload, for example [2026-10-18T09:12:03.120 | orders | 1 | 0] {"id":1,"amount_cents":1250}. One file holds every stream key, and each key is numbered from one. A new file also holds one message under the key ruststream-internal, which the crate writes when it creates the file.

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. FileBroker runs in process there, so the tests touch no file:

cargo test

Testing a file or pipeline service walks through the harness and what the in-process transports model.

The same handler in a pipeline​

The other transport reads standard input and writes standard output. The handler stays as it is; the prelude, the broker and the subscription change.

src/main.rs
use ruststream_sea_file::stdio::prelude::*;
use serde::Deserialize;

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

#[subscriber("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(StdioBroker::new(), |b| {
        b.include(take_order).out_retry(Publish).to("orders.retry");
    })
}

StdioBroker::new takes no settings, and the subscription names the stream key alone. A stage names the key its deferred copies go to with out_retry(Publish).to(..), and it does not start without one. The publish policy is called Publish in both preludes, so code that publishes reads the same on either transport.

Feed the stage a line:

echo '[orders] {"id":1,"amount_cents":1250}' | cargo run -- run

The square brackets name the stream key. The log goes to standard error, because standard output carries the stage's messages to the next process. Shell pipelines covers the line format and chained stages.

Coming from sea-streamer​

A service on sea-streamer-file writes its own consumer loop and producer. Here is where each part goes in RustStream:

sea-streamerRustStream
FileStreamer::connect with FileConnectOptionsFileBroker::new and its settings, applied at startup
create_consumer and a next() loopa #[subscriber] handler on a FileStream
decoding the message payloadthe handler's parameter type, decoded by the codec
create_producer and send_toa reply or an Out parameter, sent through Publish
set_auto_stream_reset on the consumer options, then seek_tostart_at at the mount site, or the seek handle in the handler (Resuming and seeking)
StdioStreamerStdioBroker