Промышленная эксплуатация

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

Файл

src/main.rs

/// The shop's stream file, at the path the deployment names.
fn broker() -> FileBroker {
    let path = env::var("ORDERS_FILE").unwrap_or_else(|_| "/var/lib/shop/orders.ss".to_owned());
    // A beacon every 256 KiB: a seek reads at most that far from the nearest one.
    FileBroker::new(path).beacon_interval(256 * 1024)
}

Путь развёртывание передаёт в переменной окружения. Если файла нет, запуск его создаёт. Сервис, который только читает файл, записанный другим процессом, открывает его с existing_only, поэтому неверный путь останавливает его при запуске. beacon_interval задаёт расстояние между записями индекса в файле. Перемотка читает вперёд от ближайшего маяка, поэтому частые маяки делают шаг перемотки точнее и занимают больше места. По умолчанию клиент ставит маяк через каждый 1 МиБ. Интервал должен быть положительным и кратным 1024, а любое другое значение останавливает сервис при запуске. Файловый клиент собирается только в Unix-системах.

Сервис

Макросы

src/main.rs

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

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

#[subscriber(FileStream::new("orders"), reply)]
async fn checkout(order: &Order) -> Receipt {
    tracing::info!(order.id, "order confirmed");
    Receipt { order: order.id }
}

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

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker())
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Receipt {
    order: u64,
}

struct Checkout;

impl Handle<Order, Receipt> for Checkout {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        tracing::info!(order.id, "order confirmed");
        ready(Ok(Receipt { order: order.id }))
    }
}

fn service(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(file, |b| {
        b.include(
            subscriber(FileStream::new("orders"), Checkout)
                .reply()
                .build(),
        );
    })
}

fn app() -> impl App<State = ()> {
    service(broker())
}

fn main() -> ExitCode {
    run_main(app)
}

app строит сервис на рабочем broker, и тесты запускают то же app.

Что записывает публикация

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

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

Завершение работы

При завершении работы брокер ждёт публикаций, которые ещё выполняются, пишет метку конца потока, если она задана, и синхронизирует файл. Когда завершение работы возвращает управление, файл закрыт, поэтому хук after_shutdown может переместить или скопировать его. end_with_eos подходит писателю, у которого файл закончен к моменту остановки, например ежедневному журналу: воспроизведение на метке завершается, а живой читатель на ней заканчивает чтение. Издатель, сохранённый с момента до завершения работы, возвращает SeaFileError::NotConnected и никогда не пишет в закрытый файл.

Ступень конвейера

Стандартный вывод ступени передаёт её сообщения, поэтому её журнал уходит в стандартный поток ошибок через tracing. Строка в стандартном выводе из любого другого места процесса доходит до следующей ступени как сообщение. Транспорт stdio принадлежит всему процессу: завершение работы StdioBroker закрывает в нём каждую подписку и каждого издателя stdio.

Документ AsyncAPI

С фичей asyncapi файловый брокер становится сервером генерируемого документа с протоколом file, а его описание - это путь к файлу потока. Документ видят все, кто его читает, а значит, и путь.

Проверка конфигурации

Тесты запускают рабочее app через TestApp::start, поэтому настройки проверяются так же, как их проверяет запуск, но без файла. Интервал маяков, не кратный 1024, проваливает запуск в тесте так же, как на настоящем файле.