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

Это руководство устанавливает ruststream-zeromq, запускает первый воркер и передаёт ему задания от узла в другом процессе.

Установка

Cargo.toml

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

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

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

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

Фича	Что она даёт
asyncapi	расширение x-ruststream-zeromq в генерируемом документе AsyncAPI
testing	внутрипроцессный режим, в котором работает TestApp

У ZeroMQ нет сервера, поэтому устанавливать и запускать ничего не нужно. Крейт работает по протоколу ZMTP через zeromq, клиент на Rust, поэтому сборке не нужны ни libzmq, ни компилятор C. Два процесса обмениваются сообщениями напрямую: через tcp:// или через ipc://, сокет Unix на одном хосте.

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

Макросы

src/main.rs

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

#[derive(Debug, PartialEq, Deserialize)]
struct Job {
    id: u64,
    items: u32,
}

#[subscriber("jobs")]
async fn take_job(job: &Job) -> HandlerOutcome {
    if job.items == 0 {
        tracing::warn!(job.id, "job without items dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(job.id, job.items, "job taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://127.0.0.1:5555")),
        |b| {
            b.include(take_job);
        },
    )
}

Вручную

src/main.rs

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

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

struct TakeJob;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://127.0.0.1:5555")),
        |b| {
            b.include(subscriber("jobs", TakeJob).build());
        },
    )
}

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

take_job читает задания в JSON из jobs. Задание с позициями он подтверждает, а задание без позиций отбрасывает. ZmqQueue - паттерн PUSH/PULL: каждое задание достаётся одному воркеру. ZmqEndpoint::bind означает, что этот процесс слушает адрес. ZmqQueue::new только запоминает конечную точку. Сокет открывается при запуске сервиса. #[ruststream::app] превращает app в main. В ручной форме то же делает run_main.

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

cargo run -- run

Узел в другом процессе

Отправлять задания воркеру может любой процесс, который работает по ZMTP. Сообщение состоит из трёх кадров: имени, заголовков и полезной нагрузки (формат передачи). Этот узел на Python использует pyzmq:

push_job.py

import json
import zmq

push = zmq.Context().socket(zmq.PUSH)
push.connect("tcp://127.0.0.1:5555")
push.send_multipart([b"jobs", b"", json.dumps({"id": 1, "items": 3}).encode()])
push.close(linger=1000)

Запустите его из другого терминала:

python3 push_job.py

Сервис пишет в журнал job taken с номером задания и числом позиций. За этой записью идёт предупреждение ack / nack failed: брокер не поддерживает подтверждение. В протоколе ZeroMQ подтверждения нет, поэтому задание считается выполненным, как только воркер его прочитал (подтверждение). Ctrl+C останавливает сервис. Узлом может быть и сервис на RustStream: в руководстве Конвейеры на PUSH/PULL такой сервис отправляет свои результаты другому.

Первый тест

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

cargo test

Руководство Тестирование сервиса на ZeroMQ разбирает обвязку и тот же тест на настоящих сокетах.

Переход с pyzmq или libzmq

Сервис на pyzmq или libzmq сам открывает сокеты и пишет свой цикл приёма. Вот куда переходит каждая часть в RustStream:

pyzmq или libzmq	RustStream
ctx.socket(zmq.PULL) и bind	ZmqQueue::new(ZmqEndpoint::bind(..)) и обработчик #[subscriber]
connect	ZmqEndpoint::connect; сторона входит в тип брокера
цикл recv_multipart	обработчик, который вызывается один раз на каждое сообщение
декодирование кадра полезной нагрузки	тип параметра обработчика, который декодирует кодек
send_multipart	ответ или параметр Out, отправленный через политику публикации
setsockopt(zmq.SUBSCRIBE, prefix)	имя подписки на ZmqFanout (Рассылка на PUB/SUB)
REQ и REP	ZmqRpc поверх DEALER и ROUTER (Запрос и ответ)
tcp://*:5555	tcp://0.0.0.0:5555
inproc://	TestApp в тестах; ipc:// между процессами на одном хосте
SNDHWM и RCVHWM	read_ahead (Промышленная эксплуатация)
zmq.Poller над несколькими сокетами	по подписке на каждый сокет, каждая в своей задаче