连接生产环境

生产集群有多台服务器，会要求服务表明身份，并且使用 TLS 通信。这些都是 async-nats ConnectOptions 的设置，Broker 在发起任何连接之前接收它。

服务器、凭据与 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 及其 NKey 种子，Synadia Cloud 和用 nsc 配置的账户就是这样识别服务的。应用读取这个文件，把内容交给 ConnectOptions::with_credentials，后者立即解析它。 ConnectOptions::with_credentials_file 会自己读取文件，但它是 async 的，而应用是在同步函数中构建的。无法解析的文件会让服务在连接之前就停止。服务器配置了用户名和密码时，使用 ConnectOptions::with_user_and_password；配置了令牌时使用 with_token；只配置了 NKey 时使用 with_nkey。

add_root_certificates 指定为服务器证书签名的证书颁发机构，require_tls 拒绝不使用 TLS 的连接。集群要求双向 TLS 时，add_client_certificate 加入服务自己的证书。客户端在连接时读取证书文件。 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 根据环境构建 Broker，并把它交给 service，测试调用的也是这个函数。运行时在启动时建立连接。此后，服务器下线时客户端会自行重连，并恢复各个订阅。生成的 AsyncAPI 文档报告每台服务器的主机和端口，从不包含凭据。

在框架之外构建的客户端，比如与进程中其他代码共用的客户端，可以用 ConnectedNatsBroker::from_client 变成已连接的 Broker。只有普通的 NatsBroker 适用于同步的应用函数，所以只要 with_options 能表达某项设置，就用它。

测试配置

测试在 TestApp::start 下启动生产环境的 service，配上用测试凭据构建的生产环境 broker。进程内 NATS 像客户端一样解析地址，但不发起任何连接，所以凭据和证书路径永远不会到达服务器。第二个测试把一个无法解析的 .creds 文件交给 broker，并预期得到错误。认证和 TLS 本身只在真实集群上进行。

服务器、凭据与 TLS​

服务​

测试配置​

服务器、凭据与 TLS

服务

测试配置