motorcortex-rust 0.5.0

Motorcortex Rust: a Rust client for the Motorcortex Core real-time control system (async + blocking).
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

//! Subscribe and consume *every* sample via
//! [`Subscription::stream`] — lossless within the ring capacity,
//! explicit `Err(Missed(n))` when a consumer falls behind.
//!
//! ```bash
//! cargo run --example subscribe_stream -- \
//!     wss://127.0.0.1:5568:5567 path/to/mcx.cert.crt root mypassword root/Control/dummyDouble
//! ```

use std::env;
use std::time::Duration;

use futures::StreamExt;
use motorcortex_rust::core::{Missed, Request, Subscribe};
use motorcortex_rust::{ConnectionOptions, Result, parse_url};

#[tokio::main]
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(())
}