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

В промышленном кластере несколько серверов, он требует от сервиса назвать себя и работает через TLS. Всё это - настройки ConnectOptions из async-nats, которые брокер принимает до установки соединения.

Серверы, учётные данные и TLS

src/main.rs

/// Where the cluster is and who the service is, as the deployment provides them.
struct Deployment {
    /// One server address or a comma-separated list.
    servers: String,
    /// The contents of the service's `.creds` file: its user JWT and its `NKey` seed.
    creds: String,
    /// The certificate authority that signed the servers' certificates.
    ca: PathBuf,
}

impl Deployment {
    fn from_env() -> io::Result<Self> {
        let var = |name: &str| {
            env::var(name).map_err(|_| io::Error::new(io::ErrorKind::NotFound, name.to_owned()))
        };
        Ok(Self {
            servers: var("NATS_SERVERS")?,
            creds: fs::read_to_string(var("NATS_CREDS")?)?,
            ca: var("NATS_CA")?.into(),
        })
    }
}

fn broker(deployment: Deployment) -> io::Result<NatsBroker> {
    let options = ConnectOptions::with_credentials(&deployment.creds)?
        .add_root_certificates(deployment.ca)
        .require_tls(true)
        .name("orders")
        .connection_timeout(Duration::from_secs(10));
    Ok(NatsBroker::new(deployment.servers).with_options(options))
}

Cargo.toml

[dependencies]
async-nats = "0.50"

Серверы и пути к двум файлам развёртывание передаёт в переменных окружения. NatsBroker::new принимает один адрес или список через запятую, и клиент выбирает из них тот, к которому подключится. with_options добавляет ConnectOptions и по-прежнему никуда не подключается.

Файл .creds содержит JWT пользователя и seed его ключа NKey. Так сервис называет себя в Synadia Cloud и в учётных записях, настроенных через nsc. Приложение читает файл и передаёт его содержимое в ConnectOptions::with_credentials, который сразу его разбирает. ConnectOptions::with_credentials_file читает файл сам, но он асинхронный, а приложение строится в синхронной функции. Если файл не разбирается, сервис останавливается до подключения. Для сервера, настроенного с именем пользователя и паролем, есть ConnectOptions::with_user_and_password, с токеном - with_token, с одним только ключом NKey - with_nkey.

add_root_certificates называет центр сертификации, который подписал сертификаты серверов, а require_tls отклоняет соединение без TLS. add_client_certificate добавляет собственный сертификат сервиса, если кластер требует взаимной аутентификации TLS. Файлы сертификатов клиент читает при подключении. name показывает сервис в мониторинге сервера, а connection_timeout ограничивает каждую попытку подключения.

Сервис

Макросы

src/main.rs

#[subscriber(
    JetStreamSubject::new("orders.placed", "ORDERS").durable("confirmer"),
    reply
)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { order: order.id }
}

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

#[ruststream::app]
fn app() -> impl App {
    let broker = Deployment::from_env()
        .and_then(broker)
        .expect("NATS_SERVERS, NATS_CREDS and NATS_CA describe the connection");
    service(broker)
}

Вручную

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 { order: order.id }))
    }
}

fn service(broker: NatsBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(
                JetStreamSubject::new("orders.placed", "ORDERS").durable("confirmer"),
                Confirm,
            )
            .reply()
            .build(),
        );
    })
}

fn app() -> RustStream {
    let broker = Deployment::from_env()
        .and_then(broker)
        .expect("NATS_SERVERS, NATS_CREDS and NATS_CA describe the connection");
    service(broker)
}

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

app строит брокер из окружения и передаёт его в service, ту же функцию, которую вызывают тесты. Рантайм подключается при запуске. Дальше клиент сам переподключается, когда сервер пропадает, и восстанавливает подписки. Генерируемый документ AsyncAPI указывает хост и порт каждого сервера, но никогда не учётные данные.

Клиент, созданный вне фреймворка, например общий с другим кодом процесса, становится подключённым брокером через ConnectedNatsBroker::from_client. В синхронную функцию приложения подходит только обычный NatsBroker, поэтому везде, где настройку можно выразить через with_options, используйте его.

Тест конфигурации

Тесты запускают рабочий service с рабочим broker, которому переданы тестовые учётные данные, через TestApp::start. NATS внутри процесса разбирает адреса так же, как клиент, и никуда не подключается, поэтому учётные данные и путь к сертификату не доходят ни до какого сервера. Второй тест передаёт broker файл .creds, который не разбирается, и ожидает ошибку. Сами аутентификация и TLS работают только на настоящем кластере.