跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

磁盘上的事件日志

本指南把一家商店的事件记录到流文件中，在写入的同时跟随读取，并一次读完已结束的文件做汇总。

流文件是一份日志：发布把消息追加到某个流键下，该键的每个读取者都按顺序看到它。 示例是一家商店，它为每个订单记录一张收据；还有一个审计作业，它汇总已结束文件中的收据。

记录与跟随​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
    amount_cents: u64,
    /// Where the order sits under `orders`, so a reader can move back to it.
    sequence: u64,
}

/// Records a receipt for each order, appended to the same file under `receipts`.
#[subscriber(FileStream::new("orders"), reply)]
async fn checkout(order: &Order, Ctx(at): Ctx<Position>) -> Receipt {
    let sequence = match at {
        FilePosition::Sequence(sequence) => sequence,
        // A delivery always sits at a sequence; the other forms name places to start from.
        FilePosition::Beginning | FilePosition::End | FilePosition::Timestamp(_) => 0,
    };
    Receipt {
        order: order.id,
        amount_cents: order.amount_cents,
        sequence,
    }
}

/// Follows the receipts as the shop writes them.
#[subscriber(FileStream::new("receipts"))]
async fn notify(receipt: &Receipt) -> HandlerOutcome {
    tracing::info!(receipt.order, "customer notified");
    HandlerOutcome::ack()
}

fn shop(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(file, |b| {
        b.include(checkout);
        b.include(notify);
    })
}

fn main() -> ExitCode {
    run_main(|| shop(FileBroker::new("/var/lib/shop/orders.ss")))
}

checkout 用一张 Receipt 回复每个订单，回复追加到同一个文件中，位于该类型声明的键 receipts 下。 一个文件容纳所有流键，所以路径只在 FileBroker::new 中指定一次。 发布追加消息并刷新到文件，因此调用返回之前，字节已经交给操作系统。 Broker 关闭时，文件同步到设备。

notify 从文件末尾跟随 receipts，在 checkout 追加每张收据时读取它。 打开同一路径的另一个进程也以同样的方式跟随该键。 没有指定位置的订阅从末尾开始，所以重启后的 notify 只读取它启动之后写入的收据。

每个流键的序号从一开始。 收据保存订单的序号，处理器通过 Ctx<Position> 得到它，读取者之后可以凭它回到那个订单。 之后向同一文件追加内容的运行会延续编号：重启后，下一个订单是三号，而不是一号。

文件客户端把流文件的大小上限定为 16 GiB。 达到上限的发布会返回错误，之后向该文件的每次发布也都会返回错误。 在生产环境中运行说明这对长期使用的文件意味着什么。

一次读完的作业​

src/bin/audit.rs
/// Totals the receipts of a finished file, up to 64 at a time; one outcome settles the batch.
#[subscriber(FileStream::new("receipts").replay())]
async fn total(batch: &[Receipt]) -> HandlerOutcome {
    let cents: u64 = batch.iter().map(|receipt| receipt.amount_cents).sum();
    tracing::info!(receipts = batch.len(), cents, "receipts totalled");
    HandlerOutcome::ack()
}

/// The audit job: it reads the shop's file and never creates one.
fn audit(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(file.existing_only(), |b| {
        b.include(total.batch(nonzero!(64)));
    })
}

replay() 从第一条消息开始读取文件中 receipts 下的内容，读到文件末尾即完成。 在那之后追加的收据不会被读取，所以作业读完一遍就无事可等。 existing_only 让启动在文件不存在时返回错误，这样，拼错的路径会让作业停止，而不是创建一个空文件。

total 接收一个切片，挂载点上的 batch(nonzero!(64)) 把一批的上限定为 64 张收据。 一批取走文件中已有的内容，不会等待凑满；一个结果结算整批。

写入者用 end_with_eos 把文件标记为已结束，它在关闭时追加一个流结束标记。 重放读到该标记即完成，notify 这样的实时订阅则在该标记处结束。

测试日志​

示例文件中的测试在进程内运行商店和审计：订单得到收据，notify 读到它；审计在一批中读到三张已记录的收据，并忽略第四张。 审计启动之前，它的收据记录在其 Broker 的一个克隆上，就像商店留下的那样。 重启测试用 TestApp::start_live 在一个真实文件上运行商店两次，并检查第二次运行把它的订单编为三号。 测试文件或管道服务讲解这两种模式。