Skip to main content

ek_ctp/
plugin.rs

1//! # CtpPlugin —— ek-core 宿主注册的 CTP 网关插件。
2//!
3//! 谁发谁收:
4//! - **发**:驱动输出 → `ctx.put` 上总线(纯生产者:**裸线键** `eTick.`/`eContract.`/
5//!   `eOrder.`/`eTrade.`/`ePosition.`/`eAccount.`/
6//!   `ePositionDone.`/`eLog`;0.4.0 D-EMIT 起裸键,网关身份归载荷 `gateway_name`;
7//!   线值与 ek-oms 的 wire 模块同表同值)。
8//! - **收**:`eTimer`(ek-event 秒针,2s 轮转驱动 + 行情日期刷新)与自身寻址
9//!   `eReq.<名>`(订阅/发单/撤单三项服务,对齐 oms 的 ask 面形制)。
10//!
11//! 传输层(`CtpTransport`)是 native 桥的接缝:默认构建无传输(可测纯逻辑与
12//! 请求面),native 特性(0.2.0)以官方 SDK 二进制实现该接口。
13//!
14//! 约定:
15//! - 原稿 `CtpGateway.count`(224-233 的 2 秒 gate)在此为插件内计数器;
16//! - `md.update_date` 与查询轮转同 gate(共节拍);
17//! - `EVENT_POSITION_QUERY_DONE` 落线值 `ePositionDone.<gw>`(与 ek-oms 相位门同表)。
18
19use crate::driver::{self, MdDriver, MdEvent, MdOut, TdDriver, TdEvent, TdOut};
20use crate::wire::InputOrder;
21use ek_core::context::PluginContext;
22use ek_core::envelope::{Envelope, ReplyErr, e_req_type};
23use ek_core::plugin::Plugin;
24use ek_core::service::ServiceDesc;
25use ek_data::cancel_request::CancelRequest;
26use ek_data::log::LogData;
27// 只被 native 面的 `transport_log` 使用(默认构建没有它,故按特性引入)。
28#[cfg(feature = "native")]
29use ek_data::log::LEVEL_ERROR;
30// 0.9.0 断开口落一条 INFO(默认构建也要能编译,故不挂特性)。
31use ek_data::log::LEVEL_INFO;
32use ek_data::order_request::OrderRequest;
33use ek_event::Event;
34use std::borrow::Cow;
35use std::sync::{Arc, Mutex, OnceLock};
36
37// ================ 总线线值(注意:与 ek-oms 的 wire 模块同表同值) ================
38
39const W_TICK: &str = "eTick.";
40const W_ORDER: &str = "eOrder.";
41const W_TRADE: &str = "eTrade.";
42const W_POSITION: &str = "ePosition.";
43const W_ACCOUNT: &str = "eAccount.";
44const W_CONTRACT: &str = "eContract.";
45/// 原稿 EVENT_POSITION_QUERY_DONE(event.py:17):**裸键,无 gw 后缀**,gw 放载荷
46/// ——ek-oms 侧以该裸键精确订阅(CtpPlugin emit 面),此处保持一字面。
47const W_POS_DONE: &str = "ePositionDone.";
48/// 日志线单源(ek-data::event::EVENT_LOG = "eLog",不再本地复定义)。
49use ek_data::event::EVENT_LOG as W_LOG;
50/// ek-event 秒针广播键(引用共享常量,不再重复定义字面值)。
51/// 由 [ek_event::EVENT_TIMER] 单源提供。
52pub use ek_event::EVENT_TIMER as ROUTE_TIMER;
53
54// ================ 服务请求/回包契约(ask 载荷类型) ================
55
56/// 订阅行情请求。
57#[derive(Debug, Clone)]
58pub struct SubscribeReq {
59    /// 合约代码symbols(逐笔)
60    pub symbols: Vec<String>,
61}
62
63/// 订阅请求回包。
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub struct SubscribeReply {
66    /// 是否受理(true = 已登记/已发起)
67    pub ok: bool,
68    /// 说明
69    pub msg: String,
70}
71
72/// 发单请求。
73#[derive(Debug, Clone)]
74pub struct SendOrderReq {
75    /// 下单请求
76    pub req: OrderRequest,
77}
78
79/// 发单回包。
80#[derive(Debug, Clone, PartialEq)]
81pub struct SendOrderReply {
82    /// 是否受理(原稿 `""` 面 → false)
83    pub ok: bool,
84    /// 成功 = `"{gw}.{front}_{session}_{ref}"`(OrderData.en_orderid 同式)
85    pub en_orderid: String,
86    /// 拒绝时说明
87    pub msg: String,
88}
89
90/// 撤单请求。
91#[derive(Debug, Clone)]
92pub struct CancelOrderReq {
93    /// 撤单请求
94    pub req: CancelRequest,
95}
96
97/// 撤单回包。
98#[derive(Debug, Clone, PartialEq, Eq)]
99pub struct CancelOrderReply {
100    /// 是否受理(true = 命令已交付)
101    pub ok: bool,
102    /// 说明
103    pub msg: String,
104}
105
106/// 费率/保证金率查询请求(附加接口面,超出原稿)。
107///
108/// `kind`: "margin" = 保证金率(hedge 过滤参与查询);"commission" = 手续费率
109/// (无 hedge 过滤)。查询结果经 TD 回调 → 驱动 eLog 落线(不新增事件实体)。
110#[derive(Debug, Clone)]
111pub struct QueryRateReq {
112    /// 查询类型("margin" / "commission")
113    pub kind: String,
114    /// 合约代码
115    pub instrument_id: String,
116    /// 投机套保档(仅 margin 查询用;'1'投机)
117    pub hedge_flag: String,
118    /// 交易所代码(过滤;可空)
119    pub exchange_id: String,
120}
121
122/// 费率查询回包(是否已交付 + RT 返回码文本)。
123#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct QueryRateReply {
125    /// 是否受理(true = 命令已交付 native)
126    pub ok: bool,
127    /// 说明(交付失败时含错误代码)
128    pub msg: String,
129}
130
131/// 网关配置请求(**插件自带配置面**,2026-10-17 定盘:配置归插件自身,
132/// 界面/宿主手工输入经 ask 进来;插件永不读配置文件)。
133///
134/// 语义 = 更新账号/前置 + 复位连接守卫 + 立即发起连接(先 TD 后 MD,与
135/// `connect` 同式)。`md_user`/`md_password` 留空回落 `td_user`/`td_password`
136/// (SimNow 同账号常态,界面少填两格)。
137#[derive(Debug, Clone, Default)]
138pub struct ConfigureReq {
139    /// 经纪公司代码
140    pub broker_id: String,
141    /// 交易前置(空 = 不连 TD;与 md_addr 至少一个非空)
142    pub td_addr: String,
143    /// 交易用户
144    pub td_user: String,
145    /// 交易密码
146    pub td_password: String,
147    /// 授权编码(空 = 直接登录)
148    pub td_auth_code: String,
149    /// 产品名称(AppID)
150    pub td_app_id: String,
151    /// 行情前置(空 = 不连 MD)
152    pub md_addr: String,
153    /// 行情用户(空回落 td_user)
154    pub md_user: String,
155    /// 行情密码(空回落 td_password)
156    pub md_password: String,
157    /// 收录的产品类白名单(CTP 单字符码,如 `["1"]`;空 = 回落默认「期货」)。
158    /// 界面勾选面:期货 `'1'` / 期权 `'2'`(现货期权 `'6'` 可配但不在默认勾选里)。
159    pub product_classes: Vec<String>,
160}
161
162/// 配置回包(ok = 已应用并发起连接;false = 校验未过,msg 报字段)。
163#[derive(Debug, Clone, PartialEq, Eq)]
164pub struct ConfigureReply {
165    /// 是否受理
166    pub ok: bool,
167    /// 说明
168    pub msg: String,
169    /// 实际生效的产品类白名单(已归一化;空请求回落默认「期货」`'1'`)。
170    /// 校验未过时回带**当前仍在生效**的那一份 —— 不是空表:空表会让界面误清空勾选,
171    /// 而网关其实还跑在旧勾选上。
172    pub product_classes: Vec<String>,
173}
174
175/// 状态查询请求(0.9.0 `{gw}.status`;无参数,单位结构体只为契约类型名占位)。
176#[derive(Debug, Clone, Default, PartialEq, Eq)]
177pub struct StatusReq;
178
179/// 网关状态读数(0.9.0 新增;`{gw}.status` 的返回体 = 界面状态机的唯一事实源)。
180///
181/// 口径边界(消费侧文案必须照这个说,不得越界):
182/// - `*_handle_up` = API 句柄已建并 registerFront,**不等于在线**(掉线时这一位仍为真);
183/// - `*_logged_in` = 最近一次登录成功且未断开 —— 「在线/掉线」的唯一硬证据;
184/// - 「查询断供」= `td_query_reject_streak` 达阈 **或** 距 `td_last_account_ms` 超时;
185///   阈值归消费侧(ek-ui)配置,网关只给原料不裁决(它不知道界面多久刷一次)。
186///   0.9.1 起合约查询被拒也进同一条连续串(消费侧要的是「查询这条路连续被拒几次」,
187///   不分路);要分路看 `td_instrument_rejects` 这个分项。
188/// - 计数与「最近时刻」跨重连保留:掉过几次线是历史事实,不因重连清零;
189/// - 从未发生过 = `None`(界面须显示「无证据」,不得显示 0 秒前)。
190#[derive(Debug, Clone, Default, PartialEq, Eq)]
191pub struct GatewayStatus {
192    /// 网关注册名(派生名 `ctp_<broker>_<账号>`)
193    pub gateway: String,
194    /// 读数时刻(纪元毫秒;消费侧据此算「距最后一次回报多久」)
195    pub sampled_ms: i64,
196    // ---- 交易侧 ----
197    /// TD 句柄已建(不等于在线)
198    pub td_handle_up: bool,
199    /// TD 已登录
200    pub td_logged_in: bool,
201    /// 最近一次 TD 登录失败(成功后清零)
202    pub td_login_failed: bool,
203    /// TD 认证通过
204    pub td_auth_ok: bool,
205    /// TD 认证失败
206    pub td_auth_failed: bool,
207    /// 登录回报带的 FrontID(0 = 从未登录成功)
208    pub td_front_id: i64,
209    /// 登录回报带的 SessionID
210    pub td_session_id: i64,
211    /// TD 断开累计次数
212    pub td_disconnects: u32,
213    /// 最近一次 TD 登录成功时刻
214    pub td_last_login_ok_ms: Option<i64>,
215    /// 最近一次资金回报时刻
216    pub td_last_account_ms: Option<i64>,
217    /// 最近一次持仓回报时刻
218    pub td_last_position_ms: Option<i64>,
219    /// 查询被拒累计次数(合约/持仓/资金三条查询路,0.9.1 起含合约)
220    pub td_query_rejects: u32,
221    /// 连续被拒次数(成功回报清零)
222    pub td_query_reject_streak: u32,
223    /// 合约查询被拒累计次数(0.9.1;`td_query_rejects` 的分项,不是另一本账)
224    pub td_instrument_rejects: u32,
225    /// 查询轮转闸门(未 arm = 一条轮转查询都不会发)
226    pub td_rotation_armed: bool,
227    /// 最近一次 TD 错误回报
228    pub td_last_error: Option<String>,
229    // ---- 行情侧 ----
230    /// MD 句柄已建(不等于在线)
231    pub md_handle_up: bool,
232    /// MD 已登录
233    pub md_logged_in: bool,
234    /// MD 断开累计次数
235    pub md_disconnects: u32,
236    /// 最近一次 MD 登录成功时刻
237    pub md_last_login_ok_ms: Option<i64>,
238    /// 最近一条行情回报时刻
239    pub md_last_tick_ms: Option<i64>,
240    /// 订阅被拒累计次数
241    pub md_sub_rejects: u32,
242    /// 最近一次 MD 错误回报
243    pub md_last_error: Option<String>,
244    /// 已登记订阅数(断开不清订阅集)
245    pub md_subscribed: usize,
246}
247
248/// 主动断开请求(0.9.0 `{gw}.disconnect`;裁定 1「断开 = 停会话,实例与订阅保留」)。
249#[derive(Debug, Clone, Default, PartialEq, Eq)]
250pub struct DisconnectReq;
251
252/// 断开回带(ok = 会话守卫已复位;msg 说明放掉了什么、保留了什么)。
253#[derive(Debug, Clone, PartialEq, Eq)]
254pub struct DisconnectReply {
255    /// 是否执行
256    pub ok: bool,
257    /// 说明
258    pub msg: String,
259}
260
261// ================ 传输层接缝(native 桥实现/dev 以测试桩实现) ================
262
263/// 传输出口:驱动动作落到真实 CTP 会话(0.2.0 native 才有官方实现)。
264pub trait CtpTransport: Send + Sync {
265    /// 创建行情 API 并 registerFront/init。
266    fn md_connect(&self, addr: &str, flow_path: &str);
267    /// 行情登录。
268    fn md_login(&self, user: &str, password: &str, broker: &str);
269    /// 行情订阅。
270    fn md_subscribe(&self, symbols: &[String]);
271    /// 创建交易 API 并 registerFront/init。
272    fn td_connect(&self, addr: &str, flow_path: &str);
273    /// 认证。
274    fn td_auth(&self, user: &str, broker: &str, auth_code: &str, app_id: &str);
275    /// 登录。
276    fn td_login(&self, user: &str, password: &str, broker: &str);
277    /// 结算单确认。
278    fn td_settlement_confirm(&self, broker: &str, user: &str);
279    /// 合约查询。
280    fn td_qry_instrument(&self);
281    /// 资金查询;返回 `reqQryTradingAccount` 的 rt 返回码(0.9.2)。原稿丢弃该返回码(876),
282    /// 于是发送侧被拒在计数与状态口里全隐形;非 0 现由插件喂进同一条查询连续串
283    /// (`TdDriver::note_query_send_failure`)。
284    fn td_qry_account(&self, broker: &str, user: &str) -> i32;
285    /// 持仓查询;返回 `reqQryInvestorPosition` 的 rt 返回码(0.9.2,同上,原稿 886)。
286    fn td_qry_position(&self, broker: &str, user: &str) -> i32;
287    /// 插单;返回 `reqOrderInsert` 的 rt 返回码(0 = 受理;非 0 = 原稿 846-848 的
288    /// 失败面:记「委托请求发送失败,错误代码:n」且**不回 SUBMITTING 单**)。
289    fn td_insert(&self, order: &InputOrder) -> i32;
290    /// 撤单;返回 `reqOrderAction` 的 rt 返回码(原稿不检,原样透传给 native 自定)。
291    fn td_cancel(&self, action: &crate::wire::InputOrderAction) -> i32;
292    /// 合约保证金率查询(附加接口面;返回 `ReqQryInstrumentMarginRate` rt 返回码)。
293    fn td_qry_margin_rate(&self, query: &crate::wire::QryInstrumentMarginRate) -> i32;
294    /// 合约手续费率查询(附加接口面;返回 `ReqQryInstrumentCommissionRate` rt 返回码)。
295    fn td_qry_commission_rate(&self, query: &crate::wire::QryInstrumentCommissionRate) -> i32;
296
297    // 交易状态没有查询面(SDK 6.7.7 无 ReqQryInstrumentStatus):该件纯被动,
298    // 由公有流推送经 TD 回调直达驱动,传输层不提供主动入口。
299
300    /// 关闭原生会话(宿主停插件时调用)。
301    ///
302    /// 默认空实现 = 无资源可放。官方实现里这一步是必需的:`Release()` 之后 SDK
303    /// **不再回调**,回调入口(ctx)才允许被放下 —— 顺序反了就是 use-after-free。
304    fn shutdown(&self) {}
305
306    /// 断开交易会话(释放句柄;默认空 = 纯逻辑面无会话可断)。
307    /// configure 强制重连面(0.5.0)用:旧登录态先断,新账号/密码才有显形机会。
308    fn td_release(&self) {}
309    /// 断开行情会话(释放句柄;默认空 = 纯逻辑面无会话可断)。
310    fn md_release(&self) {}
311}
312
313// ================ 网关配置 ================
314
315/// 网关配置(对齐原稿 connect setting 141-171;地址补前缀由使用方完成——
316/// 原稿在 connect 里对缺 "tcp://"/"ssl://" 的地址自动补 "tcp://",
317/// 本 crate 要求调用方先行规整,保持驱动数据纯净)。
318#[derive(Debug, Clone, Default)]
319pub struct CtpConfig {
320    /// 网关/插件注册名(唯一;默认 "ctp")
321    pub gateway_name: String,
322    /// 经纪公司代码
323    pub broker_id: String,
324    /// 交易前置
325    pub td_addr: String,
326    /// 交易用户
327    pub td_user: String,
328    /// 交易密码
329    pub td_password: String,
330    /// 授权编码(空 = 直接登录,原稿 448-451)
331    pub td_auth_code: String,
332    /// 产品名称(AppID)
333    pub td_app_id: String,
334    /// 行情前置
335    pub md_addr: String,
336    /// 行情用户
337    pub md_user: String,
338    /// 行情密码
339    pub md_password: String,
340    /// 收录的产品类白名单(D-PC-FILTER;空 = 回落默认「期货」`'1'`)。
341    /// 查询侧不过滤,收录侧按它严格丢弃(不进共享表、不上抛 `eContract`)。
342    pub product_classes: crate::map::ProductClasses,
343}
344
345impl CtpConfig {
346    /// 缺省名 "ctp" 的兜底。
347    pub fn gateway_name_or_default(&self) -> String {
348        if self.gateway_name.is_empty() {
349            "ctp".to_string()
350        } else {
351            self.gateway_name.clone()
352        }
353    }
354}
355
356/// 多账号面的注册名派生(**界面唯一入口**;配置面不存网关名)。
357///
358/// 规则 `ctp_<broker>_<账号>`,例 `ctp_9999_278153`。为什么必须是这个形状:
359///
360/// - 宿主名字规则 `[A-Za-z][A-Za-z0-9_-]{0,31}`(ek-core `valid_plugin_name`)
361///   ——**不能含点号**,总长 ≤32;
362/// - `en_orderid = "{gw}.{front}_{session}_{ref}"` 的反解用 `split_once('.')`
363///   (ek-converter `holding.rs`)——gw 里带点号会把反解串成 `"ctp"`;
364/// - 派生而非手填:手填会重名/打错;同 broker+账号 重复添加时派生名相同,
365///   由宿主 `Duplicate` 直接挡下(不需要额外的账号注册表)。
366///
367/// 返回 `None` = 输入不可用(broker 或账号为空、含 `[0-9A-Za-z]` 之外的字符、
368/// 或派生名超 32 字符)。调用方(界面)须拒绝并说明,**不得静默替换**。
369#[must_use]
370pub fn derived_gateway_name(broker_id: &str, account: &str) -> Option<String> {
371    let broker = sanitize_id_field(broker_id)?;
372    let account = sanitize_id_field(account)?;
373    let name = format!("ctp_{broker}_{account}");
374    // 宿主上限 32(`[A-Za-z][A-Za-z0-9_-]{0,31}`);超限界面提前拦。
375    if name.len() > 32 {
376        return None;
377    }
378    Some(name)
379}
380
381/// 账号位清洗:去首尾空白,只允许 `[0-9A-Za-z]`;空或含非法字符 → `None`。
382fn sanitize_id_field(raw: &str) -> Option<String> {
383    let trimmed = raw.trim();
384    if trimmed.is_empty() || !trimmed.chars().all(|c| c.is_ascii_alphanumeric()) {
385        return None;
386    }
387    Some(trimmed.to_string())
388}
389
390// ================ 插件 ================
391
392/// CTP 网关插件(纯生产者 + 自服务)。
393pub struct CtpPlugin {
394    gw: String,
395    /// 当前网关配置(connect 空地址守卫与费率查询的单一事实源)。
396    cfg: Mutex<CtpConfig>,
397    /// 经纪公司代码(费率查询的 broker_id;configure 可换)。
398    broker: Mutex<String>,
399    /// 投资者代码(费率查询的 investor_id;configure 可换)。
400    user: Mutex<String>,
401    md: Mutex<MdDriver>,
402    td: Mutex<TdDriver>,
403    /// 2s gate 计数器(原稿 self.count)
404    timer_count: Mutex<u32>,
405    /// 传输层(默认构建 None;native/测试注入)
406    trans: Option<Box<dyn CtpTransport>>,
407    ctx: OnceLock<PluginContext>,
408    /// 「未启用传输」提示只发一次
409    trans_warned: Mutex<bool>,
410}
411
412impl CtpPlugin {
413    /// 纯逻辑构造(native 尚未接传输)。
414    pub fn new(cfg: CtpConfig) -> Self {
415        let gw = cfg.gateway_name_or_default();
416        Self::with_parts(cfg, gw, None)
417    }
418
419    /// 注入传输层(native 桥或测试桩)。
420    pub fn with_transport(cfg: CtpConfig, trans: Box<dyn CtpTransport>) -> Self {
421        let gw = cfg.gateway_name_or_default();
422        Self::with_parts(cfg, gw, Some(trans))
423    }
424
425    fn with_parts(cfg: CtpConfig, gw: String, trans: Option<Box<dyn CtpTransport>>) -> Self {
426        let contracts = driver::contracts_empty();
427        let cfg_broker = cfg.broker_id.clone();
428        let cfg_user = cfg.td_user.clone();
429        let md_cfg = driver::MdConfig {
430            addr: cfg.md_addr.clone(),
431            user: cfg.md_user.clone(),
432            password: cfg.md_password.clone(),
433            broker: cfg.broker_id.clone(),
434        };
435        let td_cfg = driver::TdConfig {
436            addr: cfg.td_addr.clone(),
437            user: cfg.td_user.clone(),
438            password: cfg.td_password.clone(),
439            broker: cfg.broker_id.clone(),
440            auth_code: cfg.td_auth_code.clone(),
441            app_id: cfg.td_app_id.clone(),
442            product_classes: cfg.product_classes.clone(),
443        };
444        // 原稿 md.current_date 由 datetime.now() 起手;这里初值 = 启动时的上海日期
445        let md = Mutex::new(MdDriver::new(
446            &gw,
447            md_cfg,
448            contracts.clone(),
449            crate::map::shanghai_now_date(),
450        ));
451        let td = Mutex::new(TdDriver::new(&gw, td_cfg, contracts.clone()));
452        Self {
453            gw,
454            cfg: Mutex::new(cfg),
455            broker: Mutex::new(cfg_broker),
456            user: Mutex::new(cfg_user),
457            md,
458            td,
459            timer_count: Mutex::new(0),
460            trans,
461            ctx: OnceLock::new(),
462            trans_warned: Mutex::new(false),
463        }
464    }
465
466    /// native 构造:装配官方 SDK 传输层并完成回调接线(见 `native::transport` 文档)。
467    ///
468    /// 两相接线:① 先建回调入口 + 传输层(传输层持有 `Arc<Ingress>`);② 插件建好后
469    /// 回填 `Weak<CtpPlugin>` —— 必须用弱引用断环,否则 插件 ↔ 传输层 互相持有,
470    /// 两者永不析构(回调也不会停)。
471    ///
472    /// `flow_root` = CTP 流文件落盘根目录(D-FLOWPATH)。构造本身**不碰磁盘**:
473    /// 目录在真正连接时才按 `<flow_root>/<网关名小写>/{Md,Td}` 创建。
474    #[cfg(feature = "native")]
475    pub fn with_native(cfg: CtpConfig, flow_root: impl Into<std::path::PathBuf>) -> Arc<Self> {
476        let (trans, ingress) = crate::native::transport::native_parts(flow_root);
477        let plugin = Arc::new(Self::with_transport(cfg, trans));
478        crate::native::transport::attach_plugin(&ingress, &plugin);
479        plugin
480    }
481
482    /// 连接(原稿 `CtpGateway.connect` 162-190:先 TD 后 MD;`init_query` 只是挂秒针)。
483    ///
484    /// **连接触发点住在本插件里**(不放到核心/公共层,每个插件自持自己的连接):
485    /// 宿主 `start()` 会经 [`Plugin::on_start`] 调它一次;要显式控制(先配后连、
486    /// 断线后重连)也可以直接调。幂等由驱动守卫裁决 —— 原稿 MD 重复调用静默,
487    /// TD 重复调用转认证(776-777),两者都在 `MdDriver::connect`/`TdDriver::connect`。
488    ///
489    /// 原稿末尾的 `init_query` 只是把查询轮转注册到 `EVENT_TIMER`(237-240);
490    /// 本 crate 的对应面是宿主 eTimer → [`CtpPlugin::on_timer`],这里无需动作。
491    pub fn connect(&self) {
492        // 空地址守卫(0.3.0):td/md 前置都空 → 不瞎连(on_start 空配置形态),
493        // 等 `{gw}.configure` 手工注入后再发起(插件自带配置面,定盘口径)。
494        let (td_empty, md_empty) = {
495            let c = self.cfg.lock().unwrap();
496            (c.td_addr.is_empty(), c.md_addr.is_empty())
497        };
498        if td_empty && md_empty {
499            self.put_log(LogData {
500                gateway_name: self.gw.clone(),
501                msg: "未配置前置地址,跳过连接(等待 configure)".to_string(),
502                level: 20,
503            });
504            return;
505        }
506        // 先 TD 后 MD:与原稿 connect 的调用次序一致。
507        // 锁在各自语句结束时释放 —— emit 可能落到会阻塞的原生调用(如合约查询的
508        // 流控重发),绝不能在持驱动锁时做。
509        let td_outs = if td_empty {
510            Vec::new()
511        } else {
512            self.td.lock().unwrap().connect()
513        };
514        let md_outs = if md_empty {
515            Vec::new()
516        } else {
517            self.md.lock().unwrap().connect()
518        };
519        for o in td_outs {
520            self.emit_td(o);
521        }
522        for o in md_outs {
523            self.emit_md(o);
524        }
525    }
526
527    /// 传输层级失败上报:native 桥在回调面之外没有别的出口。
528    ///
529    /// 只用在本插件/原生层自己的故障上(建不出 API、建不出流目录)。原稿没有对应面
530    /// ——那里失败会直接抛异常,进程级可见;Rust 侧不能抛,所以落一条 ERROR 日志。
531    /// **请求返回码不在此列**:原稿除下单外一律丢弃返回码,那些日志会是原稿没有的噪音。
532    #[cfg(feature = "native")]
533    pub(crate) fn transport_log(&self, msg: String) {
534        self.put_log(LogData {
535            gateway_name: self.gw.clone(),
536            msg,
537            level: LEVEL_ERROR,
538        });
539    }
540
541    /// 原生适配入口(native 桥的回调直接调这四个):MD 回报步进 + 总线上抛。
542    pub fn on_md_event(&self, ev: MdEvent) {
543        let outs = self.md.lock().unwrap().step(ev);
544        for o in outs {
545            self.emit_md(o);
546        }
547    }
548
549    /// 原生适配入口(native 桥的回调直接调这四个):TD 回报步进 + 总线上抛。
550    pub fn on_td_event(&self, ev: TdEvent) {
551        let outs = self.td.lock().unwrap().step(ev);
552        for o in outs {
553            self.emit_td(o);
554        }
555    }
556
557    /// eTimer 一拍(原稿 gateway.process_timer_event):先做 1s 轻量判定
558    /// (0.7.0 轮转闸门的 3 秒兜底,必须按 1s 计,放进 2s gate 会退化成 4s)
559    /// → 2s gate 轮转查询 + 行情日期刷新。这样独立出来服务测试与 native 适配。
560    pub fn on_timer(&self) {
561        let outs = self.td.lock().unwrap().on_timer_tick();
562        for o in outs {
563            self.emit_td(o);
564        }
565        let gate_pass = {
566            let mut c = self.timer_count.lock().unwrap();
567            *c += 1;
568            if *c < 2 {
569                false
570            } else {
571                *c = 0;
572                true
573            }
574        };
575        if !gate_pass {
576            return;
577        }
578        // 原稿 231-235 次序:先 pop 查询函数执行,再 md_api.update_date
579        let outs = self.td.lock().unwrap().on_timer();
580        for o in outs {
581            self.emit_td(o);
582        }
583        let date = crate::map::shanghai_now_date();
584        self.md.lock().unwrap().update_date(date);
585    }
586
587    fn emit_md(&self, o: MdOut) {
588        match o {
589            MdOut::Log(l) => self.put_log(l),
590            MdOut::Tick(t) => self.put_data(W_TICK, t),
591            MdOut::ReqConnect { addr, flow_path } => {
592                if let Some(t) = &self.trans {
593                    t.md_connect(&addr, &flow_path);
594                } else {
595                    self.warn_no_transport();
596                }
597            }
598            MdOut::ReqLogin {
599                user,
600                password,
601                broker,
602            } => {
603                if let Some(t) = &self.trans {
604                    t.md_login(&user, &password, &broker);
605                } else {
606                    self.warn_no_transport();
607                }
608            }
609            MdOut::ReqSubscribe(symbols) => {
610                if let Some(t) = &self.trans {
611                    t.md_subscribe(&symbols);
612                } else {
613                    self.warn_no_transport();
614                }
615            }
616        }
617    }
618
619    fn emit_td(&self, o: TdOut) {
620        match o {
621            TdOut::Log(l) => self.put_log(l),
622            TdOut::Contract(c) => self.put_data(W_CONTRACT, c),
623            TdOut::Order(o) => self.put_data(W_ORDER, o),
624            TdOut::Trade(t) => self.put_data(W_TRADE, t),
625            TdOut::Position(p) => self.put_data(W_POSITION, p),
626            TdOut::Account(a) => self.put_data(W_ACCOUNT, a),
627            TdOut::PositionQueryDone(gw) => {
628                // 原稿 593:on_event(EVENT_POSITION_QUERY_DONE, self.gateway_name)
629                // → 裸键 "ePositionDone." + 载荷 gw(ek-oms 以此精确订阅)
630                if let Some(ctx) = self.ctx.get() {
631                    let _ = ctx.put(Event::with_data(W_POS_DONE, Arc::new(gw)));
632                }
633            }
634            // 原稿 EVENT_CONTRACT_FINISH = "eContractFinish":全局线无 gw 后缀;
635            // 载荷放网关名(多网关辨源,D-CFIN)。
636            TdOut::ContractFinish(gw) => {
637                if let Some(ctx) = self.ctx.get() {
638                    let _ = ctx.put(Event::with_data("eContractFinish", Arc::new(gw)));
639                }
640            }
641            TdOut::ReqConnect { addr, flow_path } => {
642                if let Some(t) = &self.trans {
643                    t.td_connect(&addr, &flow_path);
644                } else {
645                    self.warn_no_transport();
646                }
647            }
648            TdOut::ReqAuthenticate {
649                user,
650                broker,
651                auth_code,
652                app_id,
653            } => {
654                if let Some(t) = &self.trans {
655                    t.td_auth(&user, &broker, &auth_code, &app_id);
656                } else {
657                    self.warn_no_transport();
658                }
659            }
660            TdOut::ReqLogin {
661                user,
662                password,
663                broker,
664            } => {
665                if let Some(t) = &self.trans {
666                    t.td_login(&user, &password, &broker);
667                } else {
668                    self.warn_no_transport();
669                }
670            }
671            TdOut::ReqSettlementConfirm { broker, user } => {
672                if let Some(t) = &self.trans {
673                    t.td_settlement_confirm(&broker, &user);
674                } else {
675                    self.warn_no_transport();
676                }
677            }
678            TdOut::ReqQryInstrument => {
679                if let Some(t) = &self.trans {
680                    t.td_qry_instrument();
681                } else {
682                    self.warn_no_transport();
683                }
684            }
685            TdOut::ReqQryAccount { broker, user } => {
686                if let Some(t) = &self.trans {
687                    let n = t.td_qry_account(&broker, &user);
688                    for l in self.td.lock().unwrap().note_query_send_failure("资金", n) {
689                        self.put_log(l);
690                    }
691                } else {
692                    self.warn_no_transport();
693                }
694            }
695            TdOut::ReqQryPosition { broker, user } => {
696                if let Some(t) = &self.trans {
697                    let n = t.td_qry_position(&broker, &user);
698                    for l in self.td.lock().unwrap().note_query_send_failure("持仓", n) {
699                        self.put_log(l);
700                    }
701                } else {
702                    self.warn_no_transport();
703                }
704            }
705            TdOut::Insert(o) => {
706                if let Some(t) = &self.trans {
707                    t.td_insert(&o);
708                } else {
709                    self.warn_no_transport();
710                }
711            }
712            TdOut::Cancel(a) => {
713                if let Some(t) = &self.trans {
714                    t.td_cancel(&a);
715                } else {
716                    self.warn_no_transport();
717                }
718            }
719        }
720    }
721
722    /// 业务日志统一出口(LogData 挂 "eLog" 固定键;ctx 文档同款)。
723    fn put_log(&self, l: LogData) {
724        if let Some(ctx) = self.ctx.get() {
725            let _ = ctx.put(Event::with_data(W_LOG, Arc::new(l)));
726        }
727    }
728
729    /// 数据上抛(**裸线键 `eXxx.` + 载荷**;0.4.0 集成修正)。
730    ///
731    /// 原稿 emit 面带 gw 后缀(`eOrder.ctp`),但 ek 总线派发是精确匹配
732    /// (ek-event register 精确键,ek-core 照 routes 原样注册),而消费侧
733    /// dm/oms 注册的是裸前缀(`eTick.`/`eOrder.`)——带后缀的行**没有任何
734    /// 订阅者**,整条 ctp 数据流在装配里被静默丢弃(T5 实连揭出)。网关身份
735    /// 本就在载荷 `gateway_name` 字段里,消费侧按它分流足够;sim 侧同款裸键
736    /// 先例(ui_sim 全绿)钉死此契约。见 DESIGN D-EMIT。
737    fn put_data<T: Send + Sync + 'static>(&self, prefix: &str, payload: T) {
738        if let Some(ctx) = self.ctx.get() {
739            let _ = ctx.put(Event::with_data(prefix, Arc::new(payload)));
740        }
741    }
742
743    /// 无传输时的静默风告(只发一次;info 级防刷屏)。
744    fn warn_no_transport(&self) {
745        let mut warned = self.trans_warned.lock().unwrap();
746        if !*warned {
747            *warned = true;
748            self.put_log(LogData {
749                gateway_name: self.gw.clone(),
750                msg: "传输层未接入(默认构建):连接/查询/发送动作已被吸收".to_string(),
751                level: 20,
752            });
753        }
754    }
755
756    /// 服务面:`eReq.<gw>` 分派(arm 范式与 oms 同)。
757    fn on_request(&self, event: &Event) {
758        let Some(env) = event.data.downcast_ref::<Envelope>() else {
759            return;
760        };
761        macro_rules! arm {
762            ($ty:ty, $res:ty, $body:expr) => {
763                if let Some(req) = env.payload.downcast_ref::<$ty>() {
764                    let res: $res = $body(req);
765                    let _ = env.reply.send(Box::new(Ok::<$res, ReplyErr>(res)));
766                    return;
767                }
768            };
769        }
770        arm!(SubscribeReq, SubscribeReply, |r: &SubscribeReq| self
771            .do_subscribe(r));
772        arm!(SendOrderReq, SendOrderReply, |r: &SendOrderReq| self
773            .do_send_order(r));
774        arm!(CancelOrderReq, CancelOrderReply, |r: &CancelOrderReq| self
775            .do_cancel_order(r));
776        arm!(QueryRateReq, QueryRateReply, |r: &QueryRateReq| self
777            .do_query_rate(r));
778        arm!(ConfigureReq, ConfigureReply, |r: &ConfigureReq| self
779            .do_configure(r));
780        arm!(StatusReq, GatewayStatus, |r: &StatusReq| self.do_status(r));
781        arm!(DisconnectReq, DisconnectReply, |r: &DisconnectReq| self
782            .do_disconnect(r));
783        // 载荷不是任何已声明契约 → 协议错误(宿主既有语义)
784        let _ = env
785            .reply
786            .send(Box::new(Err::<(), ReplyErr>(ReplyErr::InvalidPayload)));
787    }
788
789    fn do_subscribe(&self, r: &SubscribeReq) -> SubscribeReply {
790        let mut outs = Vec::new();
791        {
792            let mut md = self.md.lock().unwrap();
793            for s in &r.symbols {
794                outs.extend(md.subscribe(s));
795            }
796        }
797        for o in outs {
798            self.emit_md(o);
799        }
800        SubscribeReply {
801            ok: true,
802            msg: String::new(),
803        }
804    }
805
806    fn do_send_order(&self, r: &SendOrderReq) -> SendOrderReply {
807        let outcome = {
808            let mut td = self.td.lock().unwrap();
809            td.send_order(&r.req)
810        };
811        if let Some(log) = &outcome.error {
812            self.put_log(log.clone());
813        }
814        // 原稿 844-848:插单命令的 rt 返回码由传输层裁决;n != 0 →
815        // 日志「委托请求发送失败,错误代码:n」且回 ""(不回 SUBMITTING 单)。
816        // 其余命令面照常下发。
817        let mut submit_order: Option<ek_data::order::OrderData> = None;
818        let mut send_failed: Option<String> = None;
819        for o in outcome.outs {
820            match o {
821                TdOut::Insert(input) => match &self.trans {
822                    Some(t) => {
823                        let n = t.td_insert(&input);
824                        if n != 0 {
825                            send_failed = Some(format!("委托请求发送失败,错误代码:{n}"));
826                        }
827                    }
828                    None => self.warn_no_transport(),
829                },
830                TdOut::Order(od) => submit_order = Some(od),
831                other => self.emit_td(other),
832            }
833        }
834        if let Some(msg) = send_failed {
835            let log = LogData {
836                gateway_name: self.gw.clone(),
837                msg: msg.clone(),
838                level: 20,
839            };
840            self.put_log(log);
841            return SendOrderReply {
842                ok: false,
843                en_orderid: String::new(),
844                msg,
845            };
846        }
847        if let Some(od) = submit_order {
848            self.emit_td(TdOut::Order(od));
849        }
850        if outcome.en_orderid.is_empty() {
851            let msg = outcome.error.map(|l| l.msg).unwrap_or_default();
852            SendOrderReply {
853                ok: false,
854                en_orderid: String::new(),
855                msg,
856            }
857        } else if self.trans.is_none() {
858            self.warn_no_transport();
859            SendOrderReply {
860                ok: false,
861                en_orderid: outcome.en_orderid,
862                msg: "传输层未接入(默认构建)".to_string(),
863            }
864        } else {
865            SendOrderReply {
866                ok: true,
867                en_orderid: outcome.en_orderid,
868                msg: String::new(),
869            }
870        }
871    }
872
873    /// 费率查询实现(附加接口面)。margin → `td_qry_margin_rate`,
874    /// commission → `td_qry_commission_rate`;status 无 Req,不在此线
875    /// (纯被动,官文见 DESIGN)。RT 非零命中原稿式发送失败日志并回值。
876    fn do_query_rate(&self, r: &QueryRateReq) -> QueryRateReply {
877        let broker = self.broker.lock().unwrap().clone();
878        let user = self.user.lock().unwrap().clone();
879        let is_margin = r.kind == "margin";
880        let is_commission = r.kind == "commission";
881        if !is_margin && !is_commission {
882            self.put_log(LogData {
883                gateway_name: self.gw.clone(),
884                msg: format!("费率查询类型未知:{}", r.kind),
885                level: 20,
886            });
887            return QueryRateReply {
888                ok: false,
889                msg: format!("费率查询类型未知:{}", r.kind),
890            };
891        }
892        if let Some(t) = &self.trans {
893            let n = if is_margin {
894                let query = crate::wire::QryInstrumentMarginRate {
895                    broker_id: broker.clone(),
896                    investor_id: user.clone(),
897                    hedge_flag: r.hedge_flag.clone(),
898                    exchange_id: r.exchange_id.clone(),
899                    invest_unit_id: String::new(),
900                    instrument_id: r.instrument_id.clone(),
901                };
902                t.td_qry_margin_rate(&query)
903            } else {
904                let query = crate::wire::QryInstrumentCommissionRate {
905                    broker_id: broker.clone(),
906                    investor_id: user.clone(),
907                    exchange_id: r.exchange_id.clone(),
908                    invest_unit_id: String::new(),
909                    instrument_id: r.instrument_id.clone(),
910                };
911                t.td_qry_commission_rate(&query)
912            };
913            let label = if is_margin {
914                "保证金率"
915            } else {
916                "手续费率"
917            };
918            if n != 0 {
919                // 同会话查询节律 1 次/秒(eTimer 轮转同用每秒名额):RT != 0 时隔
920                // 1.05s 重试一次(对齐示例第 11 步与 D-FLCN 的节拍纪律)。
921                std::thread::sleep(std::time::Duration::from_millis(1050));
922                let n2 = if is_margin {
923                    let query = crate::wire::QryInstrumentMarginRate {
924                        broker_id: broker.clone(),
925                        investor_id: user.clone(),
926                        hedge_flag: r.hedge_flag.clone(),
927                        exchange_id: r.exchange_id.clone(),
928                        invest_unit_id: String::new(),
929                        instrument_id: r.instrument_id.clone(),
930                    };
931                    t.td_qry_margin_rate(&query)
932                } else {
933                    let query = crate::wire::QryInstrumentCommissionRate {
934                        broker_id: broker.clone(),
935                        investor_id: user.clone(),
936                        exchange_id: r.exchange_id.clone(),
937                        invest_unit_id: String::new(),
938                        instrument_id: r.instrument_id.clone(),
939                    };
940                    t.td_qry_commission_rate(&query)
941                };
942                let n = n2;
943                if n != 0 {
944                    let msg = format!("{label}查询请求发送失败,错误代码:{n}(重发一次仍拒)");
945                    self.put_log(LogData {
946                        gateway_name: self.gw.clone(),
947                        msg: msg.clone(),
948                        level: 20,
949                    });
950                    return QueryRateReply { ok: false, msg };
951                }
952                QueryRateReply {
953                    ok: true,
954                    msg: format!("{label}查询首次被拒(错误代码另附),重发后受理"),
955                }
956            } else {
957                QueryRateReply {
958                    ok: true,
959                    msg: String::new(),
960                }
961            }
962        } else {
963            self.warn_no_transport();
964            QueryRateReply {
965                ok: false,
966                msg: "传输层未接入(默认构建)".to_string(),
967            }
968        }
969    }
970
971    fn do_cancel_order(&self, r: &CancelOrderReq) -> CancelOrderReply {
972        let outs = {
973            let mut td = self.td.lock().unwrap();
974            td.cancel_order(&r.req)
975        };
976        let mut rejected_log: Option<LogData> = None;
977        let outs: Vec<TdOut> = outs
978            .into_iter()
979            .filter(|o| match o {
980                TdOut::Log(l) => {
981                    rejected_log = Some(l.clone());
982                    false
983                }
984                _ => true,
985            })
986            .collect();
987        if let Some(l) = rejected_log {
988            let msg = l.msg.clone();
989            self.put_log(l);
990            return CancelOrderReply { ok: false, msg };
991        }
992        for o in outs {
993            self.emit_td(o);
994        }
995        if self.trans.is_none() {
996            self.warn_no_transport();
997            return CancelOrderReply {
998                ok: false,
999                msg: "传输层未接入(默认构建)".to_string(),
1000            };
1001        }
1002        CancelOrderReply {
1003            ok: true,
1004            msg: String::new(),
1005        }
1006    }
1007
1008    /// 配置实现(插件自带配置面,0.3.0;定盘:界面/宿主手工输入经 ask,
1009    /// 插件不读配置文件)。
1010    ///
1011    /// 校验 → 应用(驱动 reconfigure 复位连接/认证/登录守卫 + 插件侧
1012    /// cfg/broker/user 换血)→ **强制重连**(0.5.0:先 `td_release`/`md_release`
1013    /// 断开旧会话——传输层幂等 connect 下,不断开就没有新登录显形,换账号/
1014    /// 改密码永不生效;Release 归 RAII Drop,可能阻塞到回调收尾)→ `connect()`
1015    /// (先 TD 后 MD)。纯逻辑面 release 默认空,行为不变。
1016    fn do_configure(&self, r: &ConfigureReq) -> ConfigureReply {
1017        self.configure(r)
1018    }
1019
1020    fn do_status(&self, _r: &StatusReq) -> GatewayStatus {
1021        self.status()
1022    }
1023
1024    fn do_disconnect(&self, _r: &DisconnectReq) -> DisconnectReply {
1025        self.disconnect()
1026    }
1027
1028    /// 状态口(0.9.0 新增;公开与 `configure` 同级;ask 臂 `{gw}.status` 走它)。
1029    ///
1030    /// 纪律:**只读**——不改任何守卫、不触发重连、不写日志(界面按秒级刷,
1031    /// 每次刷都写日志会把 eLog 环冲爆,512 行取证环就是这么饱和的)。
1032    pub fn status(&self) -> GatewayStatus {
1033        let td = self.td.lock().unwrap().status_view();
1034        let md = self.md.lock().unwrap().status_view();
1035        GatewayStatus {
1036            gateway: self.gw.clone(),
1037            sampled_ms: crate::map::now_epoch_ms(),
1038            td_handle_up: td.handle_up,
1039            td_logged_in: td.logged_in,
1040            td_login_failed: td.login_failed,
1041            td_auth_ok: td.auth_ok,
1042            td_auth_failed: td.auth_failed,
1043            td_front_id: td.front_id,
1044            td_session_id: td.session_id,
1045            td_disconnects: td.disconnects,
1046            td_last_login_ok_ms: td.last_login_ok_ms,
1047            td_last_account_ms: td.last_account_ms,
1048            td_last_position_ms: td.last_position_ms,
1049            td_query_rejects: td.query_rejects,
1050            td_query_reject_streak: td.query_reject_streak,
1051            td_instrument_rejects: td.instrument_rejects,
1052            td_rotation_armed: td.rotation_armed,
1053            td_last_error: td.last_error,
1054            md_handle_up: md.handle_up,
1055            md_logged_in: md.logged_in,
1056            md_disconnects: md.disconnects,
1057            md_last_login_ok_ms: md.last_login_ok_ms,
1058            md_last_tick_ms: md.last_tick_ms,
1059            md_sub_rejects: md.sub_rejects,
1060            md_last_error: md.last_error,
1061            md_subscribed: md.subscribed,
1062        }
1063    }
1064
1065    /// 断开口(0.9.0 新增;裁定 1「断开 = 停会话,实例与订阅保留」)。
1066    ///
1067    /// 放掉:两侧会话(传输层 release + 驱动守卫复位,轮转闸门关闭)。
1068    /// 保留:插件实例、账号配置、订阅集、断开计数与「最近时刻」、持仓合并缓存、
1069    /// OrderSysID 映射。恢复走『连接』(= `configure` 重走全链,登录后自动重订阅)。
1070    pub fn disconnect(&self) -> DisconnectReply {
1071        if let Some(t) = &self.trans {
1072            t.td_release();
1073            t.md_release();
1074        }
1075        {
1076            self.td.lock().unwrap().reset_session();
1077        }
1078        {
1079            self.md.lock().unwrap().reset_session();
1080        }
1081        self.put_log(LogData {
1082            gateway_name: self.gw.clone(),
1083            msg: "会话已停(实例与订阅保留;『连接』可恢复)".to_string(),
1084            level: LEVEL_INFO,
1085        });
1086        DisconnectReply {
1087            ok: true,
1088            msg: "会话已停(实例与订阅保留;『连接』可恢复)".to_string(),
1089        }
1090    }
1091
1092    /// 当前生效的产品类勾选(D-PC-FILTER)。
1093    ///
1094    /// 校验被拒时也回带它,而不是空表 —— 空表会让界面把勾选框清空,而网关其实还跑在
1095    /// 旧勾选上,那是假信息。
1096    fn effective_classes(&self) -> Vec<String> {
1097        self.td.lock().unwrap().product_classes()
1098    }
1099
1100    /// 配置口(公开,与 `connect` 同级的显式控制面;ask 臂 `{gw}.configure`
1101    /// 走的也是它)。
1102    pub fn configure(&self, r: &ConfigureReq) -> ConfigureReply {
1103        if r.td_addr.is_empty() && r.md_addr.is_empty() {
1104            return ConfigureReply {
1105                ok: false,
1106                msg: "配置校验: td_addr/md_addr 至少一个非空".to_string(),
1107                product_classes: self.effective_classes(),
1108            };
1109        }
1110        if !r.td_addr.is_empty()
1111            && (r.broker_id.is_empty() || r.td_user.is_empty() || r.td_password.is_empty())
1112        {
1113            return ConfigureReply {
1114                ok: false,
1115                msg: "配置校验: TD 侧需 broker_id/td_user/td_password 齐".to_string(),
1116                product_classes: self.effective_classes(),
1117            };
1118        }
1119        let md_user = if !r.md_user.is_empty() {
1120            r.md_user.clone()
1121        } else {
1122            r.td_user.clone()
1123        };
1124        let md_password = if !r.md_password.is_empty() {
1125            r.md_password.clone()
1126        } else {
1127            r.td_password.clone()
1128        };
1129        if !r.md_addr.is_empty() && (md_user.is_empty() || md_password.is_empty()) {
1130            return ConfigureReply {
1131                ok: false,
1132                msg: "配置校验: MD 侧需 md_user/md_password(或 td 侧回落)".to_string(),
1133                product_classes: self.effective_classes(),
1134            };
1135        }
1136        // 产品类勾选校验(D-PC-FILTER):CTP 产品类码是单字符。非法项不静默接受,
1137        // 直接拒并列出;空串项视为「没填」,由归一化忽略(全空 → 回落默认期货)。
1138        let bad_codes: Vec<String> = r
1139            .product_classes
1140            .iter()
1141            .map(|c| c.trim().to_string())
1142            .filter(|c| !c.is_empty() && c.chars().count() != 1)
1143            .collect();
1144        if !bad_codes.is_empty() {
1145            return ConfigureReply {
1146                ok: false,
1147                msg: format!(
1148                    "配置校验: product_classes 需为单字符 CTP 码(非法项:{})",
1149                    bad_codes.join(",")
1150                ),
1151                product_classes: self.effective_classes(),
1152            };
1153        }
1154        let classes = crate::map::ProductClasses(r.product_classes.clone());
1155        let effective_classes = classes.normalize();
1156        {
1157            self.td.lock().unwrap().reconfigure(driver::TdConfig {
1158                addr: r.td_addr.clone(),
1159                user: r.td_user.clone(),
1160                password: r.td_password.clone(),
1161                broker: r.broker_id.clone(),
1162                auth_code: r.td_auth_code.clone(),
1163                app_id: r.td_app_id.clone(),
1164                product_classes: classes.clone(),
1165            });
1166        }
1167        {
1168            self.md.lock().unwrap().reconfigure(driver::MdConfig {
1169                addr: r.md_addr.clone(),
1170                user: md_user,
1171                password: md_password,
1172                broker: r.broker_id.clone(),
1173            });
1174        }
1175        *self.broker.lock().unwrap() = r.broker_id.clone();
1176        *self.user.lock().unwrap() = r.td_user.clone();
1177        *self.cfg.lock().unwrap() = CtpConfig {
1178            gateway_name: self.gw.clone(),
1179            broker_id: r.broker_id.clone(),
1180            td_addr: r.td_addr.clone(),
1181            td_user: r.td_user.clone(),
1182            td_password: r.td_password.clone(),
1183            td_auth_code: r.td_auth_code.clone(),
1184            td_app_id: r.td_app_id.clone(),
1185            md_addr: r.md_addr.clone(),
1186            md_user: r.md_user.clone(),
1187            md_password: r.md_password.clone(),
1188            product_classes: classes,
1189        };
1190        // 强制重连(0.5.0):旧会话先断,connect() 才会新建句柄走全链登录。
1191        if let Some(t) = &self.trans {
1192            t.td_release();
1193            t.md_release();
1194        }
1195        self.connect();
1196        ConfigureReply {
1197            ok: true,
1198            msg: format!(
1199                "已配置并发起连接(先 TD 后 MD);收录产品类:{}",
1200                effective_classes.join(" ")
1201            ),
1202            product_classes: effective_classes,
1203        }
1204    }
1205}
1206
1207impl Default for CtpPlugin {
1208    fn default() -> Self {
1209        Self::new(CtpConfig::default())
1210    }
1211}
1212
1213impl Plugin for CtpPlugin {
1214    fn name(&self) -> &str {
1215        &self.gw
1216    }
1217
1218    fn routes(&self) -> Vec<Cow<'static, str>> {
1219        vec![Cow::Owned(e_req_type(&self.gw)), Cow::Borrowed(ROUTE_TIMER)]
1220    }
1221
1222    fn version(&self) -> Cow<'static, str> {
1223        Cow::Borrowed(env!("CARGO_PKG_VERSION"))
1224    }
1225
1226    fn services(&self) -> Vec<ServiceDesc> {
1227        vec![
1228            ServiceDesc::describe::<SubscribeReq, SubscribeReply>(
1229                format!("{}.subscribe", self.gw),
1230                "登记并(已登录时)下发行情订阅",
1231            ),
1232            ServiceDesc::describe::<SendOrderReq, SendOrderReply>(
1233                format!("{}.send_order", self.gw),
1234                "下单(InputOrder + SUBMITTING 单已上总线)",
1235            ),
1236            ServiceDesc::describe::<CancelOrderReq, CancelOrderReply>(
1237                format!("{}.cancel_order", self.gw),
1238                "撤单(三段订单号解析)",
1239            ),
1240            ServiceDesc::describe::<QueryRateReq, QueryRateReply>(
1241                format!("{}.query_rate", self.gw),
1242                "费率查询(margin=保证金率/commission=手续费率;状态为纯被动推送)",
1243            ),
1244            ServiceDesc::describe::<ConfigureReq, ConfigureReply>(
1245                format!("{}.configure", self.gw),
1246                "网关配置(插件自带配置面;手工输入账号/前置 → 复位守卫并发起连接)",
1247            ),
1248            ServiceDesc::describe::<StatusReq, GatewayStatus>(
1249                format!("{}.status", self.gw),
1250                "状态读数(0.9.0;登录态 / 断开计数 / 查询被拒计数 / 最近回报时刻;只读、不写日志)",
1251            ),
1252            ServiceDesc::describe::<DisconnectReq, DisconnectReply>(
1253                format!("{}.disconnect", self.gw),
1254                "停会话(0.9.0;实例与订阅保留,『连接』可恢复)",
1255            ),
1256        ]
1257    }
1258
1259    fn on_init(&self, ctx: &PluginContext) {
1260        let _ = self.ctx.set(ctx.clone());
1261    }
1262
1263    fn on_start(&self) {
1264        // 传输层已接入 → 按配置发起连接(原稿由 app 显式调 CtpGateway.connect)。
1265        // 无传输层的纯逻辑构建保持原状:emit 走 warn_no_transport 提示一次。
1266        self.connect();
1267    }
1268
1269    fn on_stop(&self) {
1270        // 先释放原生会话:之后 SDK 不再回调,回调入口才会随插件一起被放下。
1271        if let Some(t) = &self.trans {
1272            t.shutdown();
1273        }
1274    }
1275
1276    fn on_mail(&self, event: &Event) {
1277        if event.event_type == ROUTE_TIMER {
1278            self.on_timer();
1279            return;
1280        }
1281        if event.event_type == e_req_type(&self.gw) {
1282            self.on_request(event);
1283        }
1284    }
1285}