Shutdown

Shutdown flushes and closes the stream file; a publisher kept from before returns SeaFileError::NotConnected, and a stdio shutdown ends every stdio stream of the process.

The file broker's shutdown waits for the publishes in flight, then ends the file's writer. The writer writes the end-of-stream mark when the broker asks for one, and syncs the file to the device. When shutdown returns, the file is closed, so an after_shutdown hook of the app can move, copy or remove it. Brokers opened on one path in one process share the file's writer: the first of them to shut down ends it, and a publish through the others fails.

A publisher taken before the shutdown, such as the one a startup hook received, still points at the closed broker. A publish through it returns NotConnected and never succeeds against the closed file.

The stdio transport belongs to the whole process. Shutting a StdioBroker down ends every stdio subscription and publisher in the process, not only the ones that broker opened. Each subscription first delivers the lines it has already read, then ends.

Example

Once the broker has closed the file, the app's after_shutdown hook moves it aside, and the next run starts a new one. The test runs the hook's function after the harness shut the broker down, and the moved file opens as a whole stream file.

Macros
Manual

examples/sea_broker_shutdown.rs
/// Moves the finished stream file aside, so the next run starts a new one.
fn archive(path: &str) -> io::Result<()> {
    fs::rename(path, format!("{path}.done"))
}

fn service(path: &str) -> impl App<State = ()> {
    let finished = path.to_owned();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Runs after the broker has shut down: the file is synced and closed.
        .after_shutdown(async move |_state| archive(&finished))
        .with_broker(FileBroker::new(path), |b| {
            b.include(record);
        })
}

examples/manual/sea_broker_shutdown.rs
/// Moves the finished stream file aside, so the next run starts a new one.
fn archive(path: &str) -> io::Result<()> {
    fs::rename(path, format!("{path}.done"))
}

fn service(path: &str) -> impl App<State = ()> {
    let finished = path.to_owned();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Runs after the broker has shut down: the file is synced and closed.
        .after_shutdown(async move |_state| archive(&finished))
        .with_broker(FileBroker::new(path), |b| {
            b.include(
                subscriber(FileStream::new("orders"), Record)
                    .reply()
                    .build(),
            );
        })
}

Example​

Example