Struct Subscribe 

pub struct Subscribe { /* private fields */ }

Async handle for the pub/sub channel.

Cloning gives you another handle backed by the same driver, so every clone observes the same connection state.

Implementations

impl Subscribe

pub fn new() -> Self

Create a new handle and spawn its driver thread.

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

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

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

Examples found in repository

examples/subscribe_latest.rs (line 45)

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

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 connect_to(url: &str, opts: ConnectionOptions) -> Result<Self>

Convenience: Subscribe::new() + connect in one call.

Examples found in repository

examples/subscribe_latest.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.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 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-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 subscribe<I>( &self, req: &Request, paths: I, alias: &str, fdiv: u32, ) -> Result<Subscription>

where I: Parameters,

Create a subscription group on the server + wire it up locally so the receive thread starts pushing payloads into the returned Subscription handle.

The req handle is used only to issue the CreateGroupMsg RPC; it’s not held after this call returns.

let subscription = sub.subscribe(
    &req,
    ["root/Control/dummyDouble"],
    "my-group",
    10, // fdiv: every 10th cycle
).await?;
// Now use subscription.notify / latest / stream.

Examples found in repository

examples/subscribe_latest.rs (line 35)

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 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.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 resubscribe(&self, req: &Request) -> Result<()>

Re-register every active subscription after a server-side session restore.

Iterates the currently-live subscriptions; for each one, calls req.create_group(paths, alias, fdiv) to ask the server for a fresh group descriptor, then hands the (old_id, new_group) pairs to the driver which rebinds each Subscription in place. Outstanding Subscription clones remain valid and resume receiving payloads under the new id.

Typical use: after a ConnectionState::ConnectionLost → Connected transition where the server lost its group state (e.g. a process restart). When the session is merely restored via the token flow and groups persisted server-side, the create_group round-trip still succeeds and the rebind is a no-op on the data path.

use motorcortex_rust::ConnectionState;
let mut state = sub.state();
while state.changed().await.is_ok() {
    if *state.borrow() == ConnectionState::Connected {
        sub.resubscribe(&req).await?;
    }
}

pub async fn unsubscribe(&self, req: &Request, sub: Subscription) -> Result<()>

Drop a subscription locally + remove the group on the server. Safe to call with a stale Subscription — if the id isn’t in the active table the local side is a no-op, and the server-side remove_group returns its own StatusCode which we silently ignore here (a stale group is a cleanup success from the caller’s perspective). Transport / decode failures still propagate as Err.

Examples found in repository

examples/subscribe_latest.rs (line 44)

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

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 Subscribe

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 Subscribe

fn default() -> Self

Returns the “default value” for a type.