roplat 0.2.0

//! 旁路通讯 —— 三缓冲状态更新 集成测试
//!
//! 功能点测试：
//! 1. 透明类型（repr(C) 结构体）的三缓冲通讯
//! 2. 不透明类型的生命周期管理
//! 3. SPMC（单写多读）广播
//! 4. 惰性内存分配
//! 5. 多线程场景下的正确性
//! 6. roplat_msg 过程宏生成 C 头文件和 Python 存根

use roplat::comm::opaque::{OpaqueData, TypedOpaque};
use roplat::comm::triple_buffer::{TripleBufferChannel, create_triple_buffer};
use std::ffi::c_void;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;

// ============================================================
// 1. 透明类型通讯
// ============================================================

/// 示例透明类型：传感器数据
#[derive(Clone, Debug, PartialEq)]
#[repr(C)]
struct SensorData {
    x: f64,
    y: f64,
    z: f64,
    timestamp: u64,
}

/// 示例透明类型：电机命令
#[derive(Clone, Debug, PartialEq)]
#[repr(C)]
struct MotorCommand {
    velocity: f32,
    torque: f32,
    position: f64,
}

/// 示例透明类型：包含定长数组
#[derive(Clone, Debug, PartialEq)]
#[repr(C)]
struct JointState {
    positions: [f64; 6],
    velocities: [f64; 6],
    efforts: [f64; 6],
}

// ============================================================
// 测试：透明类型基础通讯
// ============================================================

#[test]
fn test_transparent_sensor_data_communication() {
    let (mut pub_sensor, mut subs) = create_triple_buffer::<SensorData>(1);
    let sub = &mut subs[0];

    // 初始无数据
    assert!(sub.get_latest().is_none());

    // 发布传感器数据
    pub_sensor.publish(SensorData { x: 1.0, y: 2.0, z: 3.0, timestamp: 1000 });

    let data = sub.get_latest().unwrap();
    assert_eq!(data.x, 1.0);
    assert_eq!(data.y, 2.0);
    assert_eq!(data.z, 3.0);
    assert_eq!(data.timestamp, 1000);
}

#[test]
fn test_transparent_motor_command_communication() {
    let (mut publisher, mut subs) = create_triple_buffer::<MotorCommand>(1);
    let sub = &mut subs[0];

    publisher.publish(MotorCommand { velocity: std::f32::consts::PI, torque: 1.5, position: 90.0 });

    let cmd = sub.get_latest().unwrap();
    assert_eq!(cmd.velocity, std::f32::consts::PI);
    assert_eq!(cmd.torque, 1.5);
    assert_eq!(cmd.position, 90.0);
}

#[test]
fn test_transparent_array_type_communication() {
    let (mut publisher, mut subs) = create_triple_buffer::<JointState>(1);
    let sub = &mut subs[0];

    let state = JointState {
        positions: [0.0, 1.0, 2.0, 3.0, 4.0, 5.0],
        velocities: [0.1, 0.2, 0.3, 0.4, 0.5, 0.6],
        efforts: [10.0, 20.0, 30.0, 40.0, 50.0, 60.0],
    };

    publisher.publish(state.clone());
    let received = sub.get_latest().unwrap();
    assert_eq!(received, &state);
}

// ============================================================
// 测试：SPMC 广播
// ============================================================

#[test]
fn test_spmc_broadcast_to_multiple_subscribers() {
    let (mut publisher, mut subs) = create_triple_buffer::<SensorData>(5);

    let data = SensorData { x: 42.0, y: -1.0, z: 0.5, timestamp: 999 };

    publisher.publish(data.clone());

    for (i, sub) in subs.iter_mut().enumerate() {
        let received = sub
            .get_latest()
            .unwrap_or_else(|| panic!("Subscriber {} should have data", i));
        assert_eq!(received, &data, "Subscriber {} received wrong data", i);
    }
}

#[test]
fn test_spmc_subscribers_see_latest_update() {
    let (mut publisher, mut subs) = create_triple_buffer::<i32>(3);

    publisher.publish(1);
    publisher.publish(2);
    publisher.publish(3);

    // SPMC 三缓冲中，每个 ctrl 在 publish 时被交换，
    // 后面的订阅者可能看到稍旧的值（取决于交换顺序）
    // 但所有订阅者都应该看到某个已发布的值
    for sub in &mut subs {
        let val = sub.get_latest().unwrap();
        assert!(*val >= 1 && *val <= 3, "Value {} out of range", *val);
    }
}

// ============================================================
// 测试：惰性内存分配
// ============================================================

#[test]
fn test_lazy_allocation_no_default_constructor() {
    /// 无默认构造函数的类型
    #[derive(Clone, Debug, PartialEq)]
    struct NoDefault {
        data: Vec<u8>,
        label: String,
    }

    let (mut publisher, mut subs) = create_triple_buffer::<NoDefault>(2);

    // 第一次发布触发惰性分配
    publisher.publish(NoDefault { data: vec![1, 2, 3, 4, 5], label: "test".to_string() });

    for sub in &mut subs {
        let val = sub.get_latest().unwrap();
        assert_eq!(val.data, vec![1, 2, 3, 4, 5]);
        assert_eq!(val.label, "test");
    }
}

#[test]
fn test_lazy_allocation_convergence() {
    let (mut publisher, mut subs) = create_triple_buffer::<u64>(2);

    // 多次发布后系统应自动收敛到稳态
    for i in 0..100 {
        publisher.publish(i);
    }

    // 每个订阅者应该看到接近近期的值
    for sub in &mut subs {
        let val = sub.get_latest().unwrap();
        assert!(*val >= 90, "Value {} should be near 99", *val);
    }
}

// ============================================================
// 测试：TripleBufferChannel（带生命周期管理）
// ============================================================

#[test]
fn test_channel_lifecycle() {
    let channel = TripleBufferChannel::<SensorData>::new(2);
    let mut publisher = channel.publisher();
    let mut sub0 = channel.subscriber(0);
    let mut sub1 = channel.subscriber(1);

    publisher.publish(SensorData { x: 10.0, y: 20.0, z: 30.0, timestamp: 42 });

    assert_eq!(sub0.get_latest().unwrap().x, 10.0);
    assert_eq!(sub1.get_latest().unwrap().x, 10.0);
}

// ============================================================
// 测试：多线程场景
// ============================================================

#[test]
fn test_multithreaded_publish_subscribe() {
    use std::sync::Arc;

    let (mut publisher, mut subs) = create_triple_buffer::<u64>(1);
    let mut sub = subs.pop().unwrap();

    let done = Arc::new(AtomicBool::new(false));
    let done_clone = done.clone();

    let producer = thread::spawn(move || {
        for i in 0..1000u64 {
            publisher.publish(i);
        }
        done_clone.store(true, Ordering::SeqCst);
    });

    let consumer = thread::spawn(move || {
        // 注：本三缓冲实现为无锁 SPMC，提供"最新值广播"语义而非"严格单调"。
        // 消费者把自己刚读到的旧 buffer 还给 ready 槽位后，若生产者尚未再次发
        // 布，可能再次读到自己之前的旧值。因此这里仅断言读到的值落在合法区间，
        // 并校验最终能看见接近 999 的值。
        let mut max_seen: Option<u64> = None;
        let mut any_value: Option<u64> = None;
        loop {
            if let Some(&val) = sub.get_latest() {
                assert!(val < 1000, "value out of range: {}", val);
                any_value = Some(val);
                max_seen = Some(max_seen.map_or(val, |m| m.max(val)));
            }
            if done.load(Ordering::SeqCst) {
                if let Some(&val) = sub.get_latest() {
                    assert!(val < 1000);
                    any_value = Some(val);
                    max_seen = Some(max_seen.map_or(val, |m| m.max(val)));
                }
                break;
            }
        }
        (any_value, max_seen)
    });

    producer.join().unwrap();
    let (final_val, max_val) = consumer.join().unwrap();
    assert!(final_val.is_some(), "Consumer should have seen some values");
    // 由于生产者一定发布过 999，消费者应能看见接近最终值。
    assert!(
        max_val.unwrap_or(0) >= 900,
        "Consumer should observe values close to producer's tail, got max={:?}",
        max_val
    );
}

// ============================================================
// 测试：不透明数据
// ============================================================

#[test]
fn test_opaque_data_create_and_destroy() {
    static DESTROYED: AtomicBool = AtomicBool::new(false);

    unsafe extern "C" fn destroy_data(ptr: *mut c_void) {
        unsafe {
            let _ = Box::from_raw(ptr as *mut String);
        }
        DESTROYED.store(true, Ordering::SeqCst);
    }

    {
        let data = Box::into_raw(Box::new("hello opaque".to_string()));
        let opaque = unsafe {
            OpaqueData::new(
                data as *mut c_void,
                destroy_data,
                std::mem::size_of::<String>(),
                "String",
            )
        };

        assert_eq!(opaque.type_id(), "String");
        assert!(!opaque.is_null());

        unsafe {
            let s = opaque.cast_ref::<String>();
            assert_eq!(s, "hello opaque");
        }
    }

    // Drop 后应调用释放函数
    assert!(DESTROYED.load(Ordering::SeqCst));
}

#[test]
fn test_typed_opaque_marker() {
    struct CppNode;
    struct PyNode;

    unsafe extern "C" fn destroy_i32(ptr: *mut c_void) {
        unsafe {
            let _ = Box::from_raw(ptr as *mut i32);
        }
    }

    let cpp_data = Box::into_raw(Box::new(42i32));
    let cpp_opaque =
        unsafe { OpaqueData::new(cpp_data as *mut c_void, destroy_i32, 4, "cpp_node_data") };
    let typed_cpp = TypedOpaque::<CppNode>::from_opaque(cpp_opaque);

    let py_data = Box::into_raw(Box::new(100i32));
    let py_opaque =
        unsafe { OpaqueData::new(py_data as *mut c_void, destroy_i32, 4, "py_node_data") };
    let typed_py = TypedOpaque::<PyNode>::from_opaque(py_opaque);

    assert_eq!(typed_cpp.inner().type_id(), "cpp_node_data");
    assert_eq!(typed_py.inner().type_id(), "py_node_data");
}

#[test]
fn test_opaque_ffi_functions() {
    static FFI_DESTROYED: AtomicBool = AtomicBool::new(false);

    unsafe extern "C" fn ffi_destroy(ptr: *mut c_void) {
        unsafe {
            let _ = Box::from_raw(ptr as *mut [f64; 3]);
        }
        FFI_DESTROYED.store(true, Ordering::SeqCst);
    }

    let value = Box::into_raw(Box::new([1.0f64, 2.0, 3.0]));
    let type_id = std::ffi::CString::new("f64_array_3").unwrap();

    unsafe {
        use roplat::comm::opaque::{
            roplat_opaque_create, roplat_opaque_destroy, roplat_opaque_get_ptr,
            roplat_opaque_get_size,
        };

        let opaque = roplat_opaque_create(
            value as *mut c_void,
            ffi_destroy,
            std::mem::size_of::<[f64; 3]>(),
            type_id.as_ptr(),
        );

        assert!(!opaque.is_null());
        assert_eq!(roplat_opaque_get_size(opaque), 24);

        let ptr = roplat_opaque_get_ptr(opaque);
        let arr = &*(ptr as *const [f64; 3]);
        assert_eq!(arr, &[1.0, 2.0, 3.0]);

        roplat_opaque_destroy(opaque);
    }

    assert!(FFI_DESTROYED.load(Ordering::SeqCst));
}

// ============================================================
// 测试：roplat_msg 过程宏
// ============================================================

#[roplat::roplat_msg]
#[repr(C)]
#[derive(Clone, Debug, PartialEq)]
/// 用于宏展开验证的测试消息。
pub struct TestMsg {
    /// X 坐标。
    pub x: f64,
    /// Y 坐标。
    pub y: f64,
    /// Z 坐标。
    pub z: f64,
    /// 整型标识。
    pub id: u32,
    /// 状态标记。
    pub active: bool,
}

#[test]
fn test_roplat_msg_c_header_generation() {
    let header = TestMsg::C_HEADER;

    // 验证生成的 C 头文件内容
    assert!(header.contains("#ifndef ROPLAT_MSG_TESTMSG_H"));
    assert!(header.contains("#define ROPLAT_MSG_TESTMSG_H"));
    assert!(header.contains("typedef struct TestMsg"));
    assert!(header.contains("double x;"));
    assert!(header.contains("double y;"));
    assert!(header.contains("double z;"));
    assert!(header.contains("uint32_t id;"));
    assert!(header.contains("bool active;"));
    assert!(header.contains("} TestMsg;"));
    assert!(header.contains("#include <stdint.h>"));
    assert!(header.contains("extern \"C\""));
}

#[test]
fn test_roplat_msg_python_stub_generation() {
    let stub = TestMsg::PYTHON_STUB;

    // 验证生成的 Python 存根内容
    assert!(stub.contains("import ctypes"));
    assert!(stub.contains("class TestMsg(ctypes.Structure):"));
    assert!(stub.contains("_fields_ = ["));
    assert!(stub.contains("(\"x\", ctypes.c_double)"));
    assert!(stub.contains("(\"y\", ctypes.c_double)"));
    assert!(stub.contains("(\"z\", ctypes.c_double)"));
    assert!(stub.contains("(\"id\", ctypes.c_uint32)"));
    assert!(stub.contains("(\"active\", ctypes.c_bool)"));
}

/// 含数组字段的消息
#[roplat::roplat_msg]
#[repr(C)]
#[derive(Clone, Debug, PartialEq)]
/// 含数组字段的测试消息。
pub struct ArrayMsg {
    /// 固定长度浮点数组。
    pub data: [f32; 4],
    /// 有效元素个数。
    pub count: u32,
}

#[test]
fn test_roplat_msg_array_field() {
    let header = ArrayMsg::C_HEADER;
    assert!(header.contains("float data[4];"));
    assert!(header.contains("uint32_t count;"));

    let stub = ArrayMsg::PYTHON_STUB;
    assert!(stub.contains("(\"data\", ctypes.c_float * 4)"));
    assert!(stub.contains("(\"count\", ctypes.c_uint32)"));
}

// ============================================================
// 测试：透明类型通过三缓冲通讯
// ============================================================

#[test]
fn test_roplat_msg_type_with_triple_buffer() {
    let (mut publisher, mut subs) = create_triple_buffer::<TestMsg>(2);

    let msg = TestMsg { x: 1.5, y: 2.5, z: 3.5, id: 42, active: true };

    publisher.publish(msg.clone());

    for sub in &mut subs {
        let received = sub.get_latest().unwrap();
        assert_eq!(received, &msg);
    }

    // 更新
    let msg2 = TestMsg { x: 10.0, y: 20.0, z: 30.0, id: 43, active: false };

    publisher.publish(msg2.clone());

    for sub in &mut subs {
        let received = sub.get_latest().unwrap();
        assert_eq!(received, &msg2);
    }
}

// ============================================================
// 测试：语言内不透明类型通讯（通过三缓冲传递 OpaqueWrapper）
// ============================================================

/// 模拟不透明数据的包装类型（可通过三缓冲传递）
#[derive(Clone, Debug)]
struct OpaqueWrapper {
    type_id: &'static str,
    data: Vec<u8>,
}

#[test]
fn test_opaque_type_intra_language_comm() {
    let (mut publisher, mut subs) = create_triple_buffer::<OpaqueWrapper>(1);
    let sub = &mut subs[0];

    let wrapper = OpaqueWrapper { type_id: "CppSensorNode", data: vec![0xDE, 0xAD, 0xBE, 0xEF] };

    publisher.publish(wrapper.clone());

    let received = sub.get_latest().unwrap();
    assert_eq!(received.type_id, "CppSensorNode");
    assert_eq!(received.data, vec![0xDE, 0xAD, 0xBE, 0xEF]);
}

// ============================================================
// 测试：环形队列通讯
// ============================================================

use roplat::comm::ring_buffer::{RingBufferChannel, create_ring_buffer};

#[test]
fn test_ring_buffer_transparent_sensor_data() {
    let (mut writer, mut reader) = create_ring_buffer::<SensorData>(4);

    let s1 = SensorData { x: 1.0, y: 2.0, z: 3.0, timestamp: 100 };
    let s2 = SensorData { x: 4.0, y: 5.0, z: 6.0, timestamp: 200 };

    assert!(writer.try_push(&s1));
    assert!(writer.try_push(&s2));

    let r1 = reader.try_pop().unwrap();
    assert_eq!(r1.x, 1.0);
    assert_eq!(r1.timestamp, 100);

    let r2 = reader.try_pop().unwrap();
    assert_eq!(r2.x, 4.0);
    assert_eq!(r2.timestamp, 200);

    assert!(reader.try_pop().is_none());
}

#[test]
fn test_ring_buffer_fifo_ordering() {
    let (mut writer, mut reader) = create_ring_buffer::<MotorCommand>(8);

    for i in 0..8 {
        writer.try_push(&MotorCommand { velocity: i as f32, torque: 0.0, position: 0.0 });
    }

    for i in 0..8 {
        let cmd = reader.try_pop().unwrap();
        assert_eq!(cmd.velocity, i as f32, "FIFO ordering broken at index {i}");
    }
}

#[test]
fn test_ring_buffer_full_behavior() {
    let (mut writer, mut reader) = create_ring_buffer::<u64>(3);

    assert!(writer.try_push(&1));
    assert!(writer.try_push(&2));
    assert!(writer.try_push(&3));
    assert!(!writer.try_push(&4)); // 满，拒绝

    assert_eq!(reader.try_pop(), Some(1));
    assert!(writer.try_push(&4)); // 有空位
}

#[test]
fn test_ring_buffer_force_push_drops_oldest() {
    let (mut writer, mut reader) = create_ring_buffer::<i32>(2);

    writer.force_push(&10);
    writer.force_push(&20);
    writer.force_push(&30); // 丢弃 10

    assert_eq!(reader.try_pop(), Some(20));
    assert_eq!(reader.try_pop(), Some(30));
    assert!(reader.try_pop().is_none());
}

#[test]
fn test_ring_buffer_channel_lifecycle() {
    let channel = RingBufferChannel::<SensorData>::new(4);
    assert_eq!(channel.capacity(), 4);

    let mut writer = channel.writer();
    let mut reader = channel.reader();

    let data = SensorData { x: 9.0, y: 8.0, z: 7.0, timestamp: 999 };
    writer.try_push(&data);

    let got = reader.try_pop().unwrap();
    assert_eq!(got.z, 7.0);
}

#[test]
fn test_ring_buffer_spsc_multithreaded() {
    let (mut writer, mut reader) = create_ring_buffer::<u64>(128);

    let count = 50_000u64;

    let producer = thread::spawn(move || {
        for i in 0..count {
            while !writer.try_push(&i) {
                std::hint::spin_loop();
            }
        }
    });

    let consumer = thread::spawn(move || {
        let mut received = Vec::with_capacity(count as usize);
        while received.len() < count as usize {
            if let Some(v) = reader.try_pop() {
                received.push(v);
            } else {
                std::hint::spin_loop();
            }
        }
        received
    });

    producer.join().unwrap();
    let received = consumer.join().unwrap();

    assert_eq!(received.len(), count as usize);
    for (i, &v) in received.iter().enumerate() {
        assert_eq!(v, i as u64, "FIFO violated at position {i}");
    }
}