Устройство

Где стоит ruststream-zeromq: между рантаймом RustStream и клиентом zeromq, свой сокет на каждую подписку и каждого издателя, без сервера.

Схема текстом

1. Ваш сервис: #[subscriber] обработчики-подписчики; RustStream::new(..) .with_broker(..)
2. Ядро RustStream: Рантайм: диспетчеризация, кодеки, повторы, пакеты, собранные на стороне клиента
3. ruststream-zeromq: Подписка: задача читает сокет в ограниченный канал; Издатель: сокет bind: при старте dial: при отправке; ZmqQueue, ZmqFanout, ZmqRpc; Разделяемое состояние: конечная точка, слушатель, закрытие; Раскладка кадров имя, заголовки, нагрузка
4. Клиент zeromq: PULL, SUB, ROUTER; PUSH, PUB, XPUB, ROUTER, DEALER
5. #[subscriber] обработчики-подписчики → Рантайм: диспетчеризация, кодеки, повторы, пакеты, собранные на стороне клиента: монтируются
6. Рантайм: диспетчеризация, кодеки, повторы, пакеты, собранные на стороне клиента → Подписка: задача читает сокет в ограниченный канал: опрашивает
7. Подписка: задача читает сокет в ограниченный канал → PULL, SUB, ROUTER
8. Издатель: сокет bind: при старте dial: при отправке → PUSH, PUB, XPUB, ROUTER, DEALER
9. PULL, SUB, ROUTER → Узлы: другие процессы на любом языке, без сервера между ними
10. PUSH, PUB, XPUB, ROUTER, DEALER → Узлы: другие процессы на любом языке, без сервера между ними
11. С фичей testing место сокетов может занять внутрипроцессный транспорт

Крейт реализует трейты брокера из ядра поверх zeromq, клиента ZeroMQ, написанного на Rust. Рантайм над ним один для всех брокеров. Всё, что относится к ZeroMQ, лежит в этом крейте: три брокера, по одному на паттерн сокетов, раскладка кадров, смысл ack и адрес копии для повтора.

Каждый брокер хранит сторону своей конечной точки в типе: Bind или Connect. От стороны зависит, может ли подписка сама указать адрес для своих копий повтора, и тип делает это фактом времени компиляции. У ZmqRpc параметра стороны нет: его копии всегда уходят туда, куда указывает точка монтирования.

Подписка - это имя, брокер хранит read_ahead, а остальная конфигурация - это конечная точка. Публикация несёт только свои кадры, потому что отправка ZMTP принимает только кадры.

Каждая подписка открывает собственный сокет и задачу, которая им владеет. Задача читает по одному сообщению в канал, из которого их берёт обработчик, как показывает страница Обратное давление. У каждого издателя собственный сокет: издатель, который привязывается к адресу, подключает его при старте, поэтому узлы успевают подключиться к нему до первого сообщения, а издатель, который подключается сам, подключает сокет при первой публикации. Издатель на конечной точке, которую этот сервис привязывает для одной из своих подписок, подключается к слушателю этой подписки. В широковещательном брокере это XPUB, который один раз, при подключении, дожидается фильтра подписки, поэтому его первое сообщение не отбрасывается.

Каждый сокет подключается и каждая задача выполняется на том рантайме, на котором подключился брокер, какой бы рантайм ни сделал запрос. Обработчик в собственном потоке может публиковать, и сокет, который он подключил, переживает рантайм этого потока.

Отправка в очередь переживает отмену публикации. Пока отправка узлу ждёт, клиент убирает этот узел из очерёдности, поэтому отправка, отменённая в этот момент, потеряла бы узел навсегда. Издатель держит незавершённую отправку рядом со своим сокетом, и следующая публикация сначала завершает её, а потом выполняет свою. Крейт запрещает код unsafe, поэтому отправку, которая заимствует свой сокет, держат крейты selfie и aliasable. Сокет размещается в куче один раз на издателя, и публикация не добавляет выделений памяти к тем, что делает клиент.

Внутри процесса

Фича testing добавляет второй транспорт, который TestApp подключает вместо сокетов. Каждый брокер сохраняет свою подключённую форму, свой учёт и свои политики публикации; меняется только сокет под ними. Публикация собирается в кадры так же, как это делает издатель на сокете, а подписка читает кадры так же, как их читает её задача сокета.

Кто получит сообщение, решает правило паттерна. Подписка на очередь, которая привязывается, получает всё, что в неё отправлено, а подписки на очередь, которые подключаются, получают сообщения поочерёдно. Широковещательное сообщение доходит до каждой подписки, имя которой служит его префиксом. Запрос доходит до отвечающей стороны через идентичность узла, а ответ возвращается к тому запросу, который его задал.

Отказы, которые выдают сокеты, действуют и внутри процесса, с тем же текстом. Завершение доставки и там отвечает Unsupported.

Внутрипроцессный транспорт проверяет логику сервиса, а не транспорт. Насколько далеко вперёд читает подписка, обратное давление и настоящее рукопожатие проверяются на сокетах через loopback, в живом режиме того же теста. В рабочую сборку ничего из этого не входит: крейт проверяет размер доставки и каждого дескриптора во время компиляции.