Struct async_nats::ConnectOptions

pub struct ConnectOptions { /* private fields */ }

Connect options. Used to connect with NATS when custom config is needed.

Examples

let mut options =
async_nats::ConnectOptions::new()
    .require_tls(true)
    .ping_interval(std::time::Duration::from_secs(10))
    .connect("demo.nats.io").await?;

Implementations

impl ConnectOptions

pub fn new() -> ConnectOptions

Enables customization of NATS connection.

Examples

let mut options =
async_nats::ConnectOptions::new()
    .require_tls(true)
    .ping_interval(std::time::Duration::from_secs(10))
    .connect("demo.nats.io").await?;

pub async fn connect<A: ToServerAddrs>(self, addrs: A) -> Result<Client>

Connect to the NATS Server leveraging all passed options.

Examples

let nc = async_nats::ConnectOptions::new().require_tls(true).connect("demo.nats.io").await?;

pub fn with_token(token: String) -> Self

Auth against NATS Server with provided token.

Examples

let nc =
async_nats::ConnectOptions::with_token("t0k3n!".into()).connect("demo.nats.io").await?;

pub fn with_user_and_password(user: String, pass: String) -> Self

Auth against NATS Server with provided username and password.

Examples

let nc = async_nats::ConnectOptions::with_user_and_password("derek".into(), "s3cr3t!".into())
    .connect("demo.nats.io").await?;

pub fn with_jwt<F, Fut>(jwt: String, sign_cb: F) -> Self where
    F: Fn(Vec<u8>) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = Result<Vec<u8>, AuthError>> + 'static + Send + Sync, 

Authenticate with a JWT. Requires function to sign the server nonce. The signing function is asynchronous

Example

let seed = "SUANQDPB2RUOE4ETUA26CNX7FUKE5ZZKFCQIIW63OX225F2CO7UEXTM7ZY";
let key_pair = std::sync::Arc::new(nkeys::KeyPair::from_seed(seed).unwrap());
// load jwt from creds file or other secure source
async fn load_jwt() -> std::io::Result<String> { todo!(); }
let jwt = load_jwt().await?;
let nc = async_nats::ConnectOptions::with_jwt(jwt,
     move |nonce| {
        let key_pair = key_pair.clone();
        async move { key_pair.sign(&nonce).map_err(async_nats::AuthError::new) }})
    .connect("localhost").await?;

pub async fn with_credentials_file(path: PathBuf) -> Result<Self>

Authenticate with NATS using a .creds file. Open the provided file, load its creds, and perform the desired authentication

Example

let nc = async_nats::ConnectOptions::with_credentials_file("path/to/my.creds".into()).await?
    .connect("connect.ngs.global").await?;

pub fn with_credentials(creds: &str) -> Result<Self>

Authenticate with NATS using a credential str, in the creds file format.

Example

let creds =
"-----BEGIN NATS USER JWT-----
eyJ0eXAiOiJqd3QiLCJhbGciOiJlZDI1NTE5...
------END NATS USER JWT------

************************* IMPORTANT *************************
NKEY Seed printed below can be used sign and prove identity.
NKEYs are sensitive and should be treated as secrets.

-----BEGIN USER NKEY SEED-----
SUAIO3FHUX5PNV2LQIIP7TZ3N4L7TX3W53MQGEIVYFIGA635OZCKEYHFLM
------END USER NKEY SEED------
";

let nc = async_nats::ConnectOptions::with_credentials(creds)
    .expect("failed to parse static creds")
    .connect("connect.ngs.global").await?;

pub fn add_root_certificates(self, path: PathBuf) -> ConnectOptions

Loads root certificates by providing the path to them.

Examples

let nc =
async_nats::ConnectOptions::new().add_root_certificates("mycerts.pem".into()).connect("demo.nats.io").await?;

pub fn add_client_certificate(self, cert: PathBuf, key: PathBuf) -> ConnectOptions

Loads client certificate by providing the path to it.

Examples

let nc =
async_nats::ConnectOptions::new().add_client_certificate("cert.pem".into(), "key.pem".into()).connect("demo.nats.io").await?;

pub fn require_tls(self, is_required: bool) -> ConnectOptions

Sets or disables TLS requirement. If TLS connection is impossible while options.require_tls(true) connection will return error.

Examples

let nc =
async_nats::ConnectOptions::new().require_tls(true).connect("demo.nats.io").await?;

pub fn flush_interval(self, flush_interval: Duration) -> ConnectOptions

Sets the interval for flushing. NATS connection will send buffered data to the NATS Server whenever buffer limit is reached, but it is also necessary to flush once in a while if client is sending rarely and small messages. Flush interval allows to modify that interval.

Examples

async_nats::ConnectOptions::new().flush_interval(Duration::from_millis(100)).connect("demo.nats.io").await?;

pub fn ping_interval(self, ping_interval: Duration) -> ConnectOptions

Sets how often Client sends PING message to the server.

Examples

async_nats::ConnectOptions::new().flush_interval(Duration::from_millis(100)).connect("demo.nats.io").await?;

pub fn no_echo(self) -> ConnectOptions

Sets no_echo option which disables delivering messages that were published from the same connection.

Examples

async_nats::ConnectOptions::new().no_echo().connect("demo.nats.io").await?;

pub fn subscription_capacity(self, capacity: usize) -> ConnectOptions

Sets the capacity for Subscribers. Exceeding it will trigger slow consumer error callback and drop messages. Defualt is set to 1024 messages buffer.

Examples

async_nats::ConnectOptions::new().subscription_capacity(1024).connect("demo.nats.io").await?;

pub fn error_callback<F, Fut>(self, cb: F) -> ConnectOptions where
    F: Fn(ServerError) -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ()> + 'static + Send + Sync, 

Registers asynchronous callback for errors that are receiver over the wire from the server.

Examples

As asynchronous callbacks are stil not in stable channel, here are some examples how to work around this

Basic

If you don’t need to move anything into the closure, simple signature can be used:

async_nats::ConnectOptions::new().error_callback(|error| async move {
println!("error occured: {}", error);
}).connect("demo.nats.io").await?;

Advanced

If you need to move something into the closure, here’s an example how to do that

let (tx, mut _rx) = tokio::sync::mpsc::channel(1);
async_nats::ConnectOptions::new().error_callback(move |error| {
    let tx = tx.clone();
    async move {
        tx.send(error).await.unwrap();
        }
}).connect("demo.nats.io").await?;

pub fn reconnect_callback<F, Fut>(self, cb: F) -> ConnectOptions where
    F: Fn() -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ()> + 'static + Send + Sync, 

Registers asynchronous callback for reconnection events.

Examples

As asynchronous callbacks are stil not in stable channel, here are some examples how to work around this

Basic

If you don’t need to move anything into the closure, simple signature can be used:

async_nats::ConnectOptions::new().reconnect_callback(|| async {
println!("reconnected");
}).connect("demo.nats.io").await?;

Advanced

If you need to move something into the closure, here’s an example how to do that

let (tx, mut _rx) = tokio::sync::mpsc::channel(1);
async_nats::ConnectOptions::new().reconnect_callback(move || {
    let tx = tx.clone();
    async move {
        tx.send("reconnected").await.unwrap();
        }
}).connect("demo.nats.io").await?;

pub fn disconnect_callback<F, Fut>(self, cb: F) -> ConnectOptions where
    F: Fn() -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ()> + 'static + Send + Sync, 

Registers asynchronous callback for disconection events.

Examples

As asynchronous callbacks are stil not in stable channel, here are some examples how to work around this

Basic

If you don’t need to move anything into the closure, simple signature can be used:

async_nats::ConnectOptions::new().disconnect_callback(|| async {
println!("disconnected");
}).connect("demo.nats.io").await?;

Advanced

If you need to move something into the closure, here’s an example how to do that

let (tx, mut _rx) = tokio::sync::mpsc::channel(1);
async_nats::ConnectOptions::new().disconnect_callback(move || {
    let tx = tx.clone();
    async move {
        tx.send("disconnected").await.unwrap();
        }
}).connect("demo.nats.io").await?;

pub fn lame_duck_callback<F, Fut>(self, cb: F) -> ConnectOptions where
    F: Fn() -> Fut + Send + Sync + 'static,
    Fut: Future<Output = ()> + 'static + Send + Sync, 

Registers asynchronous callback for server entering lame duck mode

Examples

As asynchronous callbacks are stil not in stable channel, here are some examples how to work around this

Basic

If you don’t need to move anything into the closure, simple signature can be used:

async_nats::ConnectOptions::new().lame_duck_callback(|| async {
println!("server entered lame duck mode");
}).connect("demo.nats.io").await?;

Advanced

If you need to move something into the closure, here’s an example how to do that

let (tx, mut _rx) = tokio::sync::mpsc::channel(1);
async_nats::ConnectOptions::new().reconnect_callback(move || {
    let tx = tx.clone();
    async move {
        tx.send("server entered lame duck mode").await.unwrap();
        }
}).connect("demo.nats.io").await?;

pub fn client_capacity(self, capacity: usize) -> ConnectOptions

By default, Client dispatches op’s to the Client onto the channel with capacity of 128. This option enables overriding it.

Examples

async_nats::ConnectOptions::new().client_capacity(256).connect("demo.nats.io").await?;

Trait Implementations

impl Debug for ConnectOptions

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl Default for ConnectOptions

fn default() -> ConnectOptions

Returns the “default value” for a type.