Конвейеры оболочки

Это руководство запускает сервис как одну ступень конвейера оболочки, producer | stage | consumer.

StdioBroker читает стандартный ввод процесса как поток и пишет в его стандартный вывод. Подписка читает строки одного ключа потока, а публикация пишет строку для следующего процесса.

Пример - ступень расчёта цен: она читает заказы и отвечает на каждый квитанцией.

Ступень

Макросы

src/main.rs

use ruststream_sea_file::stdio::prelude::*;
use serde::{Deserialize, Serialize};

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

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

#[subscriber("orders", reply)]
async fn price(order: &Order) -> Receipt {
    let total_cents = order.quantity * order.unit_cents;
    // The log goes to standard error; standard output carries only messages.
    tracing::info!(order.id, total_cents, "order priced");
    Receipt {
        order: order.id,
        total_cents,
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("pricing", "0.1.0")).with_broker(StdioBroker::new(), |b| {
        // A deferred copy leaves on standard output under this key, for a later stage.
        b.include(price).out_retry(Publish).to("orders.retry");
    })
}

Вручную

src/main.rs

use ruststream_sea_file::stdio::prelude::*;
use serde::{Deserialize, Serialize};

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

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

struct Price;

impl Handle<Order, Receipt> for Price {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        let total_cents = order.quantity * order.unit_cents;
        // The log goes to standard error; standard output carries only messages.
        tracing::info!(order.id, total_cents, "order priced");
        ready(Ok(Receipt {
            order: order.id,
            total_cents,
        }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("pricing", "0.1.0")).with_broker(StdioBroker::new(), |b| {
        // A deferred copy leaves on standard output under this key, for a later stage.
        b.include(subscriber("orders", Price).reply().build())
            .out_retry(Publish)
            .to("orders.retry");
    })
}

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

price читает строки под ключом orders и отвечает квитанцией Receipt, которая уходит в стандартный вывод под ключом receipts. Ступени нужен out_retry(Publish).to(..), хотя price никогда не повторяет обработку. В канале отложенная копия может уйти только на следующую ступень, поэтому каждая регистрация называет ключ, и без него ступень не запускается.

Строка

Строка - это заголовок в квадратных скобках, а за ним полезная нагрузка (Строки):

[timestamp | stream_key | sequence | shard_id] payload

Каждое поле заголовка необязательно, поэтому продюсер обычно пишет только ключ:

echo '[orders] {"id":7,"quantity":3,"unit_cents":250}' | ./pricing run

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

jq -c '.[]' orders.json | ./pricing run

Клиент разбирает заголовок каждой строки до того, как её увидит сервис, а строка, которую он не может разобрать, останавливает весь процесс. Так действуют строки двух видов: ввод не в UTF-8 и начальные скобки [...], которые не являются заголовком, например массив JSON. Продюсер такой полезной нагрузки пишет перед каждой из них заголовок.

Вывод

Ступень пишет по строке на каждую публикацию, с заголовком, который заполняет клиент:

[2026-10-18T15:08:01 | receipts | 0] {"order":7,"total_cents":750}

Метка времени - это время записи с точностью до секунды, а порядковый номер считает строки каждого ключа с нуля. Инструмент, который читает простой JSON, берёт полезную нагрузку после заголовка:

./pricing run < orders.txt | sed 's/^\[[^]]*\] //' | jq .total_cents

Полезная нагрузка, которую изменила бы текстовая строка, уходит в конверте: двоичная, с переводом строки или с пробельными символами в начале или в конце, а также каждое сообщение с заголовками. Конверт - это строка, которая начинается с rs1: или rs2:. Он предназначен для следующей ступени RustStream: она вскрывает конверт и передаёт своему обработчику опубликованные байты и заголовки.

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

Цепочки ступеней

./intake run < orders.txt | ./pricing run | ./billing run

Каждая ступень читает ключи, на которые подписана, и пишет свои ответы для следующей. Заголовки и двоичная полезная нагрузка проходят каждый канал целиком, потому что каждая ступень вскрывает конверты, записанные предыдущей. Ступень, которая откладывает повторную обработку, отправляет свои копии дальше по конвейеру под ключом, который называет её точка монтирования, и их читает одна из следующих ступеней, как показывает руководство Повторы и ключ dead-letter.

Тестирование ступени

Файл примера тестирует ступень на внутрипроцессном транспорте: TestApp::start запускает её, tb.broker::<StdioBroker>().message(..) подаёт строку с заказом, а published::<Receipt>("receipts") читает записанную ступенью строку.

Живые тесты запускают собранную ступень с перенаправленными стандартными вводом и выводом. Они проверяют, что ступень отвечает на строки orders и на строки без ключа, пропускает другой ключ и останавливается на строке с массивом JSON и на вводе не в UTF-8. loopback подаёт ступени её собственный вывод, поэтому две ступени работают в одном процессе, и один тест управляет обеими.