Узлы на других языках

Это руководство связывает сервис на RustStream с узлами на других языках: PUSH, который передаёт сервису задания, SUB, который читает его оповещения, и DEALER, который отправляет ему запросы.

Узлы здесь используют pyzmq, обёртку libzmq для Python. Узел на C, C++ или другом языке составляет те же кадры через свою обёртку.

Сервис

Макросы

src/main.rs

use ruststream_zeromq::prelude::*;
use serde::{Deserialize, Serialize};

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

/// Announced on the fan-out, under a name a `jobs.` subscriber matches.
#[derive(Debug, PartialEq, Serialize, Outgoing)]
#[outgoing(name = "jobs.done")]
struct JobDone {
    id: u64,
    items: u32,
}

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

// No destination of its own: the ROUTER addresses each answer per request.
#[derive(Debug, PartialEq, Serialize, Outgoing)]
struct Score {
    customer: u64,
    approved: bool,
}

#[subscriber("jobs", reply)]
async fn work(job: &Job) -> JobDone {
    tracing::info!(job.id, job.items, "job worked");
    JobDone {
        id: job.id,
        items: job.items,
    }
}

#[subscriber("score", reply("reply"))]
async fn score(request: &ScoreRequest) -> Score {
    tracing::info!(request.customer, request.amount_cents, "scoring");
    Score {
        customer: request.customer,
        approved: request.amount_cents <= 50_000,
    }
}

/// The service on the three endpoints it binds: jobs in, announcements out, requests answered.
fn service(jobs: &str, events: &str, scoring: &str) -> RustStream {
    let events = ZmqFanout::new(ZmqEndpoint::bind(events)).bindable();
    let announce = events.bind(ZmqFanoutPublish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::bind(jobs)), |b| {
            b.include(work).out_reply(announce);
        })
        .with_broker(events, |_b| {})
        .with_broker(ZmqRpc::new(ZmqEndpoint::bind(scoring)), |b| {
            b.include(score)
                .out_reply(ZmqRpcPublish)
                .transform(ReplyToRequester)
                .out_retry(ZmqRpcPublish)
                .to("score.retry");
        })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(
        "tcp://0.0.0.0:5555",
        "tcp://0.0.0.0:5556",
        "tcp://0.0.0.0:5557",
    )
}

Вручную

src/main.rs

use ruststream_zeromq::prelude::*;
use serde::{Deserialize, Serialize};

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

/// Announced on the fan-out, under a name a `jobs.` subscriber matches.
#[derive(Debug, PartialEq, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct JobDone {
    id: u64,
    items: u32,
}

// `#[outgoing(name = "jobs.done")]` by hand.
impl OutgoingDestination for JobDone {
    type Form = FixedName;
    const DESTINATION: &'static str = "jobs.done";
}

impl MessageHeaders for JobDone {
    type Contract = NoHeaders;
}

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct ScoreRequest {
    customer: u64,
    amount_cents: u64,
}

#[derive(Debug, PartialEq, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Score {
    customer: u64,
    approved: bool,
}

// `#[derive(Outgoing)]` with no name by hand: the ROUTER addresses each answer per request.
impl OutgoingDestination for Score {
    type Form = CallerName;
}

impl MessageHeaders for Score {
    type Contract = NoHeaders;
}

struct Work;

impl Handle<Job, JobDone> for Work {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<JobDone, HandlerOutcome>> {
        tracing::info!(job.id, job.items, "job worked");
        ready(Ok(JobDone {
            id: job.id,
            items: job.items,
        }))
    }
}

struct ScoreIt;

impl Handle<ScoreRequest, Score> for ScoreIt {
    fn handle(
        &self,
        request: &ScoreRequest,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Score, HandlerOutcome>> {
        tracing::info!(request.customer, request.amount_cents, "scoring");
        ready(Ok(Score {
            customer: request.customer,
            approved: request.amount_cents <= 50_000,
        }))
    }
}

/// The service on the three endpoints it binds: jobs in, announcements out, requests answered.
fn service(jobs: &str, events: &str, scoring: &str) -> RustStream {
    let events = ZmqFanout::new(ZmqEndpoint::bind(events)).bindable();
    let announce = events.bind(ZmqFanoutPublish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::bind(jobs)), |b| {
            b.include(subscriber("jobs", Work).reply().build())
                .out_reply(announce);
        })
        .with_broker(events, |_b| {})
        .with_broker(ZmqRpc::new(ZmqEndpoint::bind(scoring)), |b| {
            b.include(subscriber("score", ScoreIt).reply().to("reply").build())
                .out_reply(ZmqRpcPublish)
                .transform(ReplyToRequester)
                .out_retry(ZmqRpcPublish)
                .to("score.retry");
        })
}

fn app() -> RustStream {
    service(
        "tcp://0.0.0.0:5555",
        "tcp://0.0.0.0:5556",
        "tcp://0.0.0.0:5557",
    )
}

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

Сервис привязывается к трём конечным точкам. jobs на порту 5555 - это ZmqQueue, которая принимает задания от PUSH. О каждом выполненном задании сервис оповещает как jobs.done через ZmqFanout на порту 5556. score на порту 5557 отвечает на запросы через ZmqRpc. ReplyToRequester - преобразование из руководства Запрос и ответ.

Три кадра

Каждое сообщение состоит из трёх кадров (формат передачи):

frame 0: name      UTF-8; on PUB/SUB also what a subscriber's prefix is matched against
frame 1: headers   UTF-8 "name: value" lines separated by \n; may be empty
frame 2: payload   what the codec wrote; JSON by default

Узел, который отправляет два кадра, имя и полезную нагрузку, отправляет сообщение без заголовков. Пустая строка в кадре заголовков пропускается, поэтому перевод строки в конце строки допустим. Если кадр заголовков или имя не в UTF-8 либо в строке заголовка нет :, подписка вместо доставки возвращает ошибку ZmqError::Wire. Сервис проверяет заголовки, которые публикует, чтобы каждый читался обратно без изменений. Если заголовок не текстовый, содержит перевод строки, имеет : в имени или начинается либо заканчивается пробельным символом, публикация возвращает ошибку ZmqError::Send с именем заголовка и адресом назначения.

PUSH, который передаёт задания

push.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"content-type: application/json",
    json.dumps({"id": 1, "items": 3}).encode(),
])
push.close(linger=1000)

Кадр имени здесь jobs, но привязанный PULL принимает каждое отправленное ему сообщение, какое бы имя в нём ни стояло.

SUB, который читает оповещения

watch.py

import json
import zmq

sub = zmq.Context().socket(zmq.SUB)
sub.connect("tcp://127.0.0.1:5556")
sub.setsockopt(zmq.SUBSCRIBE, b"jobs.")
name, headers, payload = sub.recv_multipart()
print(name.decode(), json.loads(payload))

Подписка - префикс кадра имени, как у наблюдателя на RustStream (Рассылка на PUB/SUB). SUB получает оповещения, опубликованные после того, как он присоединился к широковещательной конечной точке, и ни одного из более ранних. Запустите watch.py, затем push.py, и через секунду запустите push.py ещё раз, чтобы оповещение было опубликовано уже после присоединения SUB. watch.py печатает первое полученное оповещение и завершается: jobs.done {'id': 1, 'items': 3}.

DEALER, который отправляет запросы

ask.py

import json
import zmq

dealer = zmq.Context().socket(zmq.DEALER)
dealer.connect("tcp://127.0.0.1:5557")
dealer.send_multipart([
    b"score",
    b"correlation-id: 42",
    json.dumps({"customer": 3, "amount_cents": 1250}).encode(),
])
name, headers, payload = dealer.recv_multipart()
print(name, headers, json.loads(payload))

Запрашивающая сторона по другую сторону ZmqRpc - это DEALER, и он отправляет три кадра. ROUTER ставит перед ними идентичность DEALER, а в запрос добавляет заголовок reply-to. Ответ возвращается тремя кадрами, и кадр 0 - буквально reply (конверт). В заголовках ответа стоит correlation-id запроса, поэтому DEALER с несколькими запросами в работе сопоставляет каждый ответ со своим запросом:

b'reply' b'correlation-id: 42' {'customer': 3, 'approved': True}

Кодеки

Кадр полезной нагрузки содержит то, что записал кодек, поэтому узел договаривается с сервисом о кодеке, а не о типах Rust. По умолчанию это JSON. msgpack и cbor - фичи ruststream. Байты, которые узел оформил сам, проходят без изменений в обёртке #[derive(Outgoing, Serialized)] (формат передачи).

Версии и безопасность

Клиент приветствует каждый узел по ZMTP 3.0. libzmq 4 работает по ZMTP 3.1 и принимает узел на 3.0, а клиент zeromq проверен на совместимость с libzmq. Рукопожатие использует механизм NULL, поэтому узел присоединяется без учётных данных. Трафик не шифруется. Где его запускать, описывает руководство Промышленная эксплуатация.

Контракт в генерируемом документе

С фичей asyncapi генерируемый документ сообщает узлу, как присоединиться к каждому каналу (AsyncAPI). Сервер сообщает транспорт, адрес и сторону, которую занимает этот сервис. Канал сообщает свою пару сокетов, а на PUSH/PULL и PUB/SUB ещё и nameFrame - кадр 0, с которым узел отправляет сообщения или подписывается. Канал ответов отвечающей стороны вместо адреса указывает на заголовок reply-to.

Проверка кадров

Внутрипроцессный тест отправляет задание так, как это сделал бы узел, и проверяет оповещение, опубликованное в jobs.done. Второй тест запускает сервис на настоящих сокетах через TestApp::start_live. Роль чужого узла играет сокет DEALER клиента zeromq без фреймворка: он вручную составляет кадры запроса и проверяет три кадра ответа. Отвечающая сторона привязывается к пути сокета, который принадлежит только этому тесту, поэтому параллельные запуски никогда не встречаются.