Connecting to production

A production cluster has several servers, asks the service who it is, and speaks TLS. Every one of these is a setting of the async-nats ConnectOptions, which the broker takes before anything is dialled.

Servers, credentials and 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"

The deployment provides the servers and two file paths in environment variables. NatsBroker::new takes one address or a comma-separated list, and the client picks one of them to connect to. with_options attaches the ConnectOptions and still dials nothing.

A .creds file holds a user JWT and its NKey seed, which is how Synadia Cloud and the accounts set up with nsc identify a service. The app reads the file and hands its contents to ConnectOptions::with_credentials, which parses them at once. ConnectOptions::with_credentials_file reads the file itself, but it is async, and the app is built in a synchronous function. A file that does not parse stops the service before it connects. A server set up with a user and a password takes ConnectOptions::with_user_and_password, with a token with_token, and with an NKey alone with_nkey.

add_root_certificates names the certificate authority that signed the servers' certificates, and require_tls refuses a connection without TLS. add_client_certificate adds the service's own certificate where the cluster asks for mutual TLS. The client reads certificate files when it connects. name shows the service in the server's monitoring, and connection_timeout bounds each attempt to connect.

The service

Macros

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

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 { 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 builds the broker from the environment and hands it to service, the same function the tests call. The runtime connects at startup. After that the client reconnects on its own when a server goes away and restores the subscriptions. The generated AsyncAPI document reports each server's host and port, never the credentials.

A client built outside the framework, such as one shared with other code of the process, becomes the connected broker with ConnectedNatsBroker::from_client. Only the plain NatsBroker fits the synchronous app function, so with_options is the way wherever it can express the setting.

Testing the configuration

The tests start the production service with the production broker, given test credentials, under TestApp::start. The in-process NATS parses the addresses as the client does and dials nothing, so the credentials and the certificate path never reach a server. A second test hands broker a .creds file that does not parse and expects the error. Authentication and TLS themselves happen only on a real cluster.