1use std::sync::OnceLock;
21use std::time::Duration;
22
23use ek_core::context::PluginContext;
24use ek_core::plugin::Plugin;
25use ek_core::service::{PlugNotice, ServiceDesc};
26use ek_core::{EVENT_PLUG_DOWN, EVENT_PLUG_UP};
27use ek_ctp::plugin::{
28 ConfigureReply, ConfigureReq, DisconnectReply, DisconnectReq, GatewayStatus, StatusReq,
29 SubscribeReply as CtpSubscribeReply, SubscribeReq as CtpSubscribeReq,
30};
31use ek_data::account::AccountData;
32use ek_data::event::{EVENT_BAR, EVENT_CONTRACT, EVENT_LOG, EVENT_TICK};
33use ek_data::log::LogData;
34use ek_data::order::OrderData;
35use ek_data::position::PositionData;
36use ek_data::trade::TradeData;
37use ek_dm::plugin::{EVENT_BAR_UPDATE, ReloadReq};
38use ek_event::EVENT_TIMER;
39use ek_oms::plugin::{
40 GetAllAccountsReq, GetAllOrdersReq, GetAllPositionsReq, GetAllTradesReq, OmsCancelOrderReq,
41 OmsCancelReply, PipelineStatusReq, SubmitOrderReq,
42};
43use ek_oms::{SubmitReply, SubmitStatus};
44use ek_station::wire::{
45 ListReq, PlugRow, RemoveReq, RemoveRes, SVC_LIST, SVC_REMOVE, SVC_START, StartReq, StartRes,
46};
47
48use ek_cfg::wire::{
56 DeleteReq as CfgDeleteReq, DeleteRes as CfgDeleteRes, ListReq as CfgListReq,
57 ListRes as CfgListRes,
58};
59
60fn cfg_svc_ids() -> (String, String, String) {
63 let descs = ek_cfg::wire::cfg_services();
64 let id = |i: usize| -> String {
65 descs
66 .get(i)
67 .map_or_else(|| format!("cfg.svc[{i}]"), |d| d.id.clone())
68 };
69 (id(0), id(1), id(2))
70}
71
72fn ui_cfg_store() -> ek_cfg::Store {
76 match ek_cfg::Store::eco() {
77 Ok(s) => s,
78 Err(e) => {
79 eprintln!("ek-cfg:运维修复根 eco 失败({e}),回落仓内 ./ekcfg-local(显形口径)");
80 ek_cfg::Store::with_root(std::path::PathBuf::from("ekcfg-local"))
81 }
82 }
83}
84
85pub const THEME_NS: &str = "ui";
93pub const THEME_KEY: &str = "theme";
95
96#[must_use]
99pub fn load_theme() -> crate::tokens::Theme {
100 match ui_cfg_store().load(THEME_NS) {
101 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map
102 .get_plain(THEME_KEY)
103 .map(crate::tokens::Theme::from_str)
104 .unwrap_or_default(),
105 Ok(ek_cfg::LoadOutcome::Fresh) => crate::tokens::Theme::default(),
106 Err(e) => {
107 eprintln!("ek-cfg:读主题档失败({e}),按缺省深色起界面");
108 crate::tokens::Theme::default()
109 }
110 }
111}
112
113fn save_theme(theme: crate::tokens::Theme) -> Result<(), String> {
118 let store = ui_cfg_store();
119 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
120 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
121 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
122 };
123 map = map.set(THEME_KEY, ek_core::Field::plain(theme.as_str()));
124 store.store(THEME_NS, &map).map_err(|e| e.to_string())
125}
126
127pub fn save_auto_connect_flags_into(
135 store: &ek_cfg::Store,
136 flags: &[(String, bool)],
137) -> Result<(), String> {
138 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
139 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
140 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
141 };
142 let stale: Vec<String> = map
143 .iter()
144 .filter(|(k, _)| k.starts_with("auto_connect."))
145 .map(|(k, _)| (*k).to_string())
146 .collect();
147 for k in stale {
148 map.remove(&k);
149 }
150 for (gw, on) in flags {
151 map = map.set(
152 format!("auto_connect.{gw}"),
153 ek_core::Field::plain(if *on { "1" } else { "0" }),
154 );
155 }
156 store.store(THEME_NS, &map).map_err(|e| e.to_string())
157}
158
159pub fn save_account_aliases_into(
164 store: &ek_cfg::Store,
165 aliases: &[(String, String)],
166) -> Result<(), String> {
167 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
168 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
169 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
170 };
171 let stale: Vec<String> = map
172 .iter()
173 .filter(|(k, _)| k.starts_with("alias."))
174 .map(|(k, _)| (*k).to_string())
175 .collect();
176 for k in stale {
177 map.remove(&k);
178 }
179 for (gw, alias) in aliases {
180 let alias = alias.trim();
181 if alias.is_empty() {
182 continue;
183 }
184 map = map.set(format!("alias.{gw}"), ek_core::Field::plain(alias));
185 }
186 store.store(THEME_NS, &map).map_err(|e| e.to_string())
187}
188
189use crate::state::{BusSnap, UiShared};
190
191pub const UI_NAME: &str = "ui";
193
194pub const UI_MAILBOX_SOFT_CAP: u64 = 100_000;
203
204const W_ORDER: &str = ek_oms::wire::EVENT_ORDER;
207const W_TRADE: &str = ek_oms::wire::EVENT_TRADE;
208const W_POSITION: &str = ek_oms::wire::EVENT_POSITION;
209const W_ACCOUNT: &str = ek_oms::wire::EVENT_ACCOUNT;
210
211pub struct UiPlugin {
213 shared: std::sync::Arc<UiShared>,
214 ctx: OnceLock<PluginContext>,
215 cmd_stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
217}
218
219impl UiPlugin {
220 pub fn new(shared: std::sync::Arc<UiShared>) -> Self {
222 Self {
223 shared,
224 ctx: OnceLock::new(),
225 cmd_stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
226 }
227 }
228
229 fn refresh(&self) {
232 let Some(ctx) = self.ctx.get() else {
233 return; };
235 if self.shared.list_poll_due() {
240 self.poll_list(ctx);
241 }
242 self.seed_status(ctx);
243 if self.shared.take_pipeline_due() {
244 self.poll_pipeline(ctx);
245 }
246 if self.shared.take_cfg_due() {
247 self.poll_cfg(ctx);
250 }
251 self.flush_persist();
252 }
253
254 fn flush_persist(&self) {
260 if self.shared.take_theme_persist_due() {
261 let theme = self.shared.theme();
263 match save_theme(theme) {
264 Ok(()) => self
265 .shared
266 .log_line(format!("[ui] 主题已存(ui.theme={})", theme.as_str())),
267 Err(e) => self
268 .shared
269 .log_line(format!("[ui] 主题落盘失败:{e}(本次切换只在本次运行内生效)")),
270 }
271 }
272 let purges = self.shared.take_archive_purges();
273 for gw in purges {
274 match crate::appconfig::purge_account_archive(&ui_cfg_store(), &gw) {
279 Ok(()) => self.shared.log_line(format!(
280 "[账户] 档案段 ctp.{gw} 已抹(删除已落盘:下次启动清点不再复活)"
281 )),
282 Err(e) => self.shared.log_line(format!(
283 "[账户] 档案 ctp.{gw} 抹除失败:{e}(本次删除只在今次运行生效,重启会复活)"
284 )),
285 }
286 }
287 if self.shared.take_auto_connect_persist_due() {
288 let flags: Vec<(String, bool)> = self
291 .shared
292 .accounts()
293 .iter()
294 .map(|a| (a.gateway.clone(), a.auto_connect))
295 .collect();
296 match save_auto_connect_flags_into(&ui_cfg_store(), &flags) {
297 Ok(()) => {
298 let on: Vec<&str> = flags
299 .iter()
300 .filter(|(_, on)| *on)
301 .map(|(gw, _)| gw.as_str())
302 .collect();
303 if on.is_empty() {
304 self.shared
305 .log_line("[ui] 自动连接勾选已存(ui 段;当前全部关)".to_string());
306 } else {
307 self.shared.log_line(format!(
308 "[ui] 自动连接勾选已存(ui 段;开:{};守护/桌面启动清点按它起连)",
309 on.join(", ")
310 ));
311 }
312 }
313 Err(e) => self.shared.log_line(format!(
314 "[ui] 自动连接勾选落盘失败:{e}(本轮勾选只在本次运行内生效)"
315 )),
316 }
317 }
318 if self.shared.take_alias_persist_due() {
319 let aliases: Vec<(String, String)> = self
321 .shared
322 .accounts()
323 .iter()
324 .map(|a| (a.gateway.clone(), a.alias.clone()))
325 .collect();
326 match save_account_aliases_into(&ui_cfg_store(), &aliases) {
327 Ok(()) => {
328 let named: Vec<String> = aliases
329 .iter()
330 .filter(|(_, al)| !al.trim().is_empty())
331 .map(|(gw, al)| format!("{al}({gw})"))
332 .collect();
333 if named.is_empty() {
334 self.shared
335 .log_line("[ui] 账户别名已存(ui 段;当前全部无别名)".to_string());
336 } else {
337 self.shared.log_line(format!("[ui] 账户别名已存(ui 段;{};启动清点按它贴标签)", named.join(", ")));
338 }
339 }
340 Err(e) => self.shared.log_line(format!(
341 "[ui] 账户别名落盘失败:{e}(本次改名只在本次运行内生效)"
342 )),
343 }
344 }
345 }
346
347 fn poll_cfg(&self, ctx: &PluginContext) {
349 let (_, svc_list, _) = cfg_svc_ids();
350 match ctx.ask_service::<CfgListReq, CfgListRes>(
351 svc_list.as_str(),
352 CfgListReq,
353 Duration::from_secs(2),
354 ) {
355 Ok(res) => {
356 self.shared.set_cfg_briefs(res.entries);
357 if !res.msg.is_empty() {
358 self.shared
359 .log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
360 }
361 }
362 Err(e) => self
363 .shared
364 .log_line(format!("[cfg] 段落目录失败(cfg 缺席或超时): {e:?}")),
365 }
366 }
367
368 fn poll_list(&self, ctx: &PluginContext) {
374 match ctx.ask_service::<ListReq, Vec<PlugRow>>(SVC_LIST, ListReq, Duration::from_secs(2)) {
375 Ok(rows) => {
376 let mut services = Vec::with_capacity(rows.len());
377 for r in &rows {
378 let ids = ctx
379 .services_of(&r.name)
380 .map(|v| v.iter().map(|s| s.id.clone()).collect::<Vec<_>>())
381 .unwrap_or_default();
382 services.push((r.name.clone(), ids));
383 }
384 self.shared.sync_accounts_connected(&rows);
387 self.shared.note_plugin_spill(&rows);
390 self.shared.set_snapshot(BusSnap { rows, services });
391 }
392 Err(e) => self.shared.log_line(format!("[ui] svctl.list 失败: {e}")),
393 }
394 }
395
396 fn seed_status(&self, ctx: &PluginContext) {
403 for gw in self.shared.take_status_seeds() {
404 let svc = format!("{gw}.{}", ek_ctp::SVC_STATUS);
405 match ctx.ask_service::<StatusReq, GatewayStatus>(
406 &svc,
407 StatusReq,
408 Duration::from_secs(2),
409 ) {
410 Ok(st) => self.shared.note_gateway_status(&gw, &st),
411 Err(e) => {
412 self.shared.note_status_unreachable(&gw, &format!("{e}"));
416 }
417 }
418 }
419 }
420
421 fn poll_pipeline(&self, ctx: &PluginContext) {
426 match ctx.ask_service::<PipelineStatusReq, Vec<String>>(
427 ek_oms::SVC_PIPELINE_STATUS,
428 PipelineStatusReq,
429 Duration::from_secs(2),
430 ) {
431 Ok(lines) => self.shared.set_pipeline_lines(lines),
432 Err(e) => {
433 if !self.shared.pipeline_lines().is_empty() {
437 self.shared
438 .log_line(format!("[ui] oms.pipeline_status 失手: {e:?}"));
439 }
440 self.shared.set_pipeline_lines(Vec::new());
441 }
442 }
443 }
444
445 fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {
455 let orders = ctx.ask_service::<GetAllOrdersReq, Vec<OrderData>>(
456 "oms.get_all_orders",
457 GetAllOrdersReq,
458 Duration::from_secs(2),
459 );
460 let trades = ctx.ask_service::<GetAllTradesReq, Vec<TradeData>>(
461 "oms.get_all_trades",
462 GetAllTradesReq,
463 Duration::from_secs(2),
464 );
465 let positions = ctx.ask_service::<GetAllPositionsReq, Vec<PositionData>>(
466 "oms.get_all_positions",
467 GetAllPositionsReq,
468 Duration::from_secs(2),
469 );
470 let funds = ctx.ask_service::<GetAllAccountsReq, Vec<AccountData>>(
471 "oms.get_all_accounts",
472 GetAllAccountsReq,
473 Duration::from_secs(2),
474 );
475 let mut n = 0usize;
476 if let Ok(rows) = &orders {
477 for o in rows {
478 shared.note_order(o);
479 }
480 n += rows.len();
481 }
482 if let Ok(rows) = &trades {
483 for t in rows {
484 shared.note_trade(t);
485 }
486 n += rows.len();
487 }
488 if let Ok(rows) = &positions {
489 for p in rows {
490 shared.note_position(p);
491 }
492 n += rows.len();
493 }
494 if let Ok(rows) = &funds {
495 for a in rows {
496 shared.note_funds(a);
497 }
498 n += rows.len();
499 }
500 if n > 0 {
502 let bad = usize::from(orders.is_err())
503 + usize::from(trades.is_err())
504 + usize::from(positions.is_err())
505 + usize::from(funds.is_err());
506 shared.log_line(format!(
507 "[ui] 启动播种 {n} 行(委托 {} / 成交 {} / 持仓 {} / 资金 {};失手 {} 口)",
508 orders.as_ref().map_or(0, |v| v.len()),
509 trades.as_ref().map_or(0, |v| v.len()),
510 positions.as_ref().map_or(0, |v| v.len()),
511 funds.as_ref().map_or(0, |v| v.len()),
512 bad
513 ));
514 }
515 }
516
517 fn spawn_cmd_thread(&self) {
525 let Some(ctx) = self.ctx.get().cloned() else {
526 return; };
528 let shared = std::sync::Arc::clone(&self.shared);
529 let stop = std::sync::Arc::clone(&self.cmd_stop);
530 if std::thread::Builder::new()
531 .name("ui-cmd".to_string())
532 .spawn(move || {
533 loop {
534 if stop.load(std::sync::atomic::Ordering::Acquire) {
535 break;
536 }
537 if shared.cmd_pending() {
539 Self::run_cmds(&shared, &ctx);
540 continue;
541 }
542 if shared.take_trade_reseed() {
544 Self::reconcile_trade(&shared, &ctx);
545 continue;
546 }
547 if shared.kline_fetch_pending() {
549 Self::run_kline_fetches(&shared, &ctx);
550 continue;
551 }
552 std::thread::sleep(Duration::from_millis(20));
553 }
554 })
555 .is_err()
556 {
557 self.shared
559 .log_line("[ui] 命令线程起不来:操作指令会滞留在队列里");
560 }
561 }
562
563 fn run_kline_fetches(shared: &UiShared, ctx: &PluginContext) {
567 for f in shared.take_kline_fetches() {
568 let Some(gw) = ctx
569 .plugins()
570 .iter()
571 .map(|p| p.name.clone())
572 .find(|n| n.starts_with("tq"))
573 else {
574 shared.log_line(format!(
575 "[kline] {} tq 网关不在线,历史补数丢弃(重开 K线窗再试)",
576 f.symbol
577 ));
578 continue;
579 };
580 let req = ek_tq::KlinesReq {
581 symbol: f.symbol.clone(),
582 raw: false,
583 interval_secs: f.interval_secs,
584 range: ek_tq::RangeSpec::Length { count: 300, end_ns: None },
585 max_rows: Some(300),
586 };
587 match ctx.ask::<ek_tq::KlinesReq, ek_tq::KlinesReply>(
588 &gw,
589 req,
590 Duration::from_secs(45),
591 ) {
592 Ok(reply) => {
593 if let Some(err) = &reply.error {
594 shared.log_line(format!(
595 "[kline] {}/{}s 历史显形: {}",
596 f.symbol, f.interval_secs, err.message
597 ));
598 continue;
599 }
600 let n = reply.rows.len();
601 shared.market().note_history(&f.symbol, f.interval_secs, &reply.rows);
602 shared.bump();
603 shared.log_line(format!(
604 "[kline] {}/{}s 历史补数 +{n} 根(mode={} truncated={})",
605 f.symbol,
606 f.interval_secs,
607 reply.mode.as_str(),
608 reply.truncated
609 ));
610 }
611 Err(e) => shared.log_line(format!(
612 "[kline] {}/{}s 历史 ask 失手: {e:?}",
613 f.symbol, f.interval_secs
614 )),
615 }
616 }
617 }
618
619 fn run_cmds(shared: &UiShared, ctx: &PluginContext) { for cmd in shared.drain_cmds() {
621 match cmd {
622 crate::state::UiCmd::Start { factory, name } => {
623 match ctx.ask_service::<StartReq, StartRes>(
624 SVC_START,
625 StartReq {
626 factory: factory.clone(),
627 name: name.clone(),
628 },
629 Duration::from_secs(3),
630 ) {
631 Ok(StartRes::Started { name }) => {
632 shared.log_line(format!("[ops] svctl.start {factory} → {name} 上线"));
633 }
634 Ok(StartRes::Rejected { reason }) => {
635 shared.log_line(format!("[ops] start {factory} 被拒: {reason}"));
636 }
637 Err(e) => {
638 shared.log_line(format!("[ops] start {factory} 失败: {e:?}"));
639 }
640 }
641 shared.request_list_refresh();
643 }
644 crate::state::UiCmd::Remove { name } => {
645 if name == UI_NAME {
653 shared.log_line(
654 "[ops] 拒绝移除 ui:观测面自身不可自摘(摘掉后没有下一拍执行任何指令)"
655 .to_string(),
656 );
657 continue;
658 }
659 match ctx.ask_service::<RemoveReq, RemoveRes>(
660 SVC_REMOVE,
661 RemoveReq { name: name.clone() },
662 Duration::from_secs(3),
663 ) {
664 Ok(RemoveRes::Removed { name }) => {
665 shared.log_line(format!("[ops] svctl.remove {name} 完成"));
666 }
667 Ok(RemoveRes::NoSuchName { name }) => {
668 shared.log_line(format!("[ops] remove {name} 不存在/不可摘"));
669 }
670 Ok(RemoveRes::Rejected { name, reason }) => {
674 shared.log_line(format!("[ops] remove {name} 被拒: {reason}"));
675 }
676 Err(e) => {
677 shared.log_line(format!("[ops] remove {name} 失败: {e:?}"));
678 }
679 }
680 shared.request_list_refresh();
681 }
682 crate::state::UiCmd::Reload {
683 interval,
684 symbol,
685 bars,
686 } => {
687 match ctx.ask_service::<ReloadReq, usize>(
689 "dm.reload",
690 ReloadReq {
691 interval,
692 symbol: symbol.clone(),
693 bars: bars.clone(),
694 size: Some(bars.len()),
695 push: true,
696 },
697 Duration::from_secs(5),
698 ) {
699 Ok(n) => {
700 shared.market().note_reload();
701 shared.log_line(format!(
702 "[ops] dm.reload {symbol}@{interval}s 回灌 {n} 条"
703 ));
704 }
705 Err(e) => {
706 shared.log_line(format!("[ops] dm.reload {symbol} 失败: {e:?}"));
707 }
708 }
709 }
710 crate::state::UiCmd::SendOrder { req, gateway, mode } => {
711 match ctx.ask_service::<SubmitOrderReq, SubmitReply>(
715 "oms.submit_order",
716 SubmitOrderReq {
717 req: req.clone(),
718 gateway_name: gateway.clone(),
719 mode,
720 },
721 Duration::from_secs(10),
722 ) {
723 Ok(reply) => {
724 let tag = match reply.status {
725 SubmitStatus::Sent => "SENT",
726 SubmitStatus::Partial => "PARTIAL",
727 SubmitStatus::Rejected => "REJECTED",
728 };
729 for leg in &reply.legs {
730 match (leg.ok, &leg.en_orderid, &leg.reason) {
731 (true, Some(id), _) => shared.log_line(format!(
732 "[ops] submit {} {} {}x{} → {id} (parent {})",
733 leg.req.symbol,
734 leg.req.direction.as_str(),
735 leg.req.offset.as_str(),
736 leg.req.volume,
737 reply.parent_id
738 )),
739 _ => shared.log_line(format!(
740 "[ops] submit {} {} {}x{} 被拒: {} (parent {})",
741 leg.req.symbol,
742 leg.req.direction.as_str(),
743 leg.req.offset.as_str(),
744 leg.req.volume,
745 leg.reason.as_deref().unwrap_or("?"),
746 reply.parent_id
747 )),
748 }
749 }
750 if reply.legs.is_empty() {
751 shared.log_line(format!(
752 "[ops] submit {} {}x{} {tag}:可平不足零腿 (parent {})",
753 req.symbol,
754 req.offset.as_str(),
755 req.volume,
756 reply.parent_id
757 ));
758 }
759 }
760 Err(e) => {
761 shared.log_line(format!("[ops] oms.submit_order 失败: {e:?}"));
762 }
763 }
764 }
765 crate::state::UiCmd::CancelOrder { orderid, gateway } => {
766 match ctx.ask_service::<OmsCancelOrderReq, OmsCancelReply>(
770 "oms.cancel_order",
771 OmsCancelOrderReq {
772 en_orderid: format!("{gateway}.{orderid}"),
773 gateway_name: gateway.clone(),
774 },
775 Duration::from_secs(10),
776 ) {
777 Ok(OmsCancelReply { ok: true, msg }) => {
778 shared.log_line(format!("[ops] {msg}"));
779 }
780 Ok(OmsCancelReply { ok: false, msg }) => {
781 shared
783 .log_line(format!("[ops] {}", UiShared::cancel_reject_note(&msg)));
784 }
785 Err(e) => {
786 shared.log_line(format!("[ops] oms.cancel_order 失败: {e:?}"));
787 }
788 }
789 }
790 crate::state::UiCmd::CtpConfigure { gateway, form } => {
791 let svc = format!("{gateway}.configure");
794 match ctx.ask_service::<ConfigureReq, ConfigureReply>(
795 svc.as_str(),
796 form.to_configure_req(),
797 Duration::from_secs(5),
798 ) {
799 Ok(ConfigureReply { ok: true, msg, .. }) => {
803 shared.log_line(format!("[ops] {svc} 受理:{msg}"));
804 }
805 Ok(ConfigureReply { ok: false, msg, .. }) => {
806 shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
807 }
808 Err(e) => {
809 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
810 }
811 }
812 }
813 crate::state::UiCmd::Connect { gateway, form } => {
814 shared.note_connect_sent(&gateway);
819 let form = shared.prepare_connect_form(&gateway, form);
822 let started = match ctx.ask_service::<StartReq, StartRes>(
823 SVC_START,
824 StartReq {
825 factory: "ctp".to_string(),
826 name: gateway.clone(),
827 },
828 Duration::from_secs(3),
829 ) {
830 Ok(StartRes::Started { name }) => {
831 shared.log_line(format!("[ops] svctl.start ctp → {name} 上线"));
832 true
833 }
834 Ok(StartRes::Rejected { reason }) => {
835 if shared.account(&gateway).is_some_and(|a| a.connected) {
838 shared
839 .log_line(format!("[ops] {gateway} 已在场,重投凭据: {reason}"));
840 true
841 } else {
842 shared.log_line(format!("[ops] {gateway} 装配被拒: {reason}"));
843 shared.note_connect_result(&gateway, Err(reason));
844 continue;
845 }
846 }
847 Err(e) => {
848 shared.log_line(format!("[ops] {gateway} 装配失败: {e:?}"));
849 shared.note_connect_result(&gateway, Err(format!("{e:?}")));
850 continue;
851 }
852 };
853 if !started {
854 continue;
855 }
856 let svc = format!("{gateway}.configure");
857 match ctx.ask_service::<ConfigureReq, ConfigureReply>(
858 svc.as_str(),
859 form.to_configure_req(),
860 Duration::from_secs(5),
861 ) {
862 Ok(ConfigureReply {
866 ok: true,
867 msg,
868 product_classes,
869 }) => {
870 shared.log_line(format!("[ops] {svc} 受理:{msg}"));
871 shared.note_effective_product_classes(&gateway, &product_classes);
872 shared.note_connect_result(&gateway, Ok(()));
873 }
874 Ok(ConfigureReply { ok: false, msg, .. }) => {
875 shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
876 shared.note_connect_result(&gateway, Err(msg));
877 }
878 Err(e) => {
879 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
880 shared.note_connect_result(&gateway, Err(format!("{e:?}")));
881 }
882 }
883 }
884 crate::state::UiCmd::CtpSubscribe { gateway, symbols } => {
885 let svc = format!("{gateway}.subscribe");
889 match ctx.ask_service::<CtpSubscribeReq, CtpSubscribeReply>(
890 svc.as_str(),
891 CtpSubscribeReq {
892 symbols: symbols.clone(),
893 },
894 Duration::from_secs(5),
895 ) {
896 Ok(CtpSubscribeReply { ok: true, msg }) => {
897 shared.note_subscriptions(&symbols);
898 shared.log_line(format!(
899 "[ops] {svc} {} 受理:{}",
900 symbols.join(","),
901 msg
902 ));
903 }
904 Ok(CtpSubscribeReply { ok: false, msg }) => {
905 shared.log_line(format!(
906 "[ops] {svc} {} 被拒:{}",
907 symbols.join(","),
908 msg
909 ));
910 }
911 Err(e) => {
912 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
913 }
914 }
915 }
916 crate::state::UiCmd::Disconnect { gateway } => {
917 let svc = format!("{gateway}.disconnect");
921 match ctx.ask_service::<DisconnectReq, DisconnectReply>(
922 svc.as_str(),
923 DisconnectReq,
924 Duration::from_secs(5),
925 ) {
926 Ok(DisconnectReply { ok: true, msg }) => {
927 shared.mark_account_stopped(&gateway);
928 shared.log_line(format!("[账户] 断开 {gateway}:{msg}(实例仍在场)"));
929 }
930 Ok(DisconnectReply { ok: false, msg }) => {
931 shared.log_line(format!("[账户] 断开 {gateway} 未执行:{msg}"));
932 }
933 Err(e) => {
934 shared.log_line(format!(
935 "[账户] 断开 {gateway} 失败:{e:?}(网关不在场或超时)",
936 ));
937 }
938 }
939 }
940 crate::state::UiCmd::CfgList => {
941 let (_, svc_list, _) = cfg_svc_ids();
943 match ctx.ask_service::<CfgListReq, CfgListRes>(
944 svc_list.as_str(),
945 CfgListReq,
946 Duration::from_secs(2),
947 ) {
948 Ok(res) => {
949 shared.set_cfg_briefs(res.entries);
950 if !res.msg.is_empty() {
951 shared.log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
952 }
953 }
954 Err(e) => shared.log_line(format!("[cfg] 段落目录失败: {e:?}")),
955 }
956 }
957 crate::state::UiCmd::CfgDelete { ns } => {
958 let (svc_save, svc_list, svc_delete) = cfg_svc_ids();
960 let _ = (svc_save, svc_list);
961 match ctx.ask_service::<CfgDeleteReq, CfgDeleteRes>(
962 svc_delete.as_str(),
963 CfgDeleteReq {
964 ns: ns.clone(),
965 key: String::new(),
966 },
967 Duration::from_secs(2),
968 ) {
969 Ok(res) if res.ok => {
970 shared.log_line(format!(
971 "[cfg] 清段 {ns}:键 {} / 秘条 {}",
972 res.removed_keys, res.removed_secrets
973 ));
974 shared.request_cfg_refresh();
975 }
976 Ok(res) => shared.log_line(format!("[cfg] 清段 {ns} 未执行:{}", res.msg)),
977 Err(e) => shared.log_line(format!("[cfg] 清段 {ns} 失败: {e:?}")),
978 }
979 shared.arm_cfg_delete(None);
980 }
981 crate::state::UiCmd::CfgResave { ns } => {
982 let store = ui_cfg_store();
987 let loaded = store
988 .load(&ns)
989 .map_err(|e| e.to_string())
990 .and_then(|o| match o {
991 ek_cfg::LoadOutcome::Fresh => Err(format!("{ns} 段不存在,不持久空形")),
992 map @ ek_cfg::LoadOutcome::Loaded { .. } => Ok(map),
993 });
994 match loaded {
995 Ok(ek_cfg::LoadOutcome::Fresh) => {
996 unreachable!("Fresh 已被上面的 and_then 挡走");
997 }
998 Ok(ek_cfg::LoadOutcome::Loaded {
999 map,
1000 warnings,
1001 backup_used,
1002 }) => {
1003 if !warnings.is_empty() {
1004 shared.log_line(format!("[cfg] {ns} 重读警告:{warnings:?}"));
1005 }
1006 if backup_used {
1007 shared.log_line(format!("[cfg] {ns} 走了 .bak 回落"));
1008 }
1009 match store.store(&ns, &map) {
1010 Ok(()) => {
1011 shared.log_line(format!("[cfg] {ns} 已重存(rev {})", map.rev))
1012 }
1013 Err(e) => shared.log_line(format!("[cfg] {ns} 重存失败:{e}")),
1014 }
1015 }
1016 Err(msg) => shared.log_line(format!("[cfg] {msg}")),
1017 }
1018 }
1019 }
1020 }
1021 }
1022}
1023
1024impl Plugin for UiPlugin {
1025 fn name(&self) -> &str {
1026 UI_NAME
1027 }
1028
1029 fn deps(&self) -> &[&str] {
1030 &[] }
1032
1033 fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
1034 vec![
1035 std::borrow::Cow::Borrowed(EVENT_LOG),
1036 std::borrow::Cow::Borrowed(EVENT_PLUG_UP),
1037 std::borrow::Cow::Borrowed(EVENT_PLUG_DOWN),
1038 std::borrow::Cow::Borrowed(EVENT_BAR),
1039 std::borrow::Cow::Borrowed(EVENT_BAR_UPDATE),
1040 std::borrow::Cow::Borrowed(EVENT_TICK),
1041 std::borrow::Cow::Borrowed(EVENT_CONTRACT),
1042 std::borrow::Cow::Borrowed(ek_ctp::EVENT_CONTRACT_FINISH),
1043 std::borrow::Cow::Borrowed(W_ORDER),
1047 std::borrow::Cow::Borrowed(W_TRADE),
1048 std::borrow::Cow::Borrowed(W_POSITION),
1049 std::borrow::Cow::Borrowed(W_ACCOUNT),
1050 std::borrow::Cow::Borrowed(EVENT_TIMER),
1051 std::borrow::Cow::Borrowed(ek_ctp::EVENT_STATUS_PREFIX),
1053 ]
1054 }
1055
1056 fn mailbox_soft_cap(&self) -> u64 {
1057 UI_MAILBOX_SOFT_CAP
1058 }
1059
1060 fn version(&self) -> std::borrow::Cow<'static, str> {
1061 std::borrow::Cow::Borrowed(env!("CARGO_PKG_VERSION"))
1062 }
1063
1064 fn services(&self) -> Vec<ServiceDesc> {
1065 Vec::new() }
1067
1068 fn on_init(&self, ctx: &PluginContext) {
1069 let _ = self.ctx.set(ctx.clone());
1070 }
1071
1072 fn on_start(&self) {
1073 self.shared.log_line("[ui] 界面插件上线");
1074 self.spawn_cmd_thread();
1078 self.shared.request_trade_reseed();
1081 self.shared.request_pipeline_refresh();
1082 let theme = load_theme();
1084 self.shared.set_theme(theme);
1085 if theme == crate::tokens::Theme::Light {
1086 self.shared.log_line("[ui] 主题:浅色(读盘 ui.theme)");
1087 }
1088 }
1089
1090 fn on_stop(&self) {
1091 self.shared.log_line("[ui] 界面插件下线");
1092 self.cmd_stop
1094 .store(true, std::sync::atomic::Ordering::Release);
1095 self.flush_persist();
1099 }
1100
1101 fn on_mail(&self, event: &ek_event::Event) {
1102 let ty = event.event_type.as_str();
1103 if ty == EVENT_LOG {
1104 if let Some(d) = event.data.downcast_ref::<LogData>() {
1105 let tag = if d.is_warning_or_above() {
1106 "WARN"
1107 } else {
1108 "INFO"
1109 };
1110 self.shared
1111 .log_line(format!("[{tag}][{}] {}", d.gateway_name, d.msg));
1112 if let Some(note) = d.msg.strip_prefix(crate::state::CHAIN_STATE_PREFIX) {
1118 self.shared.note_chain(&d.gateway_name, note);
1119 }
1120 } else {
1121 self.shared.log_line("[eLog] 载荷类型异常(非 LogData)");
1122 }
1123 return;
1124 }
1125 if ty == ek_ctp::EVENT_STATUS_PREFIX {
1126 if let Some(ev) = event.data.downcast_ref::<ek_ctp::GatewayStatusEvent>() {
1128 self.shared.note_status_event(ev);
1129 } else {
1130 self.shared
1131 .log_line("[eStatus] 载荷类型异常(非 GatewayStatusEvent)");
1132 }
1133 return;
1134 }
1135 if ty == EVENT_PLUG_UP || ty == EVENT_PLUG_DOWN {
1136 if let Some(n) = event.data.downcast_ref::<PlugNotice>() {
1137 let verb = if ty == EVENT_PLUG_UP {
1138 "上线"
1139 } else {
1140 "下线"
1141 };
1142 self.shared
1143 .log_line(format!("[host] 插件{verb} {} v{}", n.name, n.version));
1144 self.shared.request_list_refresh();
1146 if ty == EVENT_PLUG_UP {
1147 if n.name == ek_oms::NAME {
1149 self.shared.request_pipeline_refresh();
1150 }
1151 let is_account = self.shared.accounts().iter().any(|a| a.gateway == n.name);
1156 if n.name == ek_oms::NAME || is_account {
1157 self.shared.request_trade_reseed();
1158 }
1159 if is_account {
1162 self.shared.request_status_seed(&n.name);
1163 }
1164 let md = self.shared.md_source();
1167 let subs = self.shared.subscriptions();
1168 if !md.is_empty() && n.name == md && !subs.is_empty() {
1169 self.shared.push_cmd(crate::state::UiCmd::CtpSubscribe {
1170 gateway: n.name.clone(),
1171 symbols: subs.clone(),
1172 });
1173 self.shared.log_line(format!(
1174 "[ops] 行情源 {} 上线,重放订阅 {} 项",
1175 n.name,
1176 subs.len()
1177 ));
1178 }
1179 } else {
1180 let removed = self.shared.clear_gateway_rows(&n.name);
1184 if removed > 0 {
1185 self.shared.log_line(format!(
1186 "[ui] 插件下线 {}:作废 {} 行交易数据",
1187 n.name, removed
1188 ));
1189 }
1190 if n.name == ek_oms::NAME {
1195 self.shared.request_pipeline_refresh();
1196 }
1197 }
1198 }
1199 self.shared.bump();
1200 return;
1201 }
1202 if ty == EVENT_BAR {
1203 if let Some(f) = event.data.downcast_ref::<ek_dm::wire::BarsFrame>() {
1204 self.shared.market().on_frame(f);
1205 }
1206 self.shared.bump();
1207 return;
1208 }
1209 if ty == EVENT_BAR_UPDATE {
1210 if let Some(b) = event.data.downcast_ref::<ek_data::bar::BarData>() {
1211 self.shared.market().on_update(b);
1212 }
1213 self.shared.bump();
1214 return;
1215 }
1216 if ty == EVENT_TICK {
1217 if let Some(t) = event.data.downcast_ref::<ek_data::tick::TickData>() {
1218 self.shared.update_tick(t);
1219 }
1220 return;
1221 }
1222 if ty == EVENT_CONTRACT {
1223 if let Some(v) = event
1226 .data
1227 .downcast_ref::<Vec<ek_data::contract::ContractData>>()
1228 {
1229 for c in v {
1230 self.shared.update_contract(c);
1231 }
1232 } else if let Some(c) = event.data.downcast_ref::<ek_data::contract::ContractData>() {
1233 self.shared.update_contract(c);
1234 }
1235 return;
1236 }
1237 if ty == ek_ctp::EVENT_CONTRACT_FINISH {
1238 let gw = event
1239 .data
1240 .downcast_ref::<String>()
1241 .cloned()
1242 .unwrap_or_else(|| "?".to_string());
1243 self.shared
1244 .log_line(format!("[ctp] 合约流完成 {gw}(查询合约窗可勾选订阅)"));
1245 return;
1246 }
1247 if ty == W_ORDER {
1250 match event.data.downcast_ref::<OrderData>() {
1251 Some(o) => self.shared.note_order(o),
1252 None => self.shared.note_bad_payload(W_ORDER),
1253 }
1254 return;
1255 }
1256 if ty == W_TRADE {
1257 match event.data.downcast_ref::<TradeData>() {
1258 Some(t) => self.shared.note_trade(t),
1259 None => self.shared.note_bad_payload(W_TRADE),
1260 }
1261 return;
1262 }
1263 if ty == W_POSITION {
1264 match event.data.downcast_ref::<PositionData>() {
1265 Some(p) => self.shared.note_position(p),
1266 None => self.shared.note_bad_payload(W_POSITION),
1267 }
1268 return;
1269 }
1270 if ty == W_ACCOUNT {
1271 match event.data.downcast_ref::<AccountData>() {
1272 Some(a) => self.shared.note_funds(a),
1273 None => self.shared.note_bad_payload(W_ACCOUNT),
1274 }
1275 return;
1276 }
1277 if ty == EVENT_TIMER {
1278 self.refresh();
1279 }
1280 }
1281}