Путь

Путь одной доставки AMQP 1.0 через сервис RustStream: задача приёма, канал, обработчик, ответ и диспозиция.

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

1. transfer на принимающей связи подписки → Задача приёма: Receiver::recv заголовки из секций свойств, байты из секции data
2. Задача приёма: Receiver::recv заголовки из секций свойств, байты из секции data → Ограниченный канал глубиной в кредит связи
3. Ограниченный канал глубиной в кредит связи → Диспетчеризация один цикл или партиция по ключу
4. Диспетчеризация один цикл или партиция по ключу → Декодирование из байтов доставки
5. Декодирование из байтов доставки → Обработчик подписчика async fn под #[subscriber]
6. Обработчик подписчика async fn под #[subscriber] → Ответ кодируется в буфер цикла
7. Ответ кодируется в буфер цикла → AmqpPublisher держит связь адреса до диспозиции брокера
8. AmqpPublisher держит связь адреса до диспозиции брокера → Урегулирование команда обратно задаче приёма
9. Урегулирование команда обратно задаче приёма → disposition accepted, modified или rejected

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

Задача приёма превращает передачу в AmqpMessage. Заголовки берутся из секций properties и application-properties, байты - из секции data, а delivery-count - из секции header. Секция значения со строкой или двоичными данными тоже читается как байты, а доставка с любым другим телом отклоняется, и подписка получает ошибку. Затем задача приёма кладёт доставку в канал глубиной в кредит связи и, пока ждёт в нём места, выполняет команды урегулирования.

Доставка, которую клиент не может прочитать, например доставка больше, чем допускает связь, передаётся подписке как ошибка, а связь остаётся присоединённой. Если брокер отправляет передачи сверх выданного ему кредита, связь разрывается, и подписка заканчивается.

Ответ обработчика кодируется в буфер, который цикл диспетчеризации использует повторно. Издатель AmqpPublisher берёт этот буфер как секцию data сообщения, без копирования. Он отправляет сообщение неурегулированным по отправляющей связи адреса и держит связь, пока не придёт диспозиция брокера. Поэтому публикации в один адрес с одного соединения уходят по одной, и каждая стоит одного кругового обмена. Диспозиция с исходом, отличным от accepted, - это ошибка. Если ответ не удалось опубликовать, доставка урегулируется повтором.

Доставка урегулируется после ответа. Её исход становится диспозицией, как показывает страница Урегулирование.