Struct Request 

pub struct Request { /* private fields */ }

Async handle for request/reply RPCs.

Cloning a Request gives a second handle that multiplexes onto the same driver thread — commands serialise through the driver, so the NNG Req/Rep ordering invariant is enforced in the type system without any user-visible Mutex.

Implementations

impl Request

pub fn new() -> Self

Create a new handle and spawn its driver thread.

The handle starts in ConnectionState::Disconnected; call connect to open the socket.

use motorcortex_rust::core::{ConnectionState, Request};
let req = Request::new();
assert_eq!(*req.state().borrow(), ConnectionState::Disconnected);

pub fn state(&self) -> Receiver<ConnectionState>

Subscribe to connection-state transitions.

Returns a watch::Receiver; consumers can state.changed().await or *state.borrow() for the current value.

pub async fn connect(&self, url: &str, opts: ConnectionOptions) -> Result<()>

Open the socket and dial url.

use motorcortex_rust::{ConnectionOptions, core::Request};
let req = Request::new();
let opts = ConnectionOptions::new("mcx.cert.crt".into(), 1000, 1000);
req.connect("wss://127.0.0.1:5568", opts).await?;

pub async fn disconnect(&self) -> Result<()>

Close the socket. Subsequent RPCs will error with MotorcortexError::Connection until connect is called again.

Examples found in repository

examples/async_request.rs (line 34)

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

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

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 async fn login(&self, user: &str, pass: &str) -> Result<StatusCode>

Authenticate with the server.

Returns the server’s StatusCode — Ok on success, WrongPassword / other variants on rejection. Transport-level failures surface as MotorcortexError.

Examples found in repository

examples/async_request.rs (line 28)

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

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

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 async fn logout(&self) -> Result<StatusCode>

Drop the current session on the server.

pub async fn request_parameter_tree(&self) -> Result<StatusCode>

Fetch the parameter tree from the server and store it in the shared cache accessible via parameter_tree.

Returns the server’s StatusCode. The cache is only updated on StatusCode::Ok; a non-OK reply leaves the previous cache intact.

Examples found in repository

examples/async_request.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).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 33)

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

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 parameter_tree(&self) -> Arc<RwLock<ParameterTree>>

Shared read handle to the local parameter-tree cache. Every cloned Request sees the same tree; reads are RwLock::read (cheap, no channel round-trip). Populated by request_parameter_tree.

pub async fn get_parameter<V>(&self, path: &str) -> Result<V>

where V: GetParameterValue + Default,

Read a single parameter. The caller-specified V is the Rust type the server value should be converted to (see GetParameterValue).

Returns MotorcortexError::ParameterNotFound if path is unknown locally — call request_parameter_tree first.

// Same path, three different Rust types — the server value is
// converted per call (lossy casts allowed).
let as_double: f64    = req.get_parameter("root/Control/dummyDouble").await?;
let as_int:    i64    = req.get_parameter("root/Control/dummyDouble").await?;
let as_text:   String = req.get_parameter("root/Control/dummyDouble").await?;

Examples found in repository

examples/async_request.rs (line 31)

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

pub async fn set_parameter<V>(&self, path: &str, value: V) -> Result<StatusCode>

where V: SetParameterValue,

Write a single parameter. Returns the server’s StatusCode.

// Scalar.
req.set_parameter("root/Control/dummyDouble", 2.345_f64).await?;
// Fixed-size array.
req.set_parameter("root/Control/dummyDoubleVec", [1.0, 2.0, 3.0]).await?;
// Dynamic Vec.
req.set_parameter("root/Control/dummyDoubleVec", vec![1.0, 2.0]).await?;

pub async fn get_parameters<T>(&self, paths: &[&str]) -> Result<T>

where T: GetParameterTuple,

Read a batch of parameters in one RPC. The generic T is a tuple type like (bool, f64, i32) whose arity matches paths.len(). Each element is decoded into its position using the dtype recorded in the local tree cache.

let (b, d, i): (bool, f64, i32) = req.get_parameters(&[
    "root/Control/dummyBool",
    "root/Control/dummyDouble",
    "root/Control/dummyInt32",
]).await?;

pub async fn set_parameters<T>( &self, paths: &[&str], values: T, ) -> Result<StatusCode>

where T: SetParameterTuple,

Write a batch of parameters in one RPC. values is a tuple (or single-element tuple) whose arity matches paths.len(). Each element is encoded against the dtype recorded in the local tree cache.

pub async fn create_group<I>( &self, paths: I, alias: &str, frequency_divider: u32, ) -> Result<GroupStatusMsg>

where I: Parameters,

Create a server-side subscription group. Returns the GroupStatusMsg the subscribe-side code uses as the group descriptor.

paths accepts anything implementing Parameters — a string literal, a Vec<String>, an array of &str, etc.

pub async fn remove_group(&self, alias: &str) -> Result<StatusCode>

Remove a previously-created subscription group by alias. Returns the server’s StatusCode unchanged — Ok on success, Failed (or similar) if the group wasn’t there. Transport / decode failures surface as MotorcortexError.

The crate-wide rule: RPCs that return Result<StatusCode> never promote a non-OK server reply into Err. Branch on the returned code if you care which way the request went.

pub async fn get_session_token(&self) -> Result<String>

Fetch a fresh session token from the server.

On success the token is cached in the driver and returned to the caller. Callers that want to persist a session across process restarts can stash the returned string and hand it back to a fresh Request via restore_session.

While a connection is live, the driver also refreshes this token periodically in the background (see ConnectionOptions::token_refresh_interval) so the cache stays warm for the automatic reconnect path.

pub async fn restore_session(&self, token: &str) -> Result<StatusCode>

Restore a previously-issued session by supplying the token. Returns the server’s StatusCode:

• Ok / ReadOnlyMode — the session was accepted, subsequent RPCs run under that identity.
• PermissionDenied / Failed — the token is stale or the server has lost its state.

The driver calls this internally on reconnect using the token stashed by the refresh loop; callers generally don’t need to invoke it explicitly unless they’re recovering from a process restart.

pub fn session_token(&self) -> Option<String>

Snapshot of the most recently cached session token, if any.

Populated by get_session_token and by the periodic refresh loop. Returns None before the first successful fetch.

pub fn session_refresh_count(&self) -> u64

Count of GetSessionToken RPCs the background refresh helper has fired since this handle was created.

Bumped by the driver only when the refresh tick runs against a live pipe — ticks that fire while the state is ConnectionLost / SessionExpired / Disconnected are skipped and don’t count. Useful for observing that the refresh loop is paused while the transport is down, or as a lightweight liveness metric.

pub async fn get_parameter_tree_hash(&self) -> Result<u32>

Fetch the server’s parameter-tree hash — useful for cheap change detection. A non-zero return on a populated server signals the caller can skip a full request_parameter_tree if the hash matches what they cached previously.

pub async fn connect_to(url: &str, opts: ConnectionOptions) -> Result<Self>

Convenience: Request::new() + connect in one call. Useful for the “I just want a connected client” entry path.

use motorcortex_rust::{ConnectionOptions, core::Request};
let opts = ConnectionOptions::new("mcx.cert.crt".into(), 1000, 1000);
let req = Request::connect_to("wss://127.0.0.1:5568", opts).await?;
req.request_parameter_tree().await?;

Examples found in repository

examples/async_request.rs (line 27)

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

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

Trait Implementations

impl Clone for Request

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Default for Request

fn default() -> Self

Returns the “default value” for a type.