mxaccess/rust/crates/mxaccess/examples/subscribe-buffered.rs

//! `subscribe-buffered` — open a buffered subscription with a 1-second cadence.
//!
//! Per `wwtools/mxaccesscli/docs/api-notes.md:97-100,138-140,154-157` (R2 in
//! `design/70-risks-and-open-questions.md`), the `update_interval_ms` knob
//! controls the **delivery cadence** — each emitted event still carries one
//! sample, **not** a multi-sample payload. The returned [`mxaccess::Subscription`]
//! is the same `Stream<Item = Result<DataChange, Error>>` as plain
//! [`mxaccess::Session::subscribe`].
//!
//! Drains up to 5 updates (or a 30s timeout, whichever first), prints each,
//! then unsubscribes cleanly. Mirrors the `subscribe.rs` shape — see that
//! example for the env-var contract and resolver shim design notes.

use std::sync::Arc;
use std::time::Duration;

use futures_util::StreamExt;
use mxaccess::{
    BufferedOptions, GalaxyTagMetadata, RecoveryPolicy, Resolver, ResolverError, Session,
    SessionOptions,
};
use mxaccess_rpc::guid::Guid;
use mxaccess_rpc::ntlm::NtlmClientContext;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let Some(env) = LiveEnv::from_process()? else {
        eprintln!(
            "MX_LIVE not set — skipping live demo. Run \
             `. tools/Setup-LiveProbeEnv.ps1` to populate the required env vars."
        );
        return Ok(());
    };

    let session = Session::connect_nmx(
        env.addr,
        SessionOptions::default(),
        NtlmClientContext::from_env()?,
        env.service_ipid,
        Arc::new(StaticResolver::new(&env.tag)),
        RecoveryPolicy::default(),
    )
    .await?;

    let opts = BufferedOptions {
        update_interval_ms: 1_000,
    };
    eprintln!(
        "buffered-subscribing to {} (requested cadence {} ms, rounded to {} ms)",
        env.tag,
        opts.update_interval_ms,
        opts.rounded_update_interval_ms()
    );
    let mut sub = session.subscribe_buffered(&env.tag, opts).await?;
    eprintln!("correlation_id = {:02x?}", sub.correlation_id());

    let mut received = 0;
    while received < 5 {
        match tokio::time::timeout(Duration::from_secs(30), sub.next()).await {
            Ok(Some(Ok(dc))) => {
                println!("{} = {:?} ts={:?}", dc.reference, dc.value, dc.timestamp);
                received += 1;
            }
            Ok(Some(Err(e))) => {
                eprintln!("subscription error: {e}");
                break;
            }
            Ok(None) => {
                eprintln!("subscription stream ended");
                break;
            }
            Err(_) => {
                eprintln!("no update within 30s; exiting after {received} updates");
                break;
            }
        }
    }

    session.unsubscribe(sub).await?;
    session.shutdown_nmx().await?;
    Ok(())
}

// ---- shared boilerplate (see subscribe.rs / connect-write-read.rs for rationale) ----

struct LiveEnv {
    addr: std::net::SocketAddr,
    service_ipid: Guid,
    tag: String,
}

impl LiveEnv {
    fn from_process() -> Result<Option<Self>, Box<dyn std::error::Error>> {
        if std::env::var_os("MX_LIVE").is_none() {
            return Ok(None);
        }
        let host = std::env::var("MX_NMX_HOST")?;
        let addr = parse_host_port(&host, 135)?;
        let service_ipid = Guid::parse_str(&std::env::var("MX_NMX_SERVICE_IPID")?)?;
        let tag = std::env::var("MX_TEST_TAG").unwrap_or_else(|_| "TestChildObject.TestInt".into());
        Ok(Some(Self {
            addr,
            service_ipid,
            tag,
        }))
    }
}

fn parse_host_port(
    s: &str,
    default_port: u16,
) -> Result<std::net::SocketAddr, Box<dyn std::error::Error>> {
    if let Ok(addr) = s.parse() {
        return Ok(addr);
    }
    let with_port = if s.contains(':') {
        s.to_string()
    } else {
        format!("{s}:{default_port}")
    };
    Ok(
        std::net::ToSocketAddrs::to_socket_addrs(&with_port.as_str())?
            .next()
            .ok_or("no addrs resolved")?,
    )
}

struct StaticResolver {
    tag_reference: String,
    metadata: GalaxyTagMetadata,
}

impl StaticResolver {
    fn new(tag_reference: &str) -> Self {
        let (object, attribute) = tag_reference
            .split_once('.')
            .unwrap_or((tag_reference, "TestInt"));
        Self {
            tag_reference: tag_reference.to_string(),
            metadata: GalaxyTagMetadata {
                object_tag_name: object.to_string(),
                attribute_name: attribute.to_string(),
                primitive_name: None,
                platform_id: 1,
                engine_id: 2,
                object_id: 3,
                primitive_id: 0,
                attribute_id: 7,
                property_id: GalaxyTagMetadata::VALUE_PROPERTY_ID,
                mx_data_type: 2,
                is_array: false,
                security_classification: 0,
                attribute_source: "dynamic".into(),
            },
        }
    }
}

#[async_trait::async_trait]
impl Resolver for StaticResolver {
    async fn resolve(&self, tag: &str) -> Result<GalaxyTagMetadata, ResolverError> {
        if tag == self.tag_reference {
            Ok(self.metadata.clone())
        } else {
            Err(ResolverError::NotFound {
                tag_reference: tag.to_string(),
            })
        }
    }
}