1use 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#[cfg(feature = "native")]
29use ek_data::log::LEVEL_ERROR;
30use 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
37const 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.";
45const W_POS_DONE: &str = "ePositionDone.";
48use ek_data::event::EVENT_LOG as W_LOG;
50pub use ek_event::EVENT_TIMER as ROUTE_TIMER;
53
54#[derive(Debug, Clone)]
58pub struct SubscribeReq {
59 pub symbols: Vec<String>,
61}
62
63#[derive(Debug, Clone, PartialEq, Eq)]
65pub struct SubscribeReply {
66 pub ok: bool,
68 pub msg: String,
70}
71
72#[derive(Debug, Clone)]
74pub struct SendOrderReq {
75 pub req: OrderRequest,
77}
78
79#[derive(Debug, Clone, PartialEq)]
81pub struct SendOrderReply {
82 pub ok: bool,
84 pub en_orderid: String,
86 pub msg: String,
88}
89
90#[derive(Debug, Clone)]
92pub struct CancelOrderReq {
93 pub req: CancelRequest,
95}
96
97#[derive(Debug, Clone, PartialEq, Eq)]
99pub struct CancelOrderReply {
100 pub ok: bool,
102 pub msg: String,
104}
105
106#[derive(Debug, Clone)]
111pub struct QueryRateReq {
112 pub kind: String,
114 pub instrument_id: String,
116 pub hedge_flag: String,
118 pub exchange_id: String,
120}
121
122#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct QueryRateReply {
125 pub ok: bool,
127 pub msg: String,
129}
130
131#[derive(Debug, Clone, Default)]
138pub struct ConfigureReq {
139 pub broker_id: String,
141 pub td_addr: String,
143 pub td_user: String,
145 pub td_password: String,
147 pub td_auth_code: String,
149 pub td_app_id: String,
151 pub md_addr: String,
153 pub md_user: String,
155 pub md_password: String,
157 pub product_classes: Vec<String>,
160}
161
162#[derive(Debug, Clone, PartialEq, Eq)]
164pub struct ConfigureReply {
165 pub ok: bool,
167 pub msg: String,
169 pub product_classes: Vec<String>,
173}
174
175#[derive(Debug, Clone, Default, PartialEq, Eq)]
177pub struct StatusReq;
178
179#[derive(Debug, Clone, Default, PartialEq, Eq)]
191pub struct GatewayStatus {
192 pub gateway: String,
194 pub sampled_ms: i64,
196 pub td_handle_up: bool,
199 pub td_logged_in: bool,
201 pub td_login_failed: bool,
203 pub td_auth_ok: bool,
205 pub td_auth_failed: bool,
207 pub td_front_id: i64,
209 pub td_session_id: i64,
211 pub td_disconnects: u32,
213 pub td_last_login_ok_ms: Option<i64>,
215 pub td_last_account_ms: Option<i64>,
217 pub td_last_position_ms: Option<i64>,
219 pub td_query_rejects: u32,
221 pub td_query_reject_streak: u32,
223 pub td_instrument_rejects: u32,
225 pub td_rotation_armed: bool,
227 pub td_last_error: Option<String>,
229 pub md_handle_up: bool,
232 pub md_logged_in: bool,
234 pub md_disconnects: u32,
236 pub md_last_login_ok_ms: Option<i64>,
238 pub md_last_tick_ms: Option<i64>,
240 pub md_sub_rejects: u32,
242 pub md_last_error: Option<String>,
244 pub md_subscribed: usize,
246}
247
248#[derive(Debug, Clone, Default, PartialEq, Eq)]
250pub struct DisconnectReq;
251
252#[derive(Debug, Clone, PartialEq, Eq)]
254pub struct DisconnectReply {
255 pub ok: bool,
257 pub msg: String,
259}
260
261pub trait CtpTransport: Send + Sync {
265 fn md_connect(&self, addr: &str, flow_path: &str);
267 fn md_login(&self, user: &str, password: &str, broker: &str);
269 fn md_subscribe(&self, symbols: &[String]);
271 fn td_connect(&self, addr: &str, flow_path: &str);
273 fn td_auth(&self, user: &str, broker: &str, auth_code: &str, app_id: &str);
275 fn td_login(&self, user: &str, password: &str, broker: &str);
277 fn td_settlement_confirm(&self, broker: &str, user: &str);
279 fn td_qry_instrument(&self);
281 fn td_qry_account(&self, broker: &str, user: &str) -> i32;
285 fn td_qry_position(&self, broker: &str, user: &str) -> i32;
287 fn td_insert(&self, order: &InputOrder) -> i32;
290 fn td_cancel(&self, action: &crate::wire::InputOrderAction) -> i32;
292 fn td_qry_margin_rate(&self, query: &crate::wire::QryInstrumentMarginRate) -> i32;
294 fn td_qry_commission_rate(&self, query: &crate::wire::QryInstrumentCommissionRate) -> i32;
296
297 fn shutdown(&self) {}
305
306 fn td_release(&self) {}
309 fn md_release(&self) {}
311}
312
313#[derive(Debug, Clone, Default)]
319pub struct CtpConfig {
320 pub gateway_name: String,
322 pub broker_id: String,
324 pub td_addr: String,
326 pub td_user: String,
328 pub td_password: String,
330 pub td_auth_code: String,
332 pub td_app_id: String,
334 pub md_addr: String,
336 pub md_user: String,
338 pub md_password: String,
340 pub product_classes: crate::map::ProductClasses,
343}
344
345impl CtpConfig {
346 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#[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 if name.len() > 32 {
376 return None;
377 }
378 Some(name)
379}
380
381fn 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
390pub struct CtpPlugin {
394 gw: String,
395 cfg: Mutex<CtpConfig>,
397 broker: Mutex<String>,
399 user: Mutex<String>,
401 md: Mutex<MdDriver>,
402 td: Mutex<TdDriver>,
403 timer_count: Mutex<u32>,
405 trans: Option<Box<dyn CtpTransport>>,
407 ctx: OnceLock<PluginContext>,
408 trans_warned: Mutex<bool>,
410}
411
412impl CtpPlugin {
413 pub fn new(cfg: CtpConfig) -> Self {
415 let gw = cfg.gateway_name_or_default();
416 Self::with_parts(cfg, gw, None)
417 }
418
419 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 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 #[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 pub fn connect(&self) {
492 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 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 #[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 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 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 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 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 if let Some(ctx) = self.ctx.get() {
631 let _ = ctx.put(Event::with_data(W_POS_DONE, Arc::new(gw)));
632 }
633 }
634 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 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 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 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 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 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 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 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 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 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 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 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 fn effective_classes(&self) -> Vec<String> {
1097 self.td.lock().unwrap().product_classes()
1098 }
1099
1100 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 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 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 self.connect();
1267 }
1268
1269 fn on_stop(&self) {
1270 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}