Skip to main content

Running in production

A stream file is the service's own storage, so the facts a networked broker keeps on its servers are the service's here: where the file lives, when its bytes reach the device and how large it grows.

The file​

src/main.rs
/// The shop's stream file, at the path the deployment names.
fn broker() -> FileBroker {
    let path = env::var("ORDERS_FILE").unwrap_or_else(|_| "/var/lib/shop/orders.ss".to_owned());
    // A beacon every 256 KiB: a seek reads at most that far from the nearest one.
    FileBroker::new(path).beacon_interval(256 * 1024)
}

The deployment names the path in an environment variable. Startup creates the file when it is missing. A service that only reads a file another process writes opens it with existing_only, so a wrong path stops it at startup. beacon_interval sets how far apart the file's index entries sit. A seek reads forward from the nearest beacon, so closer beacons seek in finer steps and take more room; the client's default is one every 1 MiB. The interval must be a positive multiple of 1024, and any other value stops the service at startup. The file client builds on Unix systems only.

The service​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
}

#[subscriber(FileStream::new("orders"), reply)]
async fn checkout(order: &Order) -> Receipt {
    tracing::info!(order.id, "order confirmed");
    Receipt { order: order.id }
}

fn service(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(file, |b| {
        b.include(checkout);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker())
}

app builds the service on the production broker, and the tests start that same app.

What a publish writes​

A publish appends the message under its stream key and flushes it, so the bytes reach the operating system before the call returns. A live reader in this process or another one sees the message from then on. The file is synced to the device when the broker shuts down. Between those points the operating system decides when the bytes reach the device.

The file client caps a stream file at 16 GiB. A publish that reaches the cap fails with SeaFileError::Publish, and so does every later publish to that file. A long-lived log therefore moves to a new file before it reaches the cap: the service starts on a new path, and readers of the old file finish it with a replay.

Shutdown​

The broker's shutdown waits for the publishes in flight, writes the end-of-stream mark when the broker asks for one, and syncs the file. When shutdown returns the file is closed, so an after_shutdown hook can move or copy it. end_with_eos suits a writer whose file is complete when it stops, such as a daily log: a replay completes at the mark, and a live reader ends on it. A publisher kept from before the shutdown returns SeaFileError::NotConnected and never writes to the closed file.

A pipeline stage​

A stage's standard output carries its messages, so its log goes to standard error through tracing. A line on standard output from anywhere else in the process reaches the next stage as a message. The stdio transport belongs to the whole process: shutting a StdioBroker down ends every stdio subscription and publisher in it.

The AsyncAPI document​

Under the asyncapi feature the file broker is a server of the generated document with the protocol file, and its description is the path of the stream file. The document is shared with the people who read it, so the path is too.

Testing the configuration​

The tests start the production app under TestApp::start, so the settings are checked the way startup checks them, without a file. A beacon interval that is not a multiple of 1024 fails the start in the test as it does on a real file.