Struct FileBroker 

pub struct FileBroker { /* private fields */ }

A log kept in one .ss stream file for the RustStream messaging framework: it survives restarts, replays what it recorded, and needs no server.

new is synchronous and records only the path; the file opens in the consuming Broker::connect, which creates it when it is missing. The file transport builds on Unix systems only, an upstream constraint of the file client.

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 at shutdown. 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. Brokers opened on one path in one process share the file’s writer: once one of them shuts down, a publish through another fails with SeaFileError::Publish carrying the client’s ProducerEnded.

The FileBroker page shows it in a service.

Examples

use ruststream_sea_file::file::prelude::*;
use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct Order {
    id: u64,
}

#[subscriber(FileStream::new("orders"), start_at(FilePosition::beginning()))]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

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

A test runs the app main runs. With the testing feature, TestApp::start connects this broker to a stream file held in memory, and the test addresses it by its production type:

use std::error::Error;

use ruststream::testing::TestApp;
use ruststream_sea_file::file::prelude::*;
use serde::{Deserialize, Serialize};

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

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

#[subscriber(FileStream::new("orders"), reply)]
async fn confirm(order: &Order) -> Receipt {
    Receipt { order: order.id }
}

/// The app `main` runs.
pub fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(FileBroker::new("/var/lib/orders.ss"), |b| {
            b.include(confirm);
        })
}

pub async fn confirms_an_order() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    // The publish returns once the handler it woke has settled.
    tb.broker::<FileBroker>()
        .message(&Order { id: 1 })
        .publish()
        .await?;

    tb.broker::<FileBroker>()
        .subscriber("orders")
        .assert_called(1)
        .with(&Order { id: 1 });
    tb.broker::<FileBroker>()
        .published::<Receipt>("receipts")
        .assert_called(1)
        .with(&Receipt { order: 1 });

    tb.shutdown().await?;
    Ok(())
}

Implementations

impl FileBroker

pub fn new(path: impl Into<String>) -> Self

Records the path of the stream file (created on connect when missing). No I/O.

pub fn existing_only(self) -> Self

Requires the file to exist instead of creating it on connect.

pub fn end_with_eos(self) -> Self

Writes an end-of-stream mark when the broker shuts down, marking the file finished.

A subscription that tails the file reads the mark and ends there, which is the only thing that ends a live subscription. A replay completes at the mark, or at the end of the file when there is none, after delivering every message before it.

pub fn beacon_interval(self, bytes: u32) -> Self

The interval of the file’s in-place index (must be a positive multiple of 1024); denser beacons make seeking finer-grained at the cost of file size.

pub fn publisher(&self) -> FilePublisher

A publisher sharing this broker’s connection cell; buildable before connect.

Trait Implementations

impl Broker for FileBroker

type Error = SeaFileError

The error type returned by broker-level operations.

type Connected = ConnectedFileBroker

The connected form of this broker: the typed witness that connect succeeded.

async fn connect(self) -> Result<Self::Connected, Self::Error>

Establishes the connection to the broker, consuming the unconnected form.

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

impl Clone for FileBroker

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for FileBroker

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl DescribeServer for FileBroker

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl InProcess for FileBroker

Available on crate feature testing only.

The in-process mode: the connected form a test runs the production app against, over a stream file held in memory in place of the one at the path.

The file is new and empty, and every clone of this broker shares it, as clones share the file they connect. The settings are checked the way connect checks them, so a broker a service could not connect is not one a test can connect either: the path must form a file address, and the beacon interval must be a positive multiple of 1024. existing_only reads the in-memory file as the one the operator provisioned.

async fn connect_in_process(self) -> Result<Self::Connected, Self::Error>

Connects to the in-process transport, consuming the configuration.