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, MxValue, RecoveryPolicy, Session, SessionOptions};
    use mxaccess_galaxy::SqlTagResolver;
    use mxaccess_rpc::ntlm::NtlmClientContext;

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

        // Real Galaxy DB resolver — the StaticResolver shim with
        // hardcoded engine_id=2 / platform_id=1 was silently accepted
        // by NmxSvc for writes (the OnWriteComplete live test still
        // works) but caused buffered RegisterReference to land at a
        // non-existent engine, returning a stub `0x11` and never
        // dispatching DataUpdates. F56 root cause.
        let galaxy_db = std::env::var("MX_GALAXY_DB")
            .expect("MX_GALAXY_DB (set via tools/Setup-LiveProbeEnv.ps1)");
        let resolver = Arc::new(
            SqlTagResolver::from_ado_string(&galaxy_db).expect("SqlTagResolver"),
        );

        // Dump resolved metadata so we can diff against captured .NET wire bytes.
        {
            use mxaccess_galaxy::Resolver as _;
            let m = resolver.resolve(&tag).await.expect("resolve test tag");
            eprintln!(
                "resolved {tag}: object_tag={:?} attribute={:?} primitive={:?} platform={} engine={} object={} attribute_id={} property_id={} mx_type={} is_array={}",
                m.object_tag_name,
                m.attribute_name,
                m.primitive_name,
                m.platform_id,
                m.engine_id,
                m.object_id,
                m.attribute_id,
                m.property_id,
                m.mx_data_type,
                m.is_array,
            );
        }

        eprintln!("connecting via Session::connect_nmx_auto");
        let session = Session::connect_nmx_auto(
            ntlm_from_test_env,
            SessionOptions::default(),
            resolver,
            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");
    }
}