Struct Subscription 

pub struct Subscription { /* private fields */ }

Async-side subscription handle. Returned by crate::core::Subscribe::subscribe.

Implementations

impl Subscription

pub fn id(&self) -> u32

Group id most recently assigned by the server. Updated by Subscribe::resubscribe if the server hands out a fresh id.

pub fn name(&self) -> &str

Group alias (the name passed to subscribe). Fixed for the life of the subscription.

pub fn fdiv(&self) -> u32

Frequency divider the subscription was created with.

pub fn paths(&self) -> Vec<String>

Parameter paths the subscription was created against, in order. Extracted from the current server layout; on resubscribe the server may hand back fresh offsets, but paths themselves don’t change.

pub fn notify<F>(&self, cb: F)

where F: Fn(&Subscription) + Send + Sync + 'static,

Install a fire-and-forget callback invoked every time a new payload arrives. The callback runs on the subscribe receive thread — don’t block it.

subscription.notify(|s| {
    if let Some((_ts, value)) = s.read::<f64>() {
        println!("got {value}");
    }
});

pub fn read<V>(&self) -> Option<(TimeSpec, V)>

where V: GetParameterTuple,

Decode the most recent payload into a tuple matching the subscription’s parameter shape. Returns None if no payload has arrived yet.

pub fn read_all<V>(&self) -> Option<(TimeSpec, Vec<V>)>

where V: GetParameterValue + Default,

Decode the most recent payload into a flat Vec<V> — every scalar element of every subscribed parameter, in order.

pub async fn latest<V>(&self) -> Result<(TimeSpec, V)>

where V: GetParameterTuple,

Await the latest payload. Resolves immediately if any payload has already arrived; otherwise waits for the first one.

The underlying channel is lossy by design — if many payloads arrive between calls, you only see the most recent. For “give me every sample” semantics, see stream.

let (_ts, value): (_, f64) = subscription.latest().await?;
println!("latest = {value}");

Examples found in repository

examples/subscribe_latest.rs (line 39)

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

pub fn stream<V>( &self, capacity: usize, ) -> impl Stream<Item = StreamResult<V>> + use<V>

where V: GetParameterTuple + Send + 'static,

Subscribe to every payload via a bounded ring buffer.

use futures::StreamExt;
use motorcortex_rust::core::Missed;

let mut stream = Box::pin(subscription.stream::<f64>(256));
while let Some(item) = stream.next().await {
    match item {
        Ok((_ts, v))       => println!("{v}"),
        Err(Missed(n))     => eprintln!("dropped {n} samples"),
    }
}

The capacity is the number of in-flight samples the buffer can hold; if a consumer falls behind by more than that, the next item is Err(Missed(n)) with n = missed samples, and the consumer can decide whether to catch up or bail. This is explicit back-pressure, as opposed to latest which is lossy by design.

The broadcast channel is created lazily on the first call — subscriptions with only notify / read / latest users pay nothing for it. Subsequent calls reuse the existing broadcast; the capacity argument is honoured only on the first call.

Examples found in repository

examples/subscribe_stream.rs (line 38)

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 Subscription

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.