mxaccess/rust/crates/mxaccess-compat/tests/buffered_subscribe_live.rs

//! Live verification of F36 — buffered subscribe (`Session::subscribe_buffered`)
//! round-trips against AVEVA and yields `DataChange`s at the requested cadence.
//!
//! F49 step 1. Asserts the structural property of F36 (single
//! `RegisterReference` with `.property(buffer)` suffix, no separate
//! `AdviseSupervisory` follow-up, no `SetBufferedUpdateInterval` RPC)
//! is preserved end-to-end. The structural piece is unit-tested
//! exhaustively in `crates/mxaccess/src/session.rs` (search
//! `subscribe_buffered_nmx`); this test confirms the wire round-trip
//! actually delivers updates.
//!
//! Gated on `MX_LIVE` env + `live-windows-com` feature. Uses
//! `Session::connect_nmx_auto` (F55-proven path).
//!
//! Run with:
//! ```text
//! cd rust
//! cargo test -p mxaccess-compat --features live-windows-com \
//!     --test buffered_subscribe_live -- --ignored --nocapture
//! ```

#![allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic
)]

#[cfg(all(windows, feature = "live-windows-com"))]
mod live {
    use std::sync::Arc;
    use std::time::{Duration, Instant};

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

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

    fn ntlm_from_test_env() -> NtlmClientContext {
        let user = std::env::var("MX_TEST_USER").expect("MX_TEST_USER");
        let password = std::env::var("MX_TEST_PASSWORD").expect("MX_TEST_PASSWORD");
        let domain = std::env::var("MX_TEST_DOMAIN").unwrap_or_default();
        let hostname = std::env::var("COMPUTERNAME").unwrap_or_default();
        NtlmClientContext::new(&user, &password, &domain, Some(&hostname))
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    #[ignore]
    async fn buffered_subscribe_yields_updates() {
        if std::env::var_os("MX_LIVE").is_none() {
            eprintln!("MX_LIVE not set — skipping live test");
            return;
        }
        let tag = std::env::var("MX_TEST_TAG")
            .unwrap_or_else(|_| "TestChildObject.TestInt".to_string());

        // Initialise tracing so RUST_LOG=trace surfaces dcom_sink +
        // router events (set by the caller). Init may fail if a
        // subscriber is already installed — ignore the result.
        let _ = tracing_subscriber::fmt()
            .with_env_filter(
                tracing_subscriber::EnvFilter::try_from_default_env()
                    .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
            )
            .with_test_writer()
            .try_init();

        eprintln!("connecting via Session::connect_nmx_auto");
        let session = Session::connect_nmx_auto(
            ntlm_from_test_env,
            SessionOptions::default(),
            Arc::new(StaticResolver::new(&tag)),
            RecoveryPolicy::default(),
        )
        .await
        .expect("connect_nmx_auto");
        eprintln!("session connected");

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

        // Buffered cadence is delivery-only — the engine pushes at the
        // configured interval but only when the value has changed.
        // Spawn a background writer that bumps the tag every 500ms so
        // the engine always has a fresh value to deliver at the next
        // cadence boundary. 30s drain window.
        let deadline = Instant::now() + Duration::from_secs(30);
        let writer_session = session.clone();
        let writer_tag = tag.clone();
        let writer_stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let writer_stop_clone = writer_stop.clone();
        let writer = tokio::spawn(async move {
            let mut value: i32 = 1_000;
            while !writer_stop_clone.load(std::sync::atomic::Ordering::Acquire) {
                if let Err(e) = writer_session
                    .write(&writer_tag, MxValue::Int32(value))
                    .await
                {
                    eprintln!("writer: write({value}) failed: {e}");
                    break;
                }
                value = value.wrapping_add(1);
                tokio::time::sleep(Duration::from_millis(500)).await;
            }
            value
        });

        let mut received = 0;
        let mut last_ts = None;
        while received < 3 && Instant::now() < deadline {
            match tokio::time::timeout(Duration::from_secs(5), sub.next()).await {
                Ok(Some(Ok(dc))) => {
                    eprintln!(
                        "[{received}] {} = {:?} ts={:?}",
                        dc.reference, dc.value, dc.timestamp
                    );
                    received += 1;
                    last_ts = Some(dc.timestamp);
                }
                Ok(Some(Err(e))) => {
                    writer_stop.store(true, std::sync::atomic::Ordering::Release);
                    let _ = writer.await;
                    panic!("subscription error: {e}");
                }
                Ok(None) => break,
                Err(_) => {
                    eprintln!("5s gap waiting for next update");
                }
            }
        }
        writer_stop.store(true, std::sync::atomic::Ordering::Release);
        let last_value = writer.await.unwrap_or(-1);
        eprintln!("writer stopped after value {last_value}");

        assert!(
            received >= 1,
            "no DataChange arrived within 15s — buffered subscribe didn't round-trip"
        );
        eprintln!("received {received} updates; last ts = {last_ts:?}");

        session.unsubscribe(sub).await.expect("unsubscribe");
        session.shutdown_nmx().await.expect("shutdown");
        eprintln!("clean shutdown");
    }
}

#[cfg(not(all(windows, feature = "live-windows-com")))]
mod live {
    #[test]
    #[ignore]
    fn buffered_subscribe_yields_updates() {
        eprintln!("test skipped: requires Windows + live-windows-com feature");
    }
}