Lifecycle

LapinBroker::new records configuration, connect opens the connection, startup opens a channel and a consumer per subscription, and shutdown closes.

The diagram as text

1. Broker: LapinBroker::new (uri); ConnectedLapinBroker; ClosedLapinBroker handshake()
2. On RabbitMQ: Configuration only, no I/O; The connection and the publish channel; Each subscription: its channel, its topology if declared, basic.qos and basic.consume; The first confirms, transaction or request opens its channel; Marked closed, then connection.close
3. LapinBroker::new (uri) → ConnectedLapinBroker: connect(self)
4. ConnectedLapinBroker → ClosedLapinBroker handshake(): shutdown(self)
5. Configuration only, no I/O → The connection and the publish channel
6. The connection and the publish channel → Each subscription: its channel, its topology if declared, basic.qos and basic.consume
7. Each subscription: its channel, its topology if declared, basic.qos and basic.consume → The first confirms, transaction or request opens its channel
8. The first confirms, transaction or request opens its channel → Marked closed, then connection.close
9. Without declaration, a missing waiting queue or delayed exchange fails startup
10. A publisher that outlives the connection answers AmqpError::Closed, never a silent success

new does no I/O, so the service's builder stays synchronous. The runtime connects the broker when the app starts. Connect opens the AMQP connection to the URI, with the connection_name the management UI shows, and opens the shared publish channel. A server that refuses the connection fails the start.

Each subscription's descriptor resolves once, against the connected broker, and the start fails on the first one that cannot open:

1. A quorum queue checks its retry declaration. max_attempts and dead_letter become the queue's own arguments, so they need each other and a broker that declares the queue.
2. The subscription opens its channel.
3. Under declare_topology(true) it declares its exchanges, its queue, its bindings and its delay target. A declaration that differs from what the server holds is refused with PRECONDITION_FAILED.
4. It sends basic.qos with its prefetch window, when one is set.
5. A delay the broker does not declare is checked to exist: its waiting queue or its delayed exchange. A missing one fails the start and names both the subscription and the target.
6. It starts its consumer with basic.consume. A queue that does not exist fails here.

The delay check exists because the delayed copy is published without waiting for an answer, and the original is acknowledged after it. The server drops a copy sent to a waiting queue that does not exist, so without the check the message would be lost with no error. It costs one round trip at startup and nothing per message.

A service that only consumes opens no further channel. The first publish through a confirms or transaction policy, and the first request, open the channel each of them needs.

Shutdown marks the connection closed first, so a publisher that still holds it cannot slip a message in, and then closes the connection with the AMQP close handshake. The server requeues whatever the subscriptions held unsettled. ClosedLapinBroker::handshake reports whether the handshake ran. A publisher or a requester that outlives the connection answers AmqpError::Closed after shutdown, and a delivery settled after shutdown reports an error instead of a settlement that never happened.