Начало работы

Это руководство устанавливает ruststream-sea-file, запускает первый сервис на файле потока и переносит тот же обработчик в конвейер оболочки. Ни одному из транспортов не нужен сервер.

Установка

Cargo.toml

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json", "logging"] }
ruststream-sea-file = "0.7"
serde = { version = "1", features = ["derive"] }
tracing = "0.1"

[dev-dependencies]
ruststream-sea-file = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

ruststream - это сам фреймворк. Фича macros даёт атрибуты, json - кодек по умолчанию, а logging выводит журнал сервиса в терминал. ruststream-sea-file содержит оба транспорта. Его фича testing добавляет внутрипроцессные транспорты, на которых работают тесты, а asyncapi добавляет вклад крейта в документ AsyncAPI. Обе фичи по умолчанию выключены. Клиент файла потока собирается только в Unix-системах, поэтому в Windows крейт не собирается.

Первый сервис

Макросы

src/main.rs

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

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

#[subscriber(FileStream::new("orders"), start_at(FilePosition::beginning()))]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

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

Вручную

src/main.rs

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

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

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        FileBroker::new("/tmp/orders.ss"),
        |b| {
            b.include(
                subscriber(FileStream::new("orders"), TakeOrder)
                    .start_at(FilePosition::beginning())
                    .build(),
            );
        },
    )
}

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

take_order читает заказы в JSON под ключом потока orders из файла /tmp/orders.ss. Заказ с суммой он подтверждает, а заказ без суммы отбрасывает. FileBroker::new только запоминает путь. Файл открывается при запуске сервиса, и если его нет, запуск его создаёт. FileStream::new называет ключ потока, а start_at(FilePosition::beginning()) читает ключ с начала файла, а не с его конца. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

Запустите сервис:

cargo run -- run

Он читает все заказы, которые есть в файле, и ждёт следующего. Другой сервис дописывает заказ, публикуя его под ключом orders в тот же файл, как показывает руководство Журнал событий на диске. Ctrl+C останавливает сервис.

Содержимое файла

В составе файлового клиента есть декодер, который печатает файл потока по сообщению на строку:

cargo install sea-streamer-file --features=executables --bin ss-decode
ss-decode --file /tmp/orders.ss

Каждая строка имеет вид [timestamp | stream key | sequence | shard] payload, например [2026-10-18T09:12:03.120 | orders | 1 | 0] {"id":1,"amount_cents":1250}. Один файл хранит все ключи потока, и каждый ключ нумеруется с единицы. Новый файл хранит ещё одно сообщение под ключом ruststream-internal: его записывает крейт, когда создаёт файл.

Первый тест

В файле примера есть и два теста. Каждый запускает то же app через TestApp::start, публикует заказ и проверяет, как take_order завершил его доставку. Там FileBroker работает внутри процесса, поэтому тесты не трогают ни одного файла:

cargo test

Руководство Тестирование сервиса на файле или в конвейере разбирает обвязку и то, что моделируют внутрипроцессные транспорты.

Тот же обработчик в конвейере

Второй транспорт читает стандартный ввод и пишет в стандартный вывод. Обработчик остаётся прежним. Меняются прелюдия, брокер и подписка.

Макросы

src/main.rs

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

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

#[subscriber("orders")]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(StdioBroker::new(), |b| {
        b.include(take_order).out_retry(Publish).to("orders.retry");
    })
}

Вручную

src/main.rs

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

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

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(StdioBroker::new(), |b| {
        b.include(subscriber("orders", TakeOrder).build())
            .out_retry(Publish)
            .to("orders.retry");
    })
}

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

StdioBroker::new не принимает настроек, а подписка называет только ключ потока. Ступень называет ключ, под который уходят её отложенные копии, через out_retry(Publish).to(..), и без него она не запускается. Политика публикации в обеих прелюдиях называется Publish, поэтому код, который публикует, выглядит одинаково на любом транспорте.

Подайте ступени строку:

echo '[orders] {"id":1,"amount_cents":1250}' | cargo run -- run

Квадратные скобки называют ключ потока. Журнал уходит в стандартный поток ошибок, потому что стандартный вывод передаёт сообщения ступени следующему процессу. Формат строки и цепочки ступеней разбирает руководство Конвейеры оболочки.

Переход с sea-streamer

Сервис на sea-streamer-file сам пишет цикл консьюмера и продюсер. Вот куда переходит каждая часть в RustStream:

sea-streamer	RustStream
FileStreamer::connect с FileConnectOptions	FileBroker::new и его настройки, которые применяются при запуске
create_consumer и цикл next()	обработчик #[subscriber] на FileStream
декодирование полезной нагрузки сообщения	тип параметра обработчика, который декодирует кодек
create_producer и send_to	ответ или параметр Out, отправленный через Publish
set_auto_stream_reset в параметрах консьюмера, затем seek_to	start_at в точке монтирования или дескриптор перемотки в обработчике (Возобновление и перемотка)
StdioStreamer	StdioBroker