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

Это руководство устанавливает ruststream-rdkafka, запускает локальную Kafka и первый сервис на ней.

Установка

Cargo.toml

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

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

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

У крейта брокера такие фичи:

Фича	Что она даёт
json (по умолчанию), msgpack, cbor	одноимённый кодек
schema-registry	клиент Confluent Schema Registry и его формат данных
avro	Avro поверх Schema Registry
protobuf	сообщения Protobuf в конверте Confluent
asyncapi	привязки Kafka в генерируемом документе AsyncAPI
testing	Kafka внутри процесса, на которой работает TestApp
ssl	TLS через системный OpenSSL
ssl-vendored	TLS через OpenSSL, собранный из исходников
zstd	кодек сжатия zstd

Системные требования

Крейт собирает из исходников librdkafka, клиент Kafka на C. Для сборки нужны компилятор C и make. Конфигурацию librdkafka выполняет собственный скрипт. В Debian и Ubuntu оба инструмента ставит пакет build-essential. Фиче ssl нужны ещё заголовки OpenSSL (libssl-dev), а ssl-vendored нужен perl, чтобы собрать OpenSSL. Сборку через CMake и системную librdkafka даёт rust-rdkafka. Добавьте rdkafka в свои зависимости с нужной фичей, и сборка её подхватит.

Локальная Kafka

Для разработки достаточно одного узла Kafka в режиме KRaft:

compose.yaml

services:
  kafka:
    image: apache/kafka:4.0.0
    ports:
      - "9092:9092"

Запустите его и создайте тему, которую читает сервис:

docker compose up -d
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create --topic orders

Брокер слушает localhost:9092.

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

Макросы

src/main.rs

#[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(
        KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
        |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(
        KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
        |b| {
            b.include(subscriber("orders", TakeOrder).build());
        },
    )
}

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

take_order читает заказы в JSON из темы orders. Заказ с суммой он подтверждает, а заказ без суммы отбрасывает. KafkaBroker::new принимает начальные адреса кластера и только запоминает их. Соединение открывается при запуске сервиса. Kafka читает тему только в составе группы консьюмеров. default_group - группа каждой подписки, которая не называет свою. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

Опубликуйте запись с заказом из другого терминала консольным продюсером, который входит в образ:

echo '{"id":1,"amount_cents":1250}' | docker compose exec -T kafka \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders

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

cargo generate создаёт такой же сервис с роутером и ответом:

cargo generate --git https://github.com/powersemmi/ruststream-rdkafka templates/kafka-topic --name my-service

Первый тест

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

cargo test

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

Переход с rust-rdkafka

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

rust-rdkafka	RustStream
ClientConfig::new().set("group.id", ..)	default_group или KafkaTopic::group для отдельной подписки
ClientConfig::set для любого другого свойства	KafkaBroker::config, producer_config, KafkaTopic::config
StreamConsumer::subscribe и цикл recv	обработчик #[subscriber]
десериализация message.payload()	тип параметра обработчика, который декодирует кодек
store_offset или commit_message после работы	Commit::Tracked и исход обработчика (Фиксация и гарантии доставки)
FutureProducer::send	ответ или параметр Out, отправленный через политику публикации
seek и assign	start_at в точке монтирования или ручка перемотки в обработчике (Повторное чтение и перемотка)
send_offsets_to_transaction	EosPublish (Транзакции и exactly-once)