Startup

Startup opens the socket, authenticates and begins the publishers' session; a refusal fails the start as AmqpError::Connect.

The runtime connects the broker once, before any handler runs. The connect opens a TCP socket to the URL's host with Nagle's algorithm off: AMQP sends a small frame and waits for the answer, which that algorithm would hold back. Over the socket it runs TLS for an amqps:// URL, the SASL login, and the AMQP open with the service's container id. It then begins one session that every publisher of the service shares. A failure of the socket, TLS, the login or the open fails startup with AmqpError::Connect, and a session the broker refuses with AmqpError::Session. The service exits before it reads anything.

Each subscription then begins a session of its own and attaches a receiver link to its address. AMQP counts flow control per session, so a slow subscription holds back neither the publishers nor another subscription. An attach the broker refuses fails startup with AmqpError::Attach, which names the address. An empty address fails with AmqpError::InvalidAddress before any I/O. A publisher attaches its sender link to an address on the first publish there, and every later publish to the address reuses that link.

A TestApp test connects in process and reads the URL as the connect reads it. A scheme other than amqp or amqps, a URL without a host, or amqps without a TLS feature fails startup in a test as in production. A refused socket or login happens only against a broker, in a live test.

Example

The deployment names the broker in AMQP_URL, credentials included.

Macros
Manual

examples/amqp_broker_startup.rs
/// The service on the broker at `url`.
fn service(url: &str) -> impl App<State = ()> + use<> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(AmqpBroker::new(url), |b| {
        b.include(place);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    // The deployment names the broker, credentials included.
    let url = env::var("AMQP_URL").unwrap_or_else(|_| "amqp://localhost:5672".to_owned());
    service(&url)
}

examples/manual/amqp_broker_startup.rs
/// The service on the broker at `url`.
fn service(url: &str) -> impl App<State = ()> + use<> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(AmqpBroker::new(url), |b| {
        let place = subscriber(AmqpAddress::queue("orders"), Place);
        b.include(place.build());
    })
}

fn app() -> impl App<State = ()> {
    // The deployment names the broker, credentials included.
    let url = env::var("AMQP_URL").unwrap_or_else(|_| "amqp://localhost:5672".to_owned());
    service(&url)
}

Example​

Example