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
159use crate::state::{BusSnap, UiShared};
160
161pub const UI_NAME: &str = "ui";
163
164pub const UI_MAILBOX_SOFT_CAP: u64 = 100_000;
173
174const W_ORDER: &str = ek_oms::wire::EVENT_ORDER;
177const W_TRADE: &str = ek_oms::wire::EVENT_TRADE;
178const W_POSITION: &str = ek_oms::wire::EVENT_POSITION;
179const W_ACCOUNT: &str = ek_oms::wire::EVENT_ACCOUNT;
180
181pub struct UiPlugin {
183 shared: std::sync::Arc<UiShared>,
184 ctx: OnceLock<PluginContext>,
185 cmd_stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
187}
188
189impl UiPlugin {
190 pub fn new(shared: std::sync::Arc<UiShared>) -> Self {
192 Self {
193 shared,
194 ctx: OnceLock::new(),
195 cmd_stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
196 }
197 }
198
199 fn refresh(&self) {
202 let Some(ctx) = self.ctx.get() else {
203 return; };
205 if self.shared.list_poll_due() {
210 self.poll_list(ctx);
211 }
212 self.seed_status(ctx);
213 if self.shared.take_pipeline_due() {
214 self.poll_pipeline(ctx);
215 }
216 if self.shared.take_cfg_due() {
217 self.poll_cfg(ctx);
220 }
221 if self.shared.take_theme_persist_due() {
222 let theme = self.shared.theme();
224 match save_theme(theme) {
225 Ok(()) => self
226 .shared
227 .log_line(format!("[ui] 主题已存(ui.theme={})", theme.as_str())),
228 Err(e) => self
229 .shared
230 .log_line(format!("[ui] 主题落盘失败:{e}(本次切换只在本次运行内生效)")),
231 }
232 }
233 if self.shared.take_auto_connect_persist_due() {
234 let flags: Vec<(String, bool)> = self
237 .shared
238 .accounts()
239 .iter()
240 .map(|a| (a.gateway.clone(), a.auto_connect))
241 .collect();
242 match save_auto_connect_flags_into(&ui_cfg_store(), &flags) {
243 Ok(()) => {
244 let on: Vec<&str> = flags
245 .iter()
246 .filter(|(_, on)| *on)
247 .map(|(gw, _)| gw.as_str())
248 .collect();
249 if on.is_empty() {
250 self.shared
251 .log_line("[ui] 自动连接勾选已存(ui 段;当前全部关)".to_string());
252 } else {
253 self.shared.log_line(format!(
254 "[ui] 自动连接勾选已存(ui 段;开:{};守护/桌面启动清点按它起连)",
255 on.join(", ")
256 ));
257 }
258 }
259 Err(e) => self.shared.log_line(format!(
260 "[ui] 自动连接勾选落盘失败:{e}(本轮勾选只在本次运行内生效)"
261 )),
262 }
263 }
264 }
265
266 fn poll_cfg(&self, ctx: &PluginContext) {
268 let (_, svc_list, _) = cfg_svc_ids();
269 match ctx.ask_service::<CfgListReq, CfgListRes>(
270 svc_list.as_str(),
271 CfgListReq,
272 Duration::from_secs(2),
273 ) {
274 Ok(res) => {
275 self.shared.set_cfg_briefs(res.entries);
276 if !res.msg.is_empty() {
277 self.shared
278 .log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
279 }
280 }
281 Err(e) => self
282 .shared
283 .log_line(format!("[cfg] 段落目录失败(cfg 缺席或超时): {e:?}")),
284 }
285 }
286
287 fn poll_list(&self, ctx: &PluginContext) {
293 match ctx.ask_service::<ListReq, Vec<PlugRow>>(SVC_LIST, ListReq, Duration::from_secs(2)) {
294 Ok(rows) => {
295 let mut services = Vec::with_capacity(rows.len());
296 for r in &rows {
297 let ids = ctx
298 .services_of(&r.name)
299 .map(|v| v.iter().map(|s| s.id.clone()).collect::<Vec<_>>())
300 .unwrap_or_default();
301 services.push((r.name.clone(), ids));
302 }
303 self.shared.sync_accounts_connected(&rows);
306 self.shared.note_plugin_spill(&rows);
309 self.shared.set_snapshot(BusSnap { rows, services });
310 }
311 Err(e) => self.shared.log_line(format!("[ui] svctl.list 失败: {e}")),
312 }
313 }
314
315 fn seed_status(&self, ctx: &PluginContext) {
322 for gw in self.shared.take_status_seeds() {
323 let svc = format!("{gw}.{}", ek_ctp::SVC_STATUS);
324 match ctx.ask_service::<StatusReq, GatewayStatus>(
325 &svc,
326 StatusReq,
327 Duration::from_secs(2),
328 ) {
329 Ok(st) => self.shared.note_gateway_status(&gw, &st),
330 Err(e) => {
331 self.shared.note_status_unreachable(&gw, &format!("{e}"));
335 }
336 }
337 }
338 }
339
340 fn poll_pipeline(&self, ctx: &PluginContext) {
345 match ctx.ask_service::<PipelineStatusReq, Vec<String>>(
346 ek_oms::SVC_PIPELINE_STATUS,
347 PipelineStatusReq,
348 Duration::from_secs(2),
349 ) {
350 Ok(lines) => self.shared.set_pipeline_lines(lines),
351 Err(e) => {
352 if !self.shared.pipeline_lines().is_empty() {
356 self.shared
357 .log_line(format!("[ui] oms.pipeline_status 失手: {e:?}"));
358 }
359 self.shared.set_pipeline_lines(Vec::new());
360 }
361 }
362 }
363
364 fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {
374 let orders = ctx.ask_service::<GetAllOrdersReq, Vec<OrderData>>(
375 "oms.get_all_orders",
376 GetAllOrdersReq,
377 Duration::from_secs(2),
378 );
379 let trades = ctx.ask_service::<GetAllTradesReq, Vec<TradeData>>(
380 "oms.get_all_trades",
381 GetAllTradesReq,
382 Duration::from_secs(2),
383 );
384 let positions = ctx.ask_service::<GetAllPositionsReq, Vec<PositionData>>(
385 "oms.get_all_positions",
386 GetAllPositionsReq,
387 Duration::from_secs(2),
388 );
389 let funds = ctx.ask_service::<GetAllAccountsReq, Vec<AccountData>>(
390 "oms.get_all_accounts",
391 GetAllAccountsReq,
392 Duration::from_secs(2),
393 );
394 let mut n = 0usize;
395 if let Ok(rows) = &orders {
396 for o in rows {
397 shared.note_order(o);
398 }
399 n += rows.len();
400 }
401 if let Ok(rows) = &trades {
402 for t in rows {
403 shared.note_trade(t);
404 }
405 n += rows.len();
406 }
407 if let Ok(rows) = &positions {
408 for p in rows {
409 shared.note_position(p);
410 }
411 n += rows.len();
412 }
413 if let Ok(rows) = &funds {
414 for a in rows {
415 shared.note_funds(a);
416 }
417 n += rows.len();
418 }
419 if n > 0 {
421 let bad = usize::from(orders.is_err())
422 + usize::from(trades.is_err())
423 + usize::from(positions.is_err())
424 + usize::from(funds.is_err());
425 shared.log_line(format!(
426 "[ui] 启动播种 {n} 行(委托 {} / 成交 {} / 持仓 {} / 资金 {};失手 {} 口)",
427 orders.as_ref().map_or(0, |v| v.len()),
428 trades.as_ref().map_or(0, |v| v.len()),
429 positions.as_ref().map_or(0, |v| v.len()),
430 funds.as_ref().map_or(0, |v| v.len()),
431 bad
432 ));
433 }
434 }
435
436 fn spawn_cmd_thread(&self) {
444 let Some(ctx) = self.ctx.get().cloned() else {
445 return; };
447 let shared = std::sync::Arc::clone(&self.shared);
448 let stop = std::sync::Arc::clone(&self.cmd_stop);
449 if std::thread::Builder::new()
450 .name("ui-cmd".to_string())
451 .spawn(move || {
452 loop {
453 if stop.load(std::sync::atomic::Ordering::Acquire) {
454 break;
455 }
456 if shared.cmd_pending() {
458 Self::run_cmds(&shared, &ctx);
459 continue;
460 }
461 if shared.take_trade_reseed() {
463 Self::reconcile_trade(&shared, &ctx);
464 continue;
465 }
466 if shared.kline_fetch_pending() {
468 Self::run_kline_fetches(&shared, &ctx);
469 continue;
470 }
471 std::thread::sleep(Duration::from_millis(20));
472 }
473 })
474 .is_err()
475 {
476 self.shared
478 .log_line("[ui] 命令线程起不来:操作指令会滞留在队列里");
479 }
480 }
481
482 fn run_kline_fetches(shared: &UiShared, ctx: &PluginContext) {
486 for f in shared.take_kline_fetches() {
487 let Some(gw) = ctx
488 .plugins()
489 .iter()
490 .map(|p| p.name.clone())
491 .find(|n| n.starts_with("tq"))
492 else {
493 shared.log_line(format!(
494 "[kline] {} tq 网关不在线,历史补数丢弃(重开 K线窗再试)",
495 f.symbol
496 ));
497 continue;
498 };
499 let req = ek_tq::KlinesReq {
500 symbol: f.symbol.clone(),
501 raw: false,
502 interval_secs: f.interval_secs,
503 range: ek_tq::RangeSpec::Length { count: 300, end_ns: None },
504 max_rows: Some(300),
505 };
506 match ctx.ask::<ek_tq::KlinesReq, ek_tq::KlinesReply>(
507 &gw,
508 req,
509 Duration::from_secs(45),
510 ) {
511 Ok(reply) => {
512 if let Some(err) = &reply.error {
513 shared.log_line(format!(
514 "[kline] {}/{}s 历史显形: {}",
515 f.symbol, f.interval_secs, err.message
516 ));
517 continue;
518 }
519 let n = reply.rows.len();
520 shared.market().note_history(&f.symbol, f.interval_secs, &reply.rows);
521 shared.bump();
522 shared.log_line(format!(
523 "[kline] {}/{}s 历史补数 +{n} 根(mode={} truncated={})",
524 f.symbol,
525 f.interval_secs,
526 reply.mode.as_str(),
527 reply.truncated
528 ));
529 }
530 Err(e) => shared.log_line(format!(
531 "[kline] {}/{}s 历史 ask 失手: {e:?}",
532 f.symbol, f.interval_secs
533 )),
534 }
535 }
536 }
537
538 fn run_cmds(shared: &UiShared, ctx: &PluginContext) { for cmd in shared.drain_cmds() {
540 match cmd {
541 crate::state::UiCmd::Start { factory, name } => {
542 match ctx.ask_service::<StartReq, StartRes>(
543 SVC_START,
544 StartReq {
545 factory: factory.clone(),
546 name: name.clone(),
547 },
548 Duration::from_secs(3),
549 ) {
550 Ok(StartRes::Started { name }) => {
551 shared.log_line(format!("[ops] svctl.start {factory} → {name} 上线"));
552 }
553 Ok(StartRes::Rejected { reason }) => {
554 shared.log_line(format!("[ops] start {factory} 被拒: {reason}"));
555 }
556 Err(e) => {
557 shared.log_line(format!("[ops] start {factory} 失败: {e:?}"));
558 }
559 }
560 shared.request_list_refresh();
562 }
563 crate::state::UiCmd::Remove { name } => {
564 if name == UI_NAME {
572 shared.log_line(
573 "[ops] 拒绝移除 ui:观测面自身不可自摘(摘掉后没有下一拍执行任何指令)"
574 .to_string(),
575 );
576 continue;
577 }
578 match ctx.ask_service::<RemoveReq, RemoveRes>(
579 SVC_REMOVE,
580 RemoveReq { name: name.clone() },
581 Duration::from_secs(3),
582 ) {
583 Ok(RemoveRes::Removed { name }) => {
584 shared.log_line(format!("[ops] svctl.remove {name} 完成"));
585 }
586 Ok(RemoveRes::NoSuchName { name }) => {
587 shared.log_line(format!("[ops] remove {name} 不存在/不可摘"));
588 }
589 Ok(RemoveRes::Rejected { name, reason }) => {
593 shared.log_line(format!("[ops] remove {name} 被拒: {reason}"));
594 }
595 Err(e) => {
596 shared.log_line(format!("[ops] remove {name} 失败: {e:?}"));
597 }
598 }
599 shared.request_list_refresh();
600 }
601 crate::state::UiCmd::Reload {
602 interval,
603 symbol,
604 bars,
605 } => {
606 match ctx.ask_service::<ReloadReq, usize>(
608 "dm.reload",
609 ReloadReq {
610 interval,
611 symbol: symbol.clone(),
612 bars: bars.clone(),
613 size: Some(bars.len()),
614 push: true,
615 },
616 Duration::from_secs(5),
617 ) {
618 Ok(n) => {
619 shared.market().note_reload();
620 shared.log_line(format!(
621 "[ops] dm.reload {symbol}@{interval}s 回灌 {n} 条"
622 ));
623 }
624 Err(e) => {
625 shared.log_line(format!("[ops] dm.reload {symbol} 失败: {e:?}"));
626 }
627 }
628 }
629 crate::state::UiCmd::SendOrder { req, gateway, mode } => {
630 match ctx.ask_service::<SubmitOrderReq, SubmitReply>(
634 "oms.submit_order",
635 SubmitOrderReq {
636 req: req.clone(),
637 gateway_name: gateway.clone(),
638 mode,
639 },
640 Duration::from_secs(10),
641 ) {
642 Ok(reply) => {
643 let tag = match reply.status {
644 SubmitStatus::Sent => "SENT",
645 SubmitStatus::Partial => "PARTIAL",
646 SubmitStatus::Rejected => "REJECTED",
647 };
648 for leg in &reply.legs {
649 match (leg.ok, &leg.en_orderid, &leg.reason) {
650 (true, Some(id), _) => shared.log_line(format!(
651 "[ops] submit {} {} {}x{} → {id} (parent {})",
652 leg.req.symbol,
653 leg.req.direction.as_str(),
654 leg.req.offset.as_str(),
655 leg.req.volume,
656 reply.parent_id
657 )),
658 _ => shared.log_line(format!(
659 "[ops] submit {} {} {}x{} 被拒: {} (parent {})",
660 leg.req.symbol,
661 leg.req.direction.as_str(),
662 leg.req.offset.as_str(),
663 leg.req.volume,
664 leg.reason.as_deref().unwrap_or("?"),
665 reply.parent_id
666 )),
667 }
668 }
669 if reply.legs.is_empty() {
670 shared.log_line(format!(
671 "[ops] submit {} {}x{} {tag}:可平不足零腿 (parent {})",
672 req.symbol,
673 req.offset.as_str(),
674 req.volume,
675 reply.parent_id
676 ));
677 }
678 }
679 Err(e) => {
680 shared.log_line(format!("[ops] oms.submit_order 失败: {e:?}"));
681 }
682 }
683 }
684 crate::state::UiCmd::CancelOrder { orderid, gateway } => {
685 match ctx.ask_service::<OmsCancelOrderReq, OmsCancelReply>(
689 "oms.cancel_order",
690 OmsCancelOrderReq {
691 en_orderid: format!("{gateway}.{orderid}"),
692 gateway_name: gateway.clone(),
693 },
694 Duration::from_secs(10),
695 ) {
696 Ok(OmsCancelReply { ok: true, msg }) => {
697 shared.log_line(format!("[ops] {msg}"));
698 }
699 Ok(OmsCancelReply { ok: false, msg }) => {
700 shared
702 .log_line(format!("[ops] {}", UiShared::cancel_reject_note(&msg)));
703 }
704 Err(e) => {
705 shared.log_line(format!("[ops] oms.cancel_order 失败: {e:?}"));
706 }
707 }
708 }
709 crate::state::UiCmd::CtpConfigure { gateway, form } => {
710 let svc = format!("{gateway}.configure");
713 match ctx.ask_service::<ConfigureReq, ConfigureReply>(
714 svc.as_str(),
715 form.to_configure_req(),
716 Duration::from_secs(5),
717 ) {
718 Ok(ConfigureReply { ok: true, msg, .. }) => {
722 shared.log_line(format!("[ops] {svc} 受理:{msg}"));
723 }
724 Ok(ConfigureReply { ok: false, msg, .. }) => {
725 shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
726 }
727 Err(e) => {
728 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
729 }
730 }
731 }
732 crate::state::UiCmd::Connect { gateway, form } => {
733 shared.note_connect_sent(&gateway);
738 let form = shared.prepare_connect_form(&gateway, form);
741 let started = match ctx.ask_service::<StartReq, StartRes>(
742 SVC_START,
743 StartReq {
744 factory: "ctp".to_string(),
745 name: gateway.clone(),
746 },
747 Duration::from_secs(3),
748 ) {
749 Ok(StartRes::Started { name }) => {
750 shared.log_line(format!("[ops] svctl.start ctp → {name} 上线"));
751 true
752 }
753 Ok(StartRes::Rejected { reason }) => {
754 if shared.account(&gateway).is_some_and(|a| a.connected) {
757 shared
758 .log_line(format!("[ops] {gateway} 已在场,重投凭据: {reason}"));
759 true
760 } else {
761 shared.log_line(format!("[ops] {gateway} 装配被拒: {reason}"));
762 shared.note_connect_result(&gateway, Err(reason));
763 continue;
764 }
765 }
766 Err(e) => {
767 shared.log_line(format!("[ops] {gateway} 装配失败: {e:?}"));
768 shared.note_connect_result(&gateway, Err(format!("{e:?}")));
769 continue;
770 }
771 };
772 if !started {
773 continue;
774 }
775 let svc = format!("{gateway}.configure");
776 match ctx.ask_service::<ConfigureReq, ConfigureReply>(
777 svc.as_str(),
778 form.to_configure_req(),
779 Duration::from_secs(5),
780 ) {
781 Ok(ConfigureReply {
785 ok: true,
786 msg,
787 product_classes,
788 }) => {
789 shared.log_line(format!("[ops] {svc} 受理:{msg}"));
790 shared.note_effective_product_classes(&gateway, &product_classes);
791 shared.note_connect_result(&gateway, Ok(()));
792 }
793 Ok(ConfigureReply { ok: false, msg, .. }) => {
794 shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
795 shared.note_connect_result(&gateway, Err(msg));
796 }
797 Err(e) => {
798 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
799 shared.note_connect_result(&gateway, Err(format!("{e:?}")));
800 }
801 }
802 }
803 crate::state::UiCmd::CtpSubscribe { gateway, symbols } => {
804 let svc = format!("{gateway}.subscribe");
808 match ctx.ask_service::<CtpSubscribeReq, CtpSubscribeReply>(
809 svc.as_str(),
810 CtpSubscribeReq {
811 symbols: symbols.clone(),
812 },
813 Duration::from_secs(5),
814 ) {
815 Ok(CtpSubscribeReply { ok: true, msg }) => {
816 shared.note_subscriptions(&symbols);
817 shared.log_line(format!(
818 "[ops] {svc} {} 受理:{}",
819 symbols.join(","),
820 msg
821 ));
822 }
823 Ok(CtpSubscribeReply { ok: false, msg }) => {
824 shared.log_line(format!(
825 "[ops] {svc} {} 被拒:{}",
826 symbols.join(","),
827 msg
828 ));
829 }
830 Err(e) => {
831 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
832 }
833 }
834 }
835 crate::state::UiCmd::Disconnect { gateway } => {
836 let svc = format!("{gateway}.disconnect");
840 match ctx.ask_service::<DisconnectReq, DisconnectReply>(
841 svc.as_str(),
842 DisconnectReq,
843 Duration::from_secs(5),
844 ) {
845 Ok(DisconnectReply { ok: true, msg }) => {
846 shared.mark_account_stopped(&gateway);
847 shared.log_line(format!("[账户] 断开 {gateway}:{msg}(实例仍在场)"));
848 }
849 Ok(DisconnectReply { ok: false, msg }) => {
850 shared.log_line(format!("[账户] 断开 {gateway} 未执行:{msg}"));
851 }
852 Err(e) => {
853 shared.log_line(format!(
854 "[账户] 断开 {gateway} 失败:{e:?}(网关不在场或超时)",
855 ));
856 }
857 }
858 }
859 crate::state::UiCmd::CfgList => {
860 let (_, svc_list, _) = cfg_svc_ids();
862 match ctx.ask_service::<CfgListReq, CfgListRes>(
863 svc_list.as_str(),
864 CfgListReq,
865 Duration::from_secs(2),
866 ) {
867 Ok(res) => {
868 shared.set_cfg_briefs(res.entries);
869 if !res.msg.is_empty() {
870 shared.log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
871 }
872 }
873 Err(e) => shared.log_line(format!("[cfg] 段落目录失败: {e:?}")),
874 }
875 }
876 crate::state::UiCmd::CfgDelete { ns } => {
877 let (svc_save, svc_list, svc_delete) = cfg_svc_ids();
879 let _ = (svc_save, svc_list);
880 match ctx.ask_service::<CfgDeleteReq, CfgDeleteRes>(
881 svc_delete.as_str(),
882 CfgDeleteReq {
883 ns: ns.clone(),
884 key: String::new(),
885 },
886 Duration::from_secs(2),
887 ) {
888 Ok(res) if res.ok => {
889 shared.log_line(format!(
890 "[cfg] 清段 {ns}:键 {} / 秘条 {}",
891 res.removed_keys, res.removed_secrets
892 ));
893 shared.request_cfg_refresh();
894 }
895 Ok(res) => shared.log_line(format!("[cfg] 清段 {ns} 未执行:{}", res.msg)),
896 Err(e) => shared.log_line(format!("[cfg] 清段 {ns} 失败: {e:?}")),
897 }
898 shared.arm_cfg_delete(None);
899 }
900 crate::state::UiCmd::CfgResave { ns } => {
901 let store = ui_cfg_store();
906 let loaded = store
907 .load(&ns)
908 .map_err(|e| e.to_string())
909 .and_then(|o| match o {
910 ek_cfg::LoadOutcome::Fresh => Err(format!("{ns} 段不存在,不持久空形")),
911 map @ ek_cfg::LoadOutcome::Loaded { .. } => Ok(map),
912 });
913 match loaded {
914 Ok(ek_cfg::LoadOutcome::Fresh) => {
915 unreachable!("Fresh 已被上面的 and_then 挡走");
916 }
917 Ok(ek_cfg::LoadOutcome::Loaded {
918 map,
919 warnings,
920 backup_used,
921 }) => {
922 if !warnings.is_empty() {
923 shared.log_line(format!("[cfg] {ns} 重读警告:{warnings:?}"));
924 }
925 if backup_used {
926 shared.log_line(format!("[cfg] {ns} 走了 .bak 回落"));
927 }
928 match store.store(&ns, &map) {
929 Ok(()) => {
930 shared.log_line(format!("[cfg] {ns} 已重存(rev {})", map.rev))
931 }
932 Err(e) => shared.log_line(format!("[cfg] {ns} 重存失败:{e}")),
933 }
934 }
935 Err(msg) => shared.log_line(format!("[cfg] {msg}")),
936 }
937 }
938 }
939 }
940 }
941}
942
943impl Plugin for UiPlugin {
944 fn name(&self) -> &str {
945 UI_NAME
946 }
947
948 fn deps(&self) -> &[&str] {
949 &[] }
951
952 fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
953 vec![
954 std::borrow::Cow::Borrowed(EVENT_LOG),
955 std::borrow::Cow::Borrowed(EVENT_PLUG_UP),
956 std::borrow::Cow::Borrowed(EVENT_PLUG_DOWN),
957 std::borrow::Cow::Borrowed(EVENT_BAR),
958 std::borrow::Cow::Borrowed(EVENT_BAR_UPDATE),
959 std::borrow::Cow::Borrowed(EVENT_TICK),
960 std::borrow::Cow::Borrowed(EVENT_CONTRACT),
961 std::borrow::Cow::Borrowed(ek_ctp::EVENT_CONTRACT_FINISH),
962 std::borrow::Cow::Borrowed(W_ORDER),
966 std::borrow::Cow::Borrowed(W_TRADE),
967 std::borrow::Cow::Borrowed(W_POSITION),
968 std::borrow::Cow::Borrowed(W_ACCOUNT),
969 std::borrow::Cow::Borrowed(EVENT_TIMER),
970 std::borrow::Cow::Borrowed(ek_ctp::EVENT_STATUS_PREFIX),
972 ]
973 }
974
975 fn mailbox_soft_cap(&self) -> u64 {
976 UI_MAILBOX_SOFT_CAP
977 }
978
979 fn version(&self) -> std::borrow::Cow<'static, str> {
980 std::borrow::Cow::Borrowed(env!("CARGO_PKG_VERSION"))
981 }
982
983 fn services(&self) -> Vec<ServiceDesc> {
984 Vec::new() }
986
987 fn on_init(&self, ctx: &PluginContext) {
988 let _ = self.ctx.set(ctx.clone());
989 }
990
991 fn on_start(&self) {
992 self.shared.log_line("[ui] 界面插件上线");
993 self.spawn_cmd_thread();
997 self.shared.request_trade_reseed();
1000 self.shared.request_pipeline_refresh();
1001 let theme = load_theme();
1003 self.shared.set_theme(theme);
1004 if theme == crate::tokens::Theme::Light {
1005 self.shared.log_line("[ui] 主题:浅色(读盘 ui.theme)");
1006 }
1007 }
1008
1009 fn on_stop(&self) {
1010 self.shared.log_line("[ui] 界面插件下线");
1011 self.cmd_stop
1013 .store(true, std::sync::atomic::Ordering::Release);
1014 }
1015
1016 fn on_mail(&self, event: &ek_event::Event) {
1017 let ty = event.event_type.as_str();
1018 if ty == EVENT_LOG {
1019 if let Some(d) = event.data.downcast_ref::<LogData>() {
1020 let tag = if d.is_warning_or_above() {
1021 "WARN"
1022 } else {
1023 "INFO"
1024 };
1025 self.shared
1026 .log_line(format!("[{tag}][{}] {}", d.gateway_name, d.msg));
1027 if let Some(note) = d.msg.strip_prefix(crate::state::CHAIN_STATE_PREFIX) {
1033 self.shared.note_chain(&d.gateway_name, note);
1034 }
1035 } else {
1036 self.shared.log_line("[eLog] 载荷类型异常(非 LogData)");
1037 }
1038 return;
1039 }
1040 if ty == ek_ctp::EVENT_STATUS_PREFIX {
1041 if let Some(ev) = event.data.downcast_ref::<ek_ctp::GatewayStatusEvent>() {
1043 self.shared.note_status_event(ev);
1044 } else {
1045 self.shared
1046 .log_line("[eStatus] 载荷类型异常(非 GatewayStatusEvent)");
1047 }
1048 return;
1049 }
1050 if ty == EVENT_PLUG_UP || ty == EVENT_PLUG_DOWN {
1051 if let Some(n) = event.data.downcast_ref::<PlugNotice>() {
1052 let verb = if ty == EVENT_PLUG_UP {
1053 "上线"
1054 } else {
1055 "下线"
1056 };
1057 self.shared
1058 .log_line(format!("[host] 插件{verb} {} v{}", n.name, n.version));
1059 self.shared.request_list_refresh();
1061 if ty == EVENT_PLUG_UP {
1062 if n.name == ek_oms::NAME {
1064 self.shared.request_pipeline_refresh();
1065 }
1066 let is_account = self.shared.accounts().iter().any(|a| a.gateway == n.name);
1071 if n.name == ek_oms::NAME || is_account {
1072 self.shared.request_trade_reseed();
1073 }
1074 if is_account {
1077 self.shared.request_status_seed(&n.name);
1078 }
1079 let md = self.shared.md_source();
1082 let subs = self.shared.subscriptions();
1083 if !md.is_empty() && n.name == md && !subs.is_empty() {
1084 self.shared.push_cmd(crate::state::UiCmd::CtpSubscribe {
1085 gateway: n.name.clone(),
1086 symbols: subs.clone(),
1087 });
1088 self.shared.log_line(format!(
1089 "[ops] 行情源 {} 上线,重放订阅 {} 项",
1090 n.name,
1091 subs.len()
1092 ));
1093 }
1094 } else {
1095 let removed = self.shared.clear_gateway_rows(&n.name);
1099 if removed > 0 {
1100 self.shared.log_line(format!(
1101 "[ui] 插件下线 {}:作废 {} 行交易数据",
1102 n.name, removed
1103 ));
1104 }
1105 if n.name == ek_oms::NAME {
1110 self.shared.request_pipeline_refresh();
1111 }
1112 }
1113 }
1114 self.shared.bump();
1115 return;
1116 }
1117 if ty == EVENT_BAR {
1118 if let Some(f) = event.data.downcast_ref::<ek_dm::wire::BarsFrame>() {
1119 self.shared.market().on_frame(f);
1120 }
1121 self.shared.bump();
1122 return;
1123 }
1124 if ty == EVENT_BAR_UPDATE {
1125 if let Some(b) = event.data.downcast_ref::<ek_data::bar::BarData>() {
1126 self.shared.market().on_update(b);
1127 }
1128 self.shared.bump();
1129 return;
1130 }
1131 if ty == EVENT_TICK {
1132 if let Some(t) = event.data.downcast_ref::<ek_data::tick::TickData>() {
1133 self.shared.update_tick(t);
1134 }
1135 return;
1136 }
1137 if ty == EVENT_CONTRACT {
1138 if let Some(v) = event
1141 .data
1142 .downcast_ref::<Vec<ek_data::contract::ContractData>>()
1143 {
1144 for c in v {
1145 self.shared.update_contract(c);
1146 }
1147 } else if let Some(c) = event.data.downcast_ref::<ek_data::contract::ContractData>() {
1148 self.shared.update_contract(c);
1149 }
1150 return;
1151 }
1152 if ty == ek_ctp::EVENT_CONTRACT_FINISH {
1153 let gw = event
1154 .data
1155 .downcast_ref::<String>()
1156 .cloned()
1157 .unwrap_or_else(|| "?".to_string());
1158 self.shared
1159 .log_line(format!("[ctp] 合约流完成 {gw}(查询合约窗可勾选订阅)"));
1160 return;
1161 }
1162 if ty == W_ORDER {
1165 match event.data.downcast_ref::<OrderData>() {
1166 Some(o) => self.shared.note_order(o),
1167 None => self.shared.note_bad_payload(W_ORDER),
1168 }
1169 return;
1170 }
1171 if ty == W_TRADE {
1172 match event.data.downcast_ref::<TradeData>() {
1173 Some(t) => self.shared.note_trade(t),
1174 None => self.shared.note_bad_payload(W_TRADE),
1175 }
1176 return;
1177 }
1178 if ty == W_POSITION {
1179 match event.data.downcast_ref::<PositionData>() {
1180 Some(p) => self.shared.note_position(p),
1181 None => self.shared.note_bad_payload(W_POSITION),
1182 }
1183 return;
1184 }
1185 if ty == W_ACCOUNT {
1186 match event.data.downcast_ref::<AccountData>() {
1187 Some(a) => self.shared.note_funds(a),
1188 None => self.shared.note_bad_payload(W_ACCOUNT),
1189 }
1190 return;
1191 }
1192 if ty == EVENT_TIMER {
1193 self.refresh();
1194 }
1195 }
1196}