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

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

Установка

Cargo.toml

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

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

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

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

Фича	Что она даёт
testing	Redis внутри процесса, на котором работает TestApp
asyncapi	расширение Redis в генерируемом документе AsyncAPI
tls-rustls, tls-rustls-ring, tls-native-tls	TLS на rustls с aws-lc-rs, на rustls с ring или на системной библиотеке TLS
sentinel-auth	отдельные учётные данные для узлов Sentinel
credential-provider	сменяемые учётные данные через колбэк

Локальный Redis

Для разработки достаточно одного сервера Redis:

docker run --name redis -p 6379:6379 redis:8.4

Сервер слушает localhost:6379. Redis 8.4 поддерживает все возможности крейта. Чтению с перехватом нужна как минимум эта версия. Valkey работает так же: RedisBroker::standalone принимает URL valkey://, а для TLS valkeys://.

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

Макросы

src/main.rs

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

#[subscriber(RedisStream::new("orders").group("orders-svc"))]
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(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(take_order);
        },
    )
}

Вручную

src/main.rs

#[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(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(
                subscriber(RedisStream::new("orders").group("orders-svc"), TakeOrder).build(),
            );
        },
    )
}

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

take_order читает заказы в JSON из потока Redis orders. Заказ с суммой он подтверждает, а заказ без суммы отбрасывает. RedisBroker::standalone принимает URL сервера и только запоминает его. Соединение открывается при запуске сервиса. Поток Redis читается через группу потребителей. RedisStream::new("orders").group("orders-svc") задаёт поток и группу, а сервис при запуске создаёт их, если их ещё нет. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

Добавьте заказ в поток из другого терминала:

docker exec redis redis-cli XADD orders '*' _payload '{"id":1,"amount_cents":1250}'

Полезная нагрузка хранится в поле _payload записи. Сервис пишет в журнал order taken с идентификатором заказа. Новая группа начинает с конца потока, поэтому сервис читает заказы, добавленные после его первого запуска. Ctrl+C останавливает сервис.

cargo generate создаёт заготовку такого же сервиса с роутером и ответом. Для каждой структуры Redis есть свой шаблон: redis-stream, redis-list и redis-pubsub.

cargo generate --git https://github.com/powersemmi/ruststream-fred templates/redis-stream --name my-service

Первый тест

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

cargo test

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

Переход с fred

Сервис на fred сам пишет цикл чтения, подтверждения и создание групп. Вот где каждая из этих частей находится в RustStream:

fred	RustStream
Builder::from_config(..) и Pool	RedisBroker::standalone, cluster или sentinel, размер задаёт pool; from_pool принимает пул, построенный вами
XGROUP CREATE .. MKSTREAM	выполняет подписка при запуске
цикл XREADGROUP	обработчик #[subscriber] на RedisStream
декодирование полей записи	тип параметра обработчика, который декодирует кодек
XACK после работы	исход обработчика
цикл XAUTOCLAIM	RedisStream::claiming или reclaim (Группы потребителей и восстановление после сбоя)
воркер на BLMOVE с LREM	RedisList::reliable (Очереди заданий на списках)
SubscriberClient и его поток сообщений	RedisPubSub (Рассылка через Pub/Sub)
XADD, LPUSH, PUBLISH	ответ или параметр Out, отправленный через политику публикации
MULTI / EXEC вокруг группы XADD	транзакция