Startup

At startup the runtime dials the servers within connection_timeout, then opens each subscription; a JetStream one finds its stream or fails the start.

The connect is one dial of the async-nats client. It covers the TCP connection, the TLS handshake and the login, and waits up to the connection_timeout of the options, five seconds by default. A server that refuses the connection or does not answer in time fails startup with NatsError::Connect, and the service exits before it reads anything.

A Core NATS subscription is open once the server has confirmed it. The broker sends the subscription and waits for one round trip to the server, so everything published after startup reaches it.

A JetStream subscription looks up the stream its JetStreamSubject names and creates its consumer there. A durable consumer that already exists takes the descriptor's settings. The stream itself belongs to the server: it is created before the service starts, with nats stream add or through jetstream. A stream the server lacks fails startup with NatsError::JetStream.

In process, a TestApp test finds every stream a subscription names, so the missing stream and the connect timing are for a live test.

Example

The connect gives up after three seconds. The tests in the same file start the service against a server: once before its stream exists, and once after.

Macros

examples/nats_broker_startup.rs

/// The servers at `addrs`; the connect gives up after three seconds.
fn broker(addrs: &str) -> NatsBroker {
    NatsBroker::new(addrs)
        .with_options(ConnectOptions::new().connection_timeout(Duration::from_secs(3)))
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker("nats://localhost:4222"))
}

Manual

examples/manual/nats_broker_startup.rs

/// The servers at `addrs`; the connect gives up after three seconds.
fn broker(addrs: &str) -> NatsBroker {
    NatsBroker::new(addrs)
        .with_options(ConnectOptions::new().connection_timeout(Duration::from_secs(3)))
}

fn app() -> impl App<State = ()> {
    service(broker("nats://localhost:4222"))
}