ek-ctp 0.12.2

//! 0.11.0 第4刀 · 合约流分批发布(判据 C16 的网关侧半边,R5 裁定)。
//!
//! 钉四件事:
//! 1. **每 2000 行发一批**:一次 4500 行的查询在总线上是 3 帧(2000+2000+尾巴 500),
//!    不是 4500 帧 —— 消费方邮箱里的帧数降两个数量级;
//! 2. **收尾语义不变**:`eContractFinish` 照发(载荷网关名),且**尾巴先于收尾**;
//! 3. **会话复位把尾巴发掉**:半批不压在锁里,也不随插件进坟墓(`on_stop` 同);
//! 4. 载荷是 `Vec<ContractData>`(批形),线名仍是裸线 `eContract.`。

use ek_ctp::driver::TdEvent;
use ek_ctp::wire::InstrumentField;
use ek_ctp::{CONTRACT_BATCH_ROWS, CtpConfig, CtpPlugin, EVENT_CONTRACT_FINISH};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

/// 合约流采集器:数帧、数行、记每帧行数、记收尾线到到货时刻。
#[derive(Default)]
struct ContractGrab {
    frames: Mutex<Vec<usize>>,
    finish: AtomicUsize,
}

impl ek_core::plugin::Plugin for ContractGrab {
    fn name(&self) -> &str {
        "contractgrab"
    }
    fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
        vec!["eContract.".into(), EVENT_CONTRACT_FINISH.into()]
    }
    fn on_mail(&self, e: &ek_event::Event) {
        match e.event_type.as_str() {
            "eContract." => {
                let n = if let Some(v) = e
                    .data
                    .downcast_ref::<Vec<ek_data::contract::ContractData>>()
                {
                    v.len()
                } else if e
                    .data
                    .downcast_ref::<ek_data::contract::ContractData>()
                    .is_some()
                {
                    1
                } else {
                    0
                };
                self.frames.lock().unwrap().push(n);
            }
            _ => {
                self.finish.fetch_add(1, Ordering::SeqCst);
            }
        }
    }
    fn on_start(&self) {}
}

fn instrument(id: &str) -> InstrumentField {
    InstrumentField {
        instrument_id: id.into(),
        exchange_id: "SHFE".into(),
        instrument_name: id.into(),
        product_class: "1".into(),
        volume_multiple: 10.0,
        price_tick: 1.0,
        max_limit_order_volume: 500.0,
        min_limit_order_volume: 1.0,
        max_market_order_volume: 30.0,
        min_market_order_volume: 1.0,
        delivery_year: 2026,
        delivery_month: 5,
        product_id: "rb".into(),
        is_trading: 1,
        open_date: String::new(),
    }
}

/// 起一台带采集器的主机 + 网关插件。
fn host_with_grab() -> (
    ek_core::MainEngine,
    Arc<ek_event::EventEngine>,
    Arc<ContractGrab>,
    Arc<CtpPlugin>,
) {
    let ev = Arc::new(ek_event::EventEngine::new());
    let host = ek_core::MainEngine::new(Some(ev.clone()));
    host.start().unwrap();
    let grab = Arc::new(ContractGrab::default());
    host.add_plugin(grab.clone()).unwrap();
    let p = Arc::new(CtpPlugin::new(CtpConfig::default()));
    host.add_plugin(p.clone()).unwrap();
    (host, ev, grab, p)
}

fn feed_rows(p: &CtpPlugin, n: usize) {
    for i in 0..n {
        let id = format!("rb{:04}", i);
        p.on_td_event(TdEvent::RspQryInstrument {
            last: false,
            row: Some(instrument(&id)),
            error: None,
        });
    }
}

fn frames_until(grab: &ContractGrab, min_frames: usize) -> Vec<usize> {
    for _ in 0..80 {
        let f = grab.frames.lock().unwrap().clone();
        if f.len() >= min_frames {
            return f;
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    }
    grab.frames.lock().unwrap().clone()
}

// D1:4500 行 → 2 整批(各 2000)+ 收尾尾巴 500,收尾线在尾巴之后到货
#[test]
fn d1_contract_stream_publishes_in_batches_of_2000() {
    let (host, _ev, grab, p) = host_with_grab();
    feed_rows(&p, 4500);
    let full = frames_until(&grab, 2);
    assert!(
        full.len() >= 2,
        "两整批要在收尾前到货(渐进显形不退化),实测 {full:?}"
    );
    assert!(
        full[0] == CONTRACT_BATCH_ROWS && full[1] == CONTRACT_BATCH_ROWS,
        "整批必须是 {CONTRACT_BATCH_ROWS} 行/帧,实测 {full:?}"
    );
    // 收尾:驱动在一行 last 上收尾(带「查询成功」回执)
    p.on_td_event(TdEvent::RspQryInstrument {
        last: true,
        row: None,
        error: None,
    });
    let mut frames = Vec::new();
    for _ in 0..80 {
        let f = grab.frames.lock().unwrap().clone();
        let fin = grab.finish.load(Ordering::SeqCst);
        if fin >= 1 {
            frames = f;
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
        frames = f;
    }
    assert_eq!(
        grab.finish.load(Ordering::SeqCst),
        1,
        "eContractFinish 要到货"
    );
    let total: usize = frames.iter().sum();
    assert_eq!(
        total, 4500,
        "行数一条不少:帧行数合计要等于投喂数,实测 {frames:?}"
    );
    assert_eq!(
        frames.len(),
        3,
        "4500 行 = 2000+2000+500 三帧,实测 {frames:?}"
    );
    assert_eq!(frames[2], 500, "尾巴 500 行,实测 {frames:?}");
    host.close_all(std::time::Duration::from_secs(4));
}

// D2:会话复位把没发出去的半批发掉(尾巴不压锁)
#[test]
fn d2_session_reset_flushes_the_partial_batch() {
    let (host, _ev, grab, p) = host_with_grab();
    feed_rows(&p, 1500);
    // 一批都不满:此刻总线应当是 0 帧
    std::thread::sleep(std::time::Duration::from_millis(200));
    assert_eq!(
        grab.frames.lock().unwrap().len(),
        0,
        "不满一批不发帧(等下一行/收尾)"
    );
    p.disconnect();
    let f = frames_until(&grab, 1);
    assert_eq!(
        f.first().copied(),
        Some(1500),
        "复位要把尾巴发掉,实测 {f:?}"
    );
    // 复位后新会话从头攒:再来 2000 行 → 一整批
    feed_rows(&p, CONTRACT_BATCH_ROWS);
    let f2 = frames_until(&grab, 2);
    assert_eq!(f2.len(), 2, "新会话独立攒批,实测 {f2:?}");
    assert_eq!(f2[1], CONTRACT_BATCH_ROWS);
    host.close_all(std::time::Duration::from_secs(4));
}