FileBroker

FileBroker::new records the path of a stream file; the runtime opens it at startup and creates it when it is missing.

A stream file holds one log per stream key, in the .ss format of sea-streamer-file. Each stream key is numbered from one, and the numbering carries on when a later run appends to the same file. A service that reads a file another service writes requires it with existing_only instead.

A publish appends the message under its stream key and flushes it, so the bytes reach the operating system before the call returns. The file is synced to the device when the broker shuts down. The file client caps a stream file at 16 GiB: a publish that reaches the cap fails, and every later publish to that file fails too.

The file client builds on Unix systems only, so the file transport does not build on Windows. Two more settings shape the file itself: end_with_eos marks it finished at shutdown, and beacon_interval sets how densely it is indexed.

Example

The service confirms each order on the stream key receipts of the same file. The test starts it on a path where no file exists yet, and the file is there once startup returns.

Macros
Manual

examples/file_broker.rs
/// The service on the stream file at `path`.
fn service(path: &str) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(FileBroker::new(path), |b| {
        b.include(confirm);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service("/var/lib/orders/orders.ss")
}

examples/manual/file_broker.rs
/// The service on the stream file at `path`.
fn service(path: &str) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(FileBroker::new(path), |b| {
        b.include(
            subscriber(FileStream::new("orders"), Confirm)
                .reply()
                .build(),
        );
    })
}

fn app() -> impl App<State = ()> {
    service("/var/lib/orders/orders.ss")
}

Example​

Example