Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Журнал событий на диске

Это руководство записывает события магазина в файл потока, читает их по мере записи и подводит итоги законченного файла за один проход.

Файл потока - это журнал: публикация дописывает сообщение под ключом потока, и каждый читатель ключа видит сообщения по порядку. Пример - магазин, который записывает квитанцию на каждый заказ, и задание аудита, которое подводит итог квитанциям законченного файла.

Запись и чтение по мере записи​

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. Публикация дописывает сообщение и сбрасывает его в файл, поэтому байты попадают в операционную систему до возврата из вызова. Файл синхронизируется с устройством при завершении работы брокера.

notify читает receipts от конца файла и получает каждую квитанцию, как только checkout её допишет. Другой процесс, который открывает тот же путь, читает ключ точно так же. Подписка без позиции начинает с конца, поэтому перезапущенный notify читает только квитанции, записанные после его запуска.

Каждый ключ потока нумеруется с единицы. Квитанция хранит порядковый номер заказа, который обработчику даёт Ctx<Position>, поэтому читатель может позже вернуться к этому заказу. Следующий запуск, который дописывает в тот же файл, продолжает нумерацию: после перезапуска следующий заказ получает номер три, а не один.

Файловый клиент ограничивает размер файла потока 16 ГиБ. Публикация, которая достигает предела, возвращает ошибку, и так же завершается каждая следующая публикация в этот файл. Что это значит для долгоживущего файла, разбирает руководство Промышленная эксплуатация.

Задание в один проход​

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 её читает, а аудит читает три записанные квитанции одним пакетом и пропускает четвёртую. Квитанции для аудита записываются через клон его брокера до запуска, так, как их оставил бы магазин. Тест перезапуска дважды запускает магазин на одном настоящем файле через TestApp::start_live и проверяет, что второй запуск даёт своему заказу номер три. Оба режима разбирает руководство Тестирование сервиса на файле или в конвейере.