Устройство

Где стоит ruststream-amqp: между рантаймом RustStream и клиентом fe2o3-amqp, отдельный сеанс на каждую подписку и один общий сеанс для издателей.

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

1. Ваш сервис: #[subscriber] обработчики; RustStream::new(..) .with_broker(..)
2. Ядро RustStream: Рантайм: диспетчер, партиции, урегулирование
3. ruststream-amqp: AmqpSubscriber задача приёма, канал; AmqpPublisher, AmqpTxnPublisher; AmqpMessage; Запрос и ответ; AmqpBroker URL, Sasl, контейнер; AmqpAddress queue, topic, raw
4. fe2o3-amqp: одно соединение: Сеанс и получатель на каждую подписку; Один сеанс и по связи на каждый адрес
5. #[subscriber] обработчики → Рантайм: диспетчер, партиции, урегулирование: монтируются
6. Рантайм: диспетчер, партиции, урегулирование → AmqpSubscriber задача приёма, канал: опрашивает поток
7. AmqpSubscriber задача приёма, канал → Сеанс и получатель на каждую подписку
8. AmqpPublisher, AmqpTxnPublisher → Один сеанс и по связи на каждый адрес
9. Сеанс и получатель на каждую подписку → Брокер AMQP 1.0
10. Один сеанс и по связи на каждый адрес → Брокер AMQP 1.0

Крейт реализует трейты брокера из ядра поверх fe2o3-amqp. Рантайм над ним один для всех брокеров. Всё, что относится к AMQP 1.0, лежит в этом крейте: дескриптор адреса, смысл ack, адрес копии для повтора и то, как проводится транзакция.

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

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

Крейт сам открывает TCP-сокет и выключает на нём алгоритм Нейгла. Публикация и запрос отправляют по небольшому кадру и ждут ответа. Именно такой обмен алгоритм Нейгла задерживает до подтверждения от другой стороны.

У сообщения нет собственного конверта фреймворка. Стандартные заголовки хранятся в секции properties, ключ партиционирования - в поле group-id, а все остальные заголовки - в секции application-properties. Сообщение от другого клиента AMQP приходит со своими заголовками, а сообщение, которое отправил этот крейт, читается как обычное сообщение AMQP.

Фича asyncapi описывает адреса объектами x-ruststream-amqp1, потому что спецификация резервирует свои привязки amqp1 и оставляет их пустыми.

Как части запускаются и останавливаются, показывает страница Жизненный цикл, а путь одного сообщения - страница Путь.

Транспорт внутри процесса

С фичей testing вызов TestApp::start подключает тот же брокер к транспорту внутри процесса вместо сервера. Дескрипторы, политики и доставки - те же, что в рабочей сборке, и публикация проходит то же преобразование в секции AMQP. Транспорт маршрутизирует по точному адресу. Получатели очереди одного адреса конкурируют, а каждая подписка на тему получает свою копию. Подписка на очередь создаёт эту очередь, и очередь хранит сообщение, пока его не заберёт получатель. Транспорт урегулирует доставки в режиме подписки, учитывает доставку с исходом modified в delivery-count, возвращает неурегулированную доставку в очередь, когда её подписка отсоединяется, и отвечает на запрос через закрытый адрес ответа. Транзакция ничего не публикует до фиксации.

Поведение, которое принадлежит самому серверу, проверяет только прогон на живом брокере: кредит связи, создание адреса, до которого первой дошла публикация, политика недоставленных сообщений за отказом, собственный предел доставок сервера и долговечность. Внутри процесса сообщение в адрес, к которому не присоединилась ни одна подписка на очередь, отбрасывается, а адрес raw доставляет сообщение один раз там, где Artemis может разослать его всем подписчикам.

Без этой фичи транспорт не компилируется. Подключённый брокер рабочей сборки - один разделяемый указатель на соединение, и компилятор проверяет этот размер.