Broker

Two brokers and no server: FileBroker keeps a log in one .ss stream file, and StdioBroker makes the process's standard input and output a stream.

A stream file is a log on disk. It keeps what was written across restarts, and a subscriber can replay it or move back to an earlier message. A stdio service is one stage of a shell pipeline: it reads lines on standard input and writes lines on standard output. Neither transport has a host to connect to, so nothing is installed and nothing listens.

Inside one transport, messages are grouped by stream key, a name such as orders. A stream key in #[subscriber("orders")] is a subscription on both brokers. A publish names the stream key it goes to.

Both brokers follow the same lifecycle. new only records the configuration and does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. At startup the runtime connects the broker: the stream file is opened, or the process's pipes are attached. Every subscription and every publisher of the service opens against that connected form, ConnectedFileBroker or ConnectedStdioBroker. At shutdown the runtime closes it.

A service on one transport imports that transport's prelude. A failure of either broker is a SeaFileError.

Example

The service reads the stream key orders of the file /var/lib/orders/orders.ss from its start. cargo run --example file_service --features testing -- run starts it, and creates the file when it is missing.

Macros

examples/file_service.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/orders/orders.ss"),
        |b| {
            b.include(handle);
        },
    )
}

Manual

examples/manual/file_service.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/orders/orders.ss"),
        |b| {
            b.include(
                subscriber(FileStream::new("orders"), Accept)
                    .start_at(FilePosition::beginning())
                    .build(),
            );
        },
    )
}