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

Это руководство устанавливает ruststream-kinesis, запускает локальный Kinesis и первый сервис на нём.

Установка

Cargo.toml

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

[dev-dependencies]
ruststream-kinesis = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt"] }

ruststream - это сам фреймворк. Фича macros даёт атрибуты, json - кодек по умолчанию, а logging выводит журнал сервиса в терминал. ruststream-kinesis - брокер Kinesis, построенный на официальном клиенте aws-sdk-kinesis. Его фича testing добавляет Kinesis внутри процесса, на котором работают тесты.

У крейта брокера такие фичи, все выключены по умолчанию:

Фича	Что она даёт
dynamodb-lease	DynamoLeaseStore, чтобы несколько экземпляров сервиса делили сегменты (Несколько экземпляров)
asyncapi	сведения Kinesis о каждом канале в генерируемом документе AsyncAPI
testing	Kinesis внутри процесса, на котором работает TestApp

Локальный Kinesis

LocalStack запускает Kinesis в одном контейнере:

compose.yaml

services:
  localstack:
    image: localstack/localstack:4.14.0
    ports:
      - "4566:4566"
    environment:
      SERVICES: kinesis
      KINESIS_LATENCY: 0

docker compose up -d

Стенд отвечает на http://localhost:4566 и запускается без потоков. KINESIS_LATENCY: 0 отключает задержку, которую LocalStack по умолчанию добавляет к каждому вызову Kinesis.

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

Макросы

src/main.rs

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

#[subscriber(KinesisStream::new("orders").create_if_missing(1))]
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(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            b.include(take_order);
        },
    )
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
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(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            b.include(
                subscriber(KinesisStream::new("orders").create_if_missing(1), TakeOrder).build(),
            );
        },
    )
}

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

take_order читает заказы в JSON из потока orders. Заказ с суммой он подтверждает, а заказ без суммы отбрасывает. KinesisStream называет поток, а create_if_missing(1) создаёт его с одним сегментом, когда открывается подписка. Промышленный сервис обходится без этого шага и получает потоки от своей инфраструктуры. KinesisBroker::new только запоминает конфигурацию. endpoint, region и test_credentials направляют его на локальный стенд. Клиент SDK создаётся при запуске сервиса. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

Подписка открывается с конца потока, поэтому читает то, что опубликовано после запуска сервиса. Опубликуйте заказ из другого терминала через AWS CLI:

export AWS_ENDPOINT_URL=http://localhost:4566 AWS_REGION=us-east-1 \
  AWS_ACCESS_KEY_ID=test AWS_SECRET_ACCESS_KEY=test
aws kinesis put-record --stream-name orders --partition-key customer-7 \
  --cli-binary-format raw-in-base64-out --data '{"id":1,"amount_cents":1250}'

Сервис пишет в журнал order taken с номером заказа. Ctrl+C останавливает его.

Первый тест

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

cargo test

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

Переход с aws-sdk-kinesis

Rust SDK читает поток по одному сегменту и по одному вызову, и сервис на нём сам пишет потребителя вокруг этих вызовов. Вот куда переходит каждая часть в RustStream:

aws-sdk-kinesis	RustStream
aws_config::load_from_env и Client::new	KinesisBroker::new или from_config с собственным SdkConfig (Работа в промышленной среде)
ListShards и GetShardIterator для каждого сегмента	имя потока в KinesisStream::new; подписка сама находит сегменты и следит за разделениями и слияниями
цикл GetRecords для каждого сегмента	обработчик #[subscriber] (Чтение потока)
декодирование record.data()	тип параметра обработчика, который декодирует кодек
контрольная точка, записанная после работы	исход обработчика (Контрольные точки и гарантии доставки)
таблица аренд, чтобы несколько экземпляров делили сегменты	DynamoLeaseStore (Несколько экземпляров)
GetShardIterator с TRIM_HORIZON или AT_TIMESTAMP	start_at в точке монтирования или дескриптор перемотки в обработчике (Повторное чтение и перемотка)
PutRecord с ключом сегмента	ответ или параметр Out, отправленный через KinesisPublish (Сегменты, ключи сегмента и порядок)