Подключение к промышленному брокеру

Промышленный брокер требует TLS и учётных данных, а его администраторы ищут сервис по имени.

TLS и учётные данные

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"] }

Адрес и учётные данные развёртывание передаёт в переменных окружения. С URL amqps:// сервис подключается по TLS. URL называет брокер по имени хоста или по адресу IPv4: TLS сверяет это имя с сертификатом брокера. Адрес IPv6 не даёт TLS имени сервера, поэтому подключение по нему завершается ошибкой. TLS требует одной из двух фич: rustls использует rustls, а native-tls - TLS платформы.

Sasl::plain выполняет вход сервиса с именем пользователя и паролем. Тогда в URL нет учётных данных. Пароль с символами, которые имеют особый смысл в URL, например /, ? или @, передаётся в Sasl::plain как есть. Sasl::anonymous подключается без учётных данных, а Sasl::external использует учётные данные, которые соединение уже установило вне SASL.

В Azure Service Bus имя пользователя - это имя политики общего доступа, а пароль - её ключ, как описывает руководство по AMQP.

Сервис

Макросы

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()))
}

Вручную

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 называет сервис брокеру, по умолчанию ruststream. Консоль брокера показывает соединение под этим именем, и связи сервиса названы по нему. Дайте каждому развёртыванию сервиса собственный идентификатор контейнера, чтобы администратор их различал.

Каждая подписка работает в собственном сеансе AMQP, а издатели делят ещё один. Управление потоком действует на уровне сеанса, поэтому медленный потребитель не сдерживает издателей и другие подписки.

Остановка

При остановке сервис закрывает соединение изнутри наружу: сначала сеансы подписок, затем связи издателей и их сеанс, затем само соединение. Издатель, которого используют после этого, возвращает ошибку AmqpError::NotConnected, а не выполняет публикацию через закрытое соединение. Документ AsyncAPI сервиса называет хост и порт брокера и никогда не содержит учётных данных.

Проверка конфигурации

Тесты запускают рабочий service с рабочим broker, которому переданы тестовые учётные данные, через TestApp::start. Брокер внутри процесса проверяет URL так же, как его читает клиент, без подключения к брокеру. URL, который клиент не может открыть, не даёт запуститься:

the scheme "https" is not an AMQP scheme

Тестам нужна та же фича TLS, что и сервису: без неё URL amqps:// тоже не даёт запуститься. Аутентификация и проверка сертификата выполняются только на настоящем брокере.