Struct ConnectionOptions 

pub struct ConnectionOptions {
    pub certificate: String,
    pub conn_timeout_ms: u32,
    pub io_timeout_ms: u32,
    pub reconnect: bool,
    pub reconnect_min: Duration,
    pub reconnect_max: Duration,
    pub token_refresh_interval: Duration,
    pub max_reconnect_attempts: Option<u32>,
}

Options used to configure the connection settings.

Fields

certificate: String

Path to the TLS certificate used for secure communication.

conn_timeout_ms: u32

Connection timeout in milliseconds.

io_timeout_ms: u32

I/O timeout in milliseconds.

reconnect: bool

Whether NNG should auto-redial the transport on drop. When true (default), NNG_OPT_RECONNMINT / NNG_OPT_RECONNMAXT are set on the dialer and the driver’s session-restore logic kicks in on the next pipe ADD event. When false, dropped transports stay Disconnected and the user must call connect again explicitly.

reconnect_min: Duration

Min backoff for NNG’s dialer redial (NNG_OPT_RECONNMINT). Ignored when reconnect == false.

reconnect_max: Duration

Max backoff ceiling for NNG’s dialer redial (NNG_OPT_RECONNMAXT). Ignored when reconnect == false.

token_refresh_interval: Duration

How often to refresh the session token on a live connection. Each refresh is a GetSessionToken RPC; the fresh token is stashed in the driver for restore_session to use after a drop. Default 30 s matches motorcortex-python. Set to Duration::ZERO to disable.

max_reconnect_attempts: Option<u32>

Safety net for the automatic reconnect path. When Some(n), the driver counts consecutive ADD_POST → RestoreSession(token) → non-Ok cycles and, on the n-th failure, disables NNG’s dialer redial and publishes ConnectionState::Disconnected. A successful restore (or a reconnect that skipped restore because no token was cached) resets the counter. None (default) means the driver keeps trying indefinitely, matching NNG’s built-in behaviour.

Implementations

impl ConnectionOptions

pub fn new( certificate: String, conn_timeout_ms: u32, io_timeout_ms: u32, ) -> Self

Creates a new ConnectionOptions instance with reconnect / token-refresh defaults. Use the with_* setters to override the reconnect knobs.

Arguments:

• certificate - Path to the TLS certificate.
• conn_timeout_ms - Timeout value for connection establishment, in milliseconds.
• io_timeout_ms - Timeout value for I/O operations, in milliseconds.

Examples found in repository

examples/blocking_request.rs (line 26)

fn main() -> Result<()> {
    let mut args = env::args().skip(1);
    let url = args.next().unwrap_or_else(|| "wss://127.0.0.1:5568:5567".into());
    let cert = args.next().unwrap_or_else(|| "tests/mcx.cert.crt".into());
    let user = args.next().unwrap_or_else(|| "root".into());
    let pass = args.next().unwrap_or_default();
    let path = args
        .next()
        .unwrap_or_else(|| "root/Control/dummyDouble".into());

    let (req_url, _sub_url) = parse_url(&url)?;
    let opts = ConnectionOptions::new(cert, 5_000, 5_000);
    let req = Request::connect_to(&req_url, opts)?;
    req.login(&user, &pass)?;
    req.request_parameter_tree()?;

    let value: f64 = req.get_parameter(&path)?;
    println!("{path} = {value}");

    req.disconnect()?;
    Ok(())
}

examples/async_request.rs (line 26)

async fn main() -> Result<()> {
    let mut args = env::args().skip(1);
    let url = args.next().unwrap_or_else(|| "wss://127.0.0.1:5568:5567".into());
    let cert = args.next().unwrap_or_else(|| "tests/mcx.cert.crt".into());
    let user = args.next().unwrap_or_else(|| "root".into());
    let pass = args.next().unwrap_or_default();
    let path = args
        .next()
        .unwrap_or_else(|| "root/Control/dummyDouble".into());

    let (req_url, _sub_url) = parse_url(&url)?;
    let opts = ConnectionOptions::new(cert, 5_000, 5_000);
    let req = Request::connect_to(&req_url, opts).await?;
    req.login(&user, &pass).await?;
    req.request_parameter_tree().await?;

    let value: f64 = req.get_parameter(&path).await?;
    println!("{path} = {value}");

    req.disconnect().await?;
    Ok(())
}

examples/subscribe_latest.rs (line 30)

async fn main() -> Result<()> {
    let mut args = env::args().skip(1);
    let url = args.next().unwrap_or_else(|| "wss://127.0.0.1:5568:5567".into());
    let cert = args.next().unwrap_or_else(|| "tests/mcx.cert.crt".into());
    let user = args.next().unwrap_or_else(|| "root".into());
    let pass = args.next().unwrap_or_default();
    let path = args
        .next()
        .unwrap_or_else(|| "root/Control/dummyDouble".into());

    let (req_url, sub_url) = parse_url(&url)?;
    let opts = ConnectionOptions::new(cert, 5_000, 5_000);
    let req = Request::connect_to(&req_url, opts.clone()).await?;
    req.login(&user, &pass).await?;
    req.request_parameter_tree().await?;
    let sub = Subscribe::connect_to(&sub_url, opts).await?;
    let subscription = sub.subscribe(&req, [&path[..]], "example-latest", 100).await?;

    // Sample the latest value every second for 5 seconds.
    for _ in 0..5 {
        let (_ts, value): (_, f64) = subscription.latest().await?;
        println!("{path} = {value}");
        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    sub.unsubscribe(&req, subscription).await?;
    sub.disconnect().await?;
    req.disconnect().await?;
    Ok(())
}

examples/subscribe_stream.rs (line 29)

async fn main() -> Result<()> {
    let mut args = env::args().skip(1);
    let url = args.next().unwrap_or_else(|| "wss://127.0.0.1:5568:5567".into());
    let cert = args.next().unwrap_or_else(|| "tests/mcx.cert.crt".into());
    let user = args.next().unwrap_or_else(|| "root".into());
    let pass = args.next().unwrap_or_default();
    let path = args
        .next()
        .unwrap_or_else(|| "root/Control/dummyDouble".into());

    let (req_url, sub_url) = parse_url(&url)?;
    let opts = ConnectionOptions::new(cert, 5_000, 5_000);
    let req = Request::connect_to(&req_url, opts.clone()).await?;
    req.login(&user, &pass).await?;
    req.request_parameter_tree().await?;
    let sub = Subscribe::connect_to(&sub_url, opts).await?;
    let subscription = sub.subscribe(&req, [&path[..]], "example-stream", 10).await?;

    // Read 20 samples from the ring. Ring size 256 means we tolerate
    // a ~256-sample consumer stall before seeing Missed.
    let mut stream = Box::pin(subscription.stream::<f64>(256));
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    let mut seen = 0;
    while seen < 20 {
        match tokio::time::timeout_at(deadline, stream.next()).await {
            Ok(Some(Ok((_ts, v)))) => {
                println!("{path} = {v}");
                seen += 1;
            }
            Ok(Some(Err(Missed(n)))) => eprintln!("lagged, dropped {n} samples"),
            Ok(None) | Err(_) => break,
        }
    }

    sub.unsubscribe(&req, subscription).await?;
    sub.disconnect().await?;
    req.disconnect().await?;
    Ok(())
}

pub fn with_reconnect(self, enabled: bool) -> Self

Disable NNG auto-redial + driver session-restore logic.

pub fn with_reconnect_backoff(self, min: Duration, max: Duration) -> Self

Override NNG’s dialer backoff window.

pub fn with_token_refresh_interval(self, interval: Duration) -> Self

Override the session-token refresh cadence. Duration::ZERO disables the background refresh entirely.

pub fn with_max_reconnect_attempts(self, attempts: Option<u32>) -> Self

Cap the number of consecutive RestoreSession failures the driver tolerates before it disables NNG’s dialer and publishes ConnectionState::Disconnected. Pass None to keep the default “retry forever” behaviour.

Trait Implementations

impl Clone for ConnectionOptions

fn clone(&self) -> ConnectionOptions

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.