Skip to main content

FileStream

FileStream names one stream key of the file and is the descriptor the generated document describes.

FileStream::new(key) goes where a stream key goes, as the first argument of #[subscriber]. It reads what the bare key reads: the messages appended under that key once the service has started. The file itself is named once, by FileBroker::new(path), so every descriptor of a service reads the same file. Where reading begins is a position, set with start_at. The descriptor adds two things to the bare key. replay() reads the retained file once and completes, a reading mode a position cannot express. Under the asyncapi feature the subscription's channel carries its stream key in the x-ruststream-file extension of the generated document, as Channels shows; a bare key describes nothing there. A retry_after copy is appended under the descriptor's key, so it comes back to the same subscription with nothing named at the mount site (retry_after). An empty key or a key the file does not take stops the service at startup. The manual form passes the descriptor to subscriber(..) where a key would go.

Example​

Two orders were in the file when the service started; the service ships the third, the first appended after it.

examples/file_stream.rs
/// Ships the orders appended under the `orders` key after the service started.
#[subscriber(FileStream::new("orders"))]
async fn ship(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "shipping");
    HandlerOutcome::ack()
}

/// The service on the stream file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(file, |b| {
        b.include(ship);
    })
}