Connecting to production

A production broker asks for TLS and credentials, and its operators look for a service by name.

TLS and credentials

src/main.rs

/// Where the broker is and who the service is, as the deployment provides them.
struct Endpoint {
    url: String,
    username: String,
    password: String,
}

impl Endpoint {
    fn from_env() -> Self {
        let read = |name: &str| env::var(name).unwrap_or_default();
        Self {
            url: read("AMQP_URL"),
            username: read("AMQP_USERNAME"),
            password: read("AMQP_PASSWORD"),
        }
    }
}

fn broker(endpoint: Endpoint) -> AmqpBroker {
    AmqpBroker::new(endpoint.url)
        .sasl(Sasl::plain(endpoint.username, endpoint.password))
        .container_id("orders-eu-1")
}

Cargo.toml

[dependencies]
ruststream-amqp = { version = "0.7", features = ["rustls"] }

The deployment provides the address and the credentials in environment variables. An amqps:// URL connects over TLS. It names the broker by a host name or an IPv4 address: TLS checks that name against the broker's certificate. An IPv6 address gives TLS no server name, so the connect fails on it. TLS needs one of two features: rustls uses rustls, and native-tls uses the platform's TLS.

Sasl::plain logs the service in with a user name and a password. The URL then carries no credentials. A password with characters that have a meaning in a URL, such as /, ? or @, goes through Sasl::plain as it is. Sasl::anonymous connects without credentials, and Sasl::external uses a credential the connection already established outside SASL.

On Azure Service Bus the user name is the name of a shared access policy and the password is its key, as its AMQP guide describes.

The service

Macros

src/main.rs

#[subscriber(AmqpAddress::queue("orders"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { id: order.id }
}

fn service(broker: AmqpBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm).out_reply(Publish);
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(broker(Endpoint::from_env()))
}

Manual

src/main.rs

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(order.id, "order confirmed");
        ready(Ok(Confirmation { id: order.id }))
    }
}

fn service(broker: AmqpBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(AmqpAddress::queue("orders"), Confirm)
                .reply()
                .build(),
        )
        .out_reply(Publish);
    })
}

fn app() -> impl App {
    service(broker(Endpoint::from_env()))
}

fn main() -> ExitCode {
    run_main(app)
}

container_id names the service to the broker, ruststream by default. The broker's console lists the connection under that name, and the links of the service are named after it. Give each deployment of a service its own container id, so an operator tells them apart.

Each subscription runs on an AMQP session of its own, and the publishers share one more. Flow control works per session, so a slow consumer does not hold back the publishers or another subscription.

Shutdown

When the service stops, it closes the connection from the inside out: the subscriptions' sessions, the publishers' links and their session, then the connection. A publisher used after that reports AmqpError::NotConnected instead of succeeding against a closed connection. The AsyncAPI document of the service names the broker's host and port and never the credentials.

Testing the configuration

The tests start the production service with the production broker, given test credentials, under TestApp::start. The in-process broker checks the URL the way the client reads it, without dialling the broker. A URL the client cannot open fails the start:

the scheme "https" is not an AMQP scheme

The tests need the same TLS feature as the service: without it, an amqps:// URL fails the start too. Authentication and the certificate check happen only on a real broker.