Промышленная эксплуатация

У ZeroMQ нет сервера между сервисами, поэтому то, что иначе решал бы брокер очереди, решает развёртывание: кто слушает адрес, в каком порядке запускаются сервисы и сколько накапливает медленный потребитель.

Конечные точки из окружения

Макросы

src/main.rs

use std::env;
use std::num::NonZeroUsize;

use ruststream_zeromq::queue::prelude::*;

/// What the deployment decides: where the worker listens, where the sink is, and how far a
/// subscription reads ahead of its handler.
struct Settings {
    jobs: String,
    sink: String,
    read_ahead: NonZeroUsize,
}

impl Settings {
    fn from_env() -> Self {
        Self::from_vars(|name| env::var(name).ok())
    }

    fn from_vars(var: impl Fn(&str) -> Option<String>) -> Self {
        Self {
            jobs: var("JOBS_BIND").unwrap_or_else(|| "tcp://0.0.0.0:5555".to_owned()),
            sink: var("SINK_ADDR").unwrap_or_else(|| "tcp://sink:5558".to_owned()),
            read_ahead: var("ZMQ_READ_AHEAD")
                .and_then(|value| value.parse().ok())
                .unwrap_or(nonzero!(1000_usize)),
        }
    }
}

Вручную

src/main.rs

use std::env;
use std::num::NonZeroUsize;

use ruststream_zeromq::queue::prelude::*;

/// What the deployment decides: where the worker listens, where the sink is, and how far a
/// subscription reads ahead of its handler.
struct Settings {
    jobs: String,
    sink: String,
    read_ahead: NonZeroUsize,
}

impl Settings {
    fn from_env() -> Self {
        Self::from_vars(|name| env::var(name).ok())
    }

    fn from_vars(var: impl Fn(&str) -> Option<String>) -> Self {
        Self {
            jobs: var("JOBS_BIND").unwrap_or_else(|| "tcp://0.0.0.0:5555".to_owned()),
            sink: var("SINK_ADDR").unwrap_or_else(|| "tcp://sink:5558".to_owned()),
            read_ahead: var("ZMQ_READ_AHEAD")
                .and_then(|value| value.parse().ok())
                .unwrap_or(nonzero!(1000_usize)),
        }
    }
}

Конечные точки и предел упреждающего чтения - вся конфигурация ZeroMQ этого сервиса. Settings читает их из переменных окружения, которые задаёт развёртывание, и у каждой есть значение по умолчанию. from_vars принимает поиск переменной как функцию, поэтому тест передаёт свои значения, не трогая окружение процесса.

Макросы

src/main.rs

use serde::{Deserialize, Serialize};

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

#[derive(Debug, PartialEq, Serialize, Outgoing)]
#[outgoing(name = "results")]
struct Done {
    id: u64,
    items: u32,
}

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

fn service(settings: Settings) -> RustStream {
    let sink = ZmqQueue::new(ZmqEndpoint::connect(settings.sink)).bindable();
    let results = sink.bind(Publish);
    let jobs = ZmqQueue::new(ZmqEndpoint::bind(settings.jobs)).read_ahead(settings.read_ahead);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(jobs, |b| {
            b.include(work).out_reply(results);
        })
        .with_broker(sink, |_b| {})
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(Settings::from_env())
}

Вручную

src/main.rs

use serde::{Deserialize, Serialize};

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

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

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

impl MessageHeaders for Done {
    type Contract = NoHeaders;
}

struct Work;

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

fn service(settings: Settings) -> RustStream {
    let sink = ZmqQueue::new(ZmqEndpoint::connect(settings.sink)).bindable();
    let results = sink.bind(Publish);
    let jobs = ZmqQueue::new(ZmqEndpoint::bind(settings.jobs)).read_ahead(settings.read_ahead);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(jobs, |b| {
            b.include(subscriber("jobs", Work).reply().build())
                .out_reply(results);
        })
        .with_broker(sink, |_b| {})
}

fn app() -> RustStream {
    service(Settings::from_env())
}

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

Воркер привязывается к JOBS_BIND и подключается к приёмнику по SINK_ADDR. tcp://0.0.0.0:5555 слушает на всех интерфейсах, а tcp://10.0.0.5:5555 - на одном. ipc:///run/jobs.sock - сокет Unix для сервисов на одном хосте. Сторона входит в тип брокера, поэтому адрес выбирает развёртывание, а сторону - код.

Порядок запуска

Подписка, которая привязывается к адресу, открывает свой порт при старте. Если порт занят другим процессом, старт возвращает ошибку ZmqError::Endpoint. Подписка, которая подключается, ждёт, пока её узел начнёт слушать. Клиент повторяет подключение до 30 секунд, после чего старт возвращает ошибку ZmqError::Endpoint с адресом. Поэтому сервис, который подключается, может запуститься раньше своего узла, если тот начнёт слушать в течение этого времени.

Издатель, который привязывается к адресу, слушает с момента запуска сервиса, поэтому узел, который к нему подключается, успевает подключиться до первого сообщения. Издатель, который подключается сам, подключается при первой публикации. Отправка без подключённого узла повторяется пять секунд, а затем возвращает ошибку ZmqError::Send с адресом назначения.

Перезапуск узла

Издатель, который подключается к своему узлу, подключается заново, когда узел пропал, и доходит до него, когда тот вернулся. Издатель очереди узнаёт о потере по неудачной отправке: сообщение, отправленное в мёртвое соединение, теряется, следующая публикация возвращает ошибку ZmqError::Send, а следующая за ней подключается заново. Широковещательный издатель узнаёт о потере от своего сокета и подключается при следующей публикации. Широковещательную подписку, которая подключается, клиент подключает заново сам. Подписка, которая подключается к узлу PUSH, остаётся на открытом ею соединении, поэтому после перезапуска этого узла она ничего не получает, пока не перезапустится и сервис.

Упреждающее чтение и обратное давление

Подписка держит в канале доставки не больше read_ahead сообщений впереди своего обработчика, по умолчанию 1000. Дойдя до этого предела, она перестаёт читать, буферы сокетов заполняются, и отправитель ждёт. Так медленный обработчик замедляет отправителя, а не раздувает память сервиса. На PUB/SUB ждёт издатель: его сдерживает самый медленный подписчик с совпавшим префиксом. Предел задаётся для каждого брокера, здесь из ZMQ_READ_AHEAD.

Безопасность

Клиент договаривается о механизме NULL: узлы подключаются без учётных данных, а трафик передаётся незашифрованным. Поэтому сервис работает в сети, которой доверяет, или внутри туннеля: VPN, WireGuard или туннеля SSH. Генерируемый документ AsyncAPI сообщает хост и порт каждой конечной точки без учётных данных (сервер).

Завершение работы

При завершении работы сервис останавливает свои подписки и закрывает сокеты. Издатель, выданный раньше, после этого возвращает ошибку ZmqError::NotConnected, а не отправляет в пустоту. Публикацию в очередь можно безопасно отменить. Публикацию в широковещательную конечную точку, ответ и запрос отменять нельзя, поэтому прекращайте ожидание по таймауту, а не через ветку select! (отмена).

Проверка конфигурации

Тесты строят Settings через from_vars. Один проверяет значения по умолчанию при пустом окружении. Другой запускает настроенный сервис через TestApp::start и проверяет, что результат доходит до приёмника. Внутрипроцессный режим не открывает сокетов, поэтому адреса могут и не существовать.