Жизненный цикл

LapinBroker::new записывает конфигурацию, connect открывает соединение, запуск открывает канал и потребителя для каждой подписки, а завершение работы закрывает соединение.

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

1. Брокер: LapinBroker::new (uri); ConnectedLapinBroker; ClosedLapinBroker handshake()
2. В RabbitMQ: Только конфигурация, без ввода-вывода; Соединение и канал публикации; Каждая подписка: свой канал, объявленная топология, basic.qos и basic.consume; Первое подтверждение, транзакция или запрос открывает свой канал; Отметка о закрытии, затем connection.close
3. LapinBroker::new (uri) → ConnectedLapinBroker: connect(self)
4. ConnectedLapinBroker → ClosedLapinBroker handshake(): shutdown(self)
5. Только конфигурация, без ввода-вывода → Соединение и канал публикации
6. Соединение и канал публикации → Каждая подписка: свой канал, объявленная топология, basic.qos и basic.consume
7. Каждая подписка: свой канал, объявленная топология, basic.qos и basic.consume → Первое подтверждение, транзакция или запрос открывает свой канал
8. Первое подтверждение, транзакция или запрос открывает свой канал → Отметка о закрытии, затем connection.close
9. Без объявления отсутствие очереди ожидания или отложенного обменника прерывает запуск
10. Издатель, переживший соединение, отвечает AmqpError::Closed, а не молчаливым успехом

new не выполняет ввода-вывода, поэтому билдер сервиса остаётся синхронным. Рантайм подключает брокер при запуске приложения. Подключение открывает соединение AMQP по URI с именем connection_name, которое показывает интерфейс управления, и открывает общий канал публикации. Если сервер отказывает в соединении, запуск прерывается.

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

1. Кворумная очередь проверяет объявление переобработки. max_attempts и dead_letter становятся аргументами самой очереди, поэтому задаются только вместе и только на брокере, который объявляет очередь.
2. Подписка открывает свой канал.
3. При declare_topology(true) она объявляет свои обменники, свою очередь, свои привязки и цель задержки. Объявление, которое расходится с тем, что хранит сервер, отклоняется с PRECONDITION_FAILED.
4. Подписка отправляет basic.qos с окном предвыборки, если оно задано.
5. Для delay, который брокер не объявляет, проверяется, что цель существует: очередь ожидания или отложенный обменник. Если цели нет, запуск прерывается с ошибкой, которая называет и подписку, и цель.
6. Подписка запускает потребителя командой basic.consume. Несуществующая очередь прерывает запуск на этом шаге.

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

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

Завершение работы сначала помечает соединение закрытым, чтобы издатель, который ещё держит его, не успел отправить сообщение, а затем закрывает соединение штатным закрытием AMQP. Сервер возвращает в очередь всё, что подписки держали неподтверждённым. ClosedLapinBroker::handshake сообщает, состоялось ли штатное закрытие. Издатель или запрашивающая сторона, пережившие соединение, после завершения работы отвечают AmqpError::Closed, а подтверждение доставки после завершения работы возвращает ошибку вместо подтверждения, которого не было.