项目结构大调整

.gitignore

@@ -1 +1,2 @@
 /target
+.env

.vscode/launch.json

@@ -0,0 +1,16 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "type": "lldb",
+            "request": "launch",
+            "name": "Debug",
+            "program": "${workspaceFolder}/<executable file>",
+            "args": [],
+            "cwd": "${workspaceFolder}"
+        }
+    ]
+}

.vscode/settings.json

@@ -18,6 +18,7 @@
         "Neng",
         "nextval",
         "oneshot",
+        "otype",
         "plpgsql",
         "prost",
         "proto",
@@ -25,6 +26,7 @@
         "protos",
         "repr",
         "reqwest",
+        "rtsa",
         "rtss",
         "rumqtt",
         "rumqttc",

Cargo.lock

@@ -121,9 +121,15 @@ dependencies = [
 
 [[package]]
 name = "anyhow"
-version = "1.0.90"
+version = "1.0.91"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "37bf3594c4c988a53154954629820791dde498571819ae4ca50ca811e060cc95"
+checksum = "c042108f3ed77fd83760a5fd79b53be043192bb3b9dba91d8c574c0ada7850c8"
+
+[[package]]
+name = "arraydeque"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236"
 
 [[package]]
 name = "ascii_utils"
@@ -133,9 +139,9 @@ checksum = "71938f30533e4d95a6d17aa530939da3842c2ab6f4f84b9dae68447e4129f74a"
 
 [[package]]
 name = "async-executor"
-version = "1.13.0"
+version = "1.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7ebdfa2ebdab6b1760375fa7d6f382b9f486eac35fc994625a00e89280bdbb7"
+checksum = "30ca9a001c1e8ba5149f91a74362376cc6bc5b919d92d988668657bd570bdcec"
 dependencies = [
  "async-task",
  "concurrent-queue",
@@ -564,7 +570,7 @@ dependencies = [
  "ahash",
  "bevy_utils_proc_macros",
  "getrandom",
- "hashbrown 0.14.5",
+ "hashbrown",
  "thread_local",
  "tracing",
  "web-time",
@@ -619,9 +625,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
 [[package]]
 name = "bytes"
-version = "1.7.2"
+version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "428d9aa8fbc0670b7b8d6030a7fadd0f86151cae55e4dbbece15f3780a3dfaf3"
+checksum = "9ac0150caa2ae65ca5bd83f25c7de183dea78d4d366469f148435e2acfbad0da"
 dependencies = [
  "serde",
 ]
@@ -713,14 +719,13 @@ dependencies = [
 
 [[package]]
 name = "config"
-version = "0.14.0"
+version = "0.14.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7328b20597b53c2454f0b1919720c25c7339051c02b72b7e05409e00b14132be"
+checksum = "68578f196d2a33ff61b27fae256c3164f65e36382648e30666dde05b8cc9dfdf"
 dependencies = [
  "async-trait",
  "convert_case",
  "json5",
- "lazy_static",
  "nom",
  "pathdiff",
  "ron",
@@ -728,7 +733,7 @@ dependencies = [
  "serde",
  "serde_json",
  "toml",
- "yaml-rust",
+ "yaml-rust2",
 ]
 
 [[package]]
@@ -1076,9 +1081,9 @@ checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
 
 [[package]]
 name = "flume"
-version = "0.11.0"
+version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "55ac459de2512911e4b674ce33cf20befaba382d05b62b008afc1c8b57cbf181"
+checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -1246,12 +1251,6 @@ dependencies = [
  "thiserror",
 ]
 
-[[package]]
-name = "hashbrown"
-version = "0.13.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43a3c133739dddd0d2990f9a4bdf8eb4b21ef50e4851ca85ab661199821d510e"
-
 [[package]]
 name = "hashbrown"
 version = "0.14.5"
@@ -1263,13 +1262,22 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "hashlink"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8094feaf31ff591f651a2664fb9cfd92bba7a60ce3197265e9482ebe753c8f7"
+dependencies = [
+ "hashbrown",
+]
+
 [[package]]
 name = "hashlink"
 version = "0.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af"
 dependencies = [
- "hashbrown 0.14.5",
+ "hashbrown",
 ]
 
 [[package]]
@@ -1491,7 +1499,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "de3fc2e30ba82dd1b3911c8de1ffc143c74a914a14e99514d7637e3099df5ea0"
 dependencies = [
  "equivalent",
- "hashbrown 0.14.5",
+ "hashbrown",
  "serde",
 ]
 
@@ -1589,12 +1597,6 @@ dependencies = [
  "vcpkg",
 ]
 
-[[package]]
-name = "linked-hash-map"
-version = "0.5.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0717cef1bc8b636c6e1c1bbdefc09e6322da8a9321966e8928ef80d20f7f770f"
-
 [[package]]
 name = "linux-raw-sys"
 version = "0.4.14"
@@ -1623,7 +1625,22 @@ version = "0.12.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "37ee39891760e7d94734f6f63fedc29a2e4a152f836120753a72503f09fcf904"
 dependencies = [
- "hashbrown 0.14.5",
+ "hashbrown",
+]
+
+[[package]]
+name = "manager"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "clap",
+ "config",
+ "enum_dispatch",
+ "rtss_api",
+ "rtss_db",
+ "rtss_log",
+ "serde",
+ "tokio",
 ]
 
 [[package]]
@@ -1834,12 +1851,12 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
 
 [[package]]
 name = "ordered-multimap"
-version = "0.6.0"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4ed8acf08e98e744e5384c8bc63ceb0364e68a6854187221c18df61c4797690e"
+checksum = "49203cdcae0030493bad186b28da2fa25645fa276a51b6fec8010d281e02ef79"
 dependencies = [
  "dlv-list",
- "hashbrown 0.13.2",
+ "hashbrown",
 ]
 
 [[package]]
@@ -2392,15 +2409,12 @@ dependencies = [
  "axum",
  "axum-extra",
  "base64 0.22.1",
- "bevy_ecs",
  "chrono",
  "jsonwebtoken",
  "reqwest",
  "rtss_db",
  "rtss_dto",
  "rtss_log",
- "rtss_sim_manage",
- "rtss_trackside",
  "serde",
  "serde_json",
  "sysinfo",
@@ -2408,22 +2422,12 @@ dependencies = [
  "tower-http",
 ]
 
-[[package]]
-name = "rtss_ci"
-version = "0.1.0"
-
-[[package]]
-name = "rtss_common"
-version = "0.1.0"
-dependencies = [
- "bevy_ecs",
-]
-
 [[package]]
 name = "rtss_db"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "lazy_static",
  "rtss_dto",
  "rtss_log",
  "serde",
@@ -2439,14 +2443,11 @@ dependencies = [
  "async-graphql",
  "prost",
  "prost-build",
+ "prost-types",
  "serde",
  "sqlx",
 ]
 
-[[package]]
-name = "rtss_iscs"
-version = "0.1.0"
-
 [[package]]
 name = "rtss_log"
 version = "0.1.0"
@@ -2463,55 +2464,12 @@ dependencies = [
  "async-trait",
  "bytes",
  "lazy_static",
- "rtss_db",
  "rtss_log",
  "rumqttc",
  "thiserror",
  "tokio",
 ]
 
-[[package]]
-name = "rtss_sim_manage"
-version = "0.1.0"
-dependencies = [
- "bevy_app",
- "bevy_core",
- "bevy_ecs",
- "bevy_time",
- "rayon",
- "rtss_common",
- "rtss_log",
- "rtss_trackside",
- "thiserror",
-]
-
-[[package]]
-name = "rtss_simulation"
-version = "0.1.0"
-dependencies = [
- "anyhow",
- "clap",
- "config",
- "enum_dispatch",
- "rtss_api",
- "rtss_db",
- "rtss_log",
- "serde",
- "tokio",
-]
-
-[[package]]
-name = "rtss_trackside"
-version = "0.1.0"
-dependencies = [
- "bevy_app",
- "bevy_core",
- "bevy_ecs",
- "bevy_time",
- "rtss_common",
- "rtss_log",
-]
-
 [[package]]
 name = "rumqttc"
 version = "0.24.0"
@@ -2533,9 +2491,9 @@ dependencies = [
 
 [[package]]
 name = "rust-ini"
-version = "0.19.0"
+version = "0.20.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e2a3bcec1f113553ef1c88aae6c020a369d03d55b58de9869a0908930385091"
+checksum = "3e0698206bcb8882bf2a9ecb4c1e7785db57ff052297085a6efd4fe42302068a"
 dependencies = [
  "cfg-if",
  "ordered-multimap",
@@ -2686,18 +2644,18 @@ dependencies = [
 
 [[package]]
 name = "serde"
-version = "1.0.210"
+version = "1.0.213"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8e3592472072e6e22e0a54d5904d9febf8508f65fb8552499a1abc7d1078c3a"
+checksum = "3ea7893ff5e2466df8d720bb615088341b295f849602c6956047f8f80f0e9bc1"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.210"
+version = "1.0.213"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "243902eda00fad750862fc144cea25caca5e20d615af0a81bee94ca738f1df1f"
+checksum = "7e85ad2009c50b58e87caa8cd6dac16bdf511bbfb7af6c33df902396aa480fa5"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2706,9 +2664,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.131"
+version = "1.0.132"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "67d42a0bd4ac281beff598909bb56a86acaf979b84483e1c79c10dcaf98f8cf3"
+checksum = "d726bfaff4b320266d395898905d0eba0345aae23b54aee3a737e260fd46db03"
 dependencies = [
  "itoa",
  "memchr",
@@ -2806,6 +2764,29 @@ dependencies = [
  "time",
 ]
 
+[[package]]
+name = "simulation"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "bevy_app",
+ "bevy_core",
+ "bevy_ecs",
+ "bevy_time",
+ "clap",
+ "config",
+ "enum_dispatch",
+ "lazy_static",
+ "rayon",
+ "rtss_db",
+ "rtss_dto",
+ "rtss_log",
+ "rtss_mqtt",
+ "serde",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "slab"
 version = "0.4.9"
@@ -2904,8 +2885,8 @@ dependencies = [
  "futures-intrusive",
  "futures-io",
  "futures-util",
- "hashbrown 0.14.5",
+ "hashbrown",
- "hashlink",
+ "hashlink 0.9.1",
  "hex",
  "indexmap",
  "log",
@@ -3127,9 +3108,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
 
 [[package]]
 name = "syn"
-version = "2.0.76"
+version = "2.0.85"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "578e081a14e0cefc3279b0472138c513f37b41a08d5a3cca9b6e4e8ceb6cd525"
+checksum = "5023162dfcd14ef8f32034d8bcd4cc5ddc61ef7a247c024a33e24e1f24d21b56"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3180,18 +3161,18 @@ dependencies = [
 
 [[package]]
 name = "thiserror"
-version = "1.0.64"
+version = "1.0.65"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d50af8abc119fb8bb6dbabcfa89656f46f84aa0ac7688088608076ad2b459a84"
+checksum = "5d11abd9594d9b38965ef50805c5e469ca9cc6f197f883f717e0269a3057b3d5"
 dependencies = [
  "thiserror-impl",
 ]
 
 [[package]]
 name = "thiserror-impl"
-version = "1.0.64"
+version = "1.0.65"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "08904e7672f5eb876eaaf87e0ce17857500934f4981c4a0ab2b4aa98baac7fc3"
+checksum = "ae71770322cbd277e69d762a16c444af02aa0575ac0d174f0b9562d3b37f8602"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3265,9 +3246,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.40.0"
+version = "1.41.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2b070231665d27ad9ec9b8df639893f46727666c6767db40317fbe920a5d998"
+checksum = "145f3413504347a2be84393cc8a7d2fb4d863b375909ea59f2158261aa258bbb"
 dependencies = [
  "backtrace",
  "bytes",
@@ -3536,9 +3517,9 @@ dependencies = [
 
 [[package]]
 name = "typeid"
-version = "1.0.0"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "059d83cc991e7a42fc37bd50941885db0888e34209f8cfd9aab07ddec03bc9cf"
+checksum = "0e13db2e0ccd5e14a544e8a246ba2312cd25223f616442d7f2cb0e3db614236e"
 
 [[package]]
 name = "typenum"
@@ -4054,12 +4035,14 @@ dependencies = [
 ]
 
 [[package]]
-name = "yaml-rust"
+name = "yaml-rust2"
-version = "0.4.5"
+version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "56c1936c4cc7a1c9ab21a1ebb602eb942ba868cbd44a99cb7cdc5892335e1c85"
+checksum = "8902160c4e6f2fb145dbe9d6760a75e3c9522d8bf796ed7047c85919ac7115f8"
 dependencies = [
- "linked-hash-map",
+ "arraydeque",
+ "encoding_rs",
+ "hashlink 0.8.4",
 ]
 
 [[package]]

Cargo.toml

@@ -1,12 +1,7 @@
-[package]
-name = "rtss_simulation"
-version = "0.1.0"
-edition = "2021"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [workspace]
-members = ["crates/*"]
+members = ["crates/*", "manager/crates/*", "manager", "simulation"]
+resolver = "2"
 
 [workspace.dependencies]
 bevy_app = "0.14"
@@ -14,29 +9,20 @@ bevy_core = "0.14"
 bevy_ecs = "0.14"
 bevy_time = "0.14"
 rayon = "1.10"
-tokio = { version = "1.40", features = ["macros", "rt-multi-thread"] }
+tokio = { version = "1.41.0", features = ["macros", "rt-multi-thread"] }
-thiserror = "1.0.64"
+thiserror = "1.0.65"
 sqlx = { version = "0.8", features = [
     "runtime-tokio",
     "postgres",
     "json",
     "chrono",
 ] }
-serde = { version = "1.0", features = ["derive"] }
+serde = { version = "1.0.213", features = ["derive"] }
-serde_json = "1.0.131"
+serde_json = "1.0.132"
-anyhow = "1.0.90"
+anyhow = "1.0.91"
 async-trait = "0.1.83"
-bytes = "1.7.2"
+bytes = "1.8.0"
 lazy_static = "1.5.0"
-
+config = "0.14.1"
-
-[dependencies]
-tokio = { version = "1.39.3", features = ["macros", "rt-multi-thread"] }
-rtss_log = { path = "crates/rtss_log" }
-rtss_api = { path = "crates/rtss_api" }
-rtss_db = { path = "crates/rtss_db" }
-serde = { workspace = true }
-config = "0.14.0"
 clap = { version = "4.5.20", features = ["derive"] }
 enum_dispatch = "0.3"
-anyhow = { workspace = true }

crates/rtss_api/src/apis/simulation.rs

@@ -1,105 +0,0 @@
-use async_graphql::{Context, InputObject, Object};
-use rtss_sim_manage::{AvailablePlugins, SimulationBuilder};
-
-use super::{MutexSimulationManager, SimulationOperation};
-
-#[derive(Default)]
-pub struct SimulationQuery;
-
-#[Object]
-impl SimulationQuery {
-    async fn simulations<'ctx>(&self, ctx: &Context<'ctx>) -> usize {
-        let sim = ctx.data::<MutexSimulationManager>().unwrap();
-        sim.lock().await.count()
-    }
-}
-
-#[derive(Default)]
-pub struct SimulationMutation;
-
-#[Object]
-impl SimulationMutation {
-    async fn start_simulation<'ctx>(
-        &self,
-        ctx: &Context<'ctx>,
-        req: StartSimulationRequest,
-    ) -> async_graphql::Result<String> {
-        // let claims = ctx.data::<Option<Claims>>().unwrap();
-        // match claims {
-        //     Some(claims) => {
-        //         info!("User {claims:?} started simulation");
-        //     }
-        //     _ => return Err("Unauthorized".into()),
-        // }
-        let sim = ctx.data::<MutexSimulationManager>().unwrap();
-        let id = sim.lock().await.start_simulation(
-            SimulationBuilder::default()
-                .id(req.user_id)
-                .plugins(vec![AvailablePlugins::TrackSideEquipmentPlugin]),
-        )?;
-        Ok(id)
-    }
-
-    async fn exit_simulation<'ctx>(
-        &self,
-        ctx: &Context<'ctx>,
-        id: String,
-    ) -> async_graphql::Result<bool> {
-        let sim = ctx.data::<MutexSimulationManager>().unwrap();
-        sim.lock().await.exit_simulation(id)?;
-        Ok(true)
-    }
-
-    async fn pause_simulation<'ctx>(
-        &self,
-        ctx: &Context<'ctx>,
-        id: String,
-    ) -> async_graphql::Result<bool> {
-        let sim = ctx.data::<MutexSimulationManager>().unwrap();
-        sim.lock().await.pause_simulation(id)?;
-        Ok(true)
-    }
-
-    async fn resume_simulation<'ctx>(
-        &self,
-        ctx: &Context<'ctx>,
-        id: String,
-    ) -> async_graphql::Result<bool> {
-        let sim = ctx.data::<MutexSimulationManager>().unwrap();
-        sim.lock().await.resume_simulation(id)?;
-        Ok(true)
-    }
-
-    async fn update_simulation_speed<'ctx>(
-        &self,
-        ctx: &Context<'ctx>,
-        id: String,
-        speed: f32,
-    ) -> async_graphql::Result<bool> {
-        let sim = ctx.data::<MutexSimulationManager>().unwrap();
-        sim.lock().await.update_simulation_speed(id, speed)?;
-        Ok(true)
-    }
-
-    async fn trigger_simulation_operation<'ctx>(
-        &self,
-        ctx: &Context<'ctx>,
-        id: String,
-        entity_uid: String,
-        operation: SimulationOperation,
-    ) -> async_graphql::Result<bool> {
-        let sim = ctx.data::<MutexSimulationManager>().unwrap();
-        sim.lock().await.trigger_entity_operation(
-            id,
-            entity_uid,
-            operation.to_operation_event(),
-        )?;
-        Ok(true)
-    }
-}
-
-#[derive(InputObject)]
-struct StartSimulationRequest {
-    user_id: String,
-    sim_def_id: String,
-}

crates/rtss_api/src/apis/simulation_definition.rs

@@ -1,35 +0,0 @@
-use std::ops::Deref;
-
-use async_graphql::Enum;
-use bevy_ecs::event::Event;
-use rtss_sim_manage::SimulationManager;
-use tokio::sync::Mutex;
-
-pub struct MutexSimulationManager(Mutex<SimulationManager>);
-impl Default for MutexSimulationManager {
-    fn default() -> Self {
-        Self(Mutex::new(SimulationManager::default()))
-    }
-}
-impl Deref for MutexSimulationManager {
-    type Target = Mutex<SimulationManager>;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-
-#[derive(Enum, Copy, Clone, Eq, PartialEq, Debug)]
-pub enum SimulationOperation {
-    TurnoutControlDC,
-    TurnoutControlFC,
-}
-
-impl SimulationOperation {
-    pub fn to_operation_event(self) -> impl Event + Copy {
-        match self {
-            SimulationOperation::TurnoutControlDC => rtss_trackside::TurnoutControlEvent::DC,
-            SimulationOperation::TurnoutControlFC => rtss_trackside::TurnoutControlEvent::FC,
-        }
-    }
-}

crates/rtss_db/Cargo.toml

@@ -17,6 +17,7 @@ sqlx = { workspace = true, features = [
     "uuid",
 ] }
 thiserror = { workspace = true }
+lazy_static = { workspace = true }
 
 rtss_dto = { path = "../rtss_dto" }
 rtss_log = { path = "../rtss_log" }

crates/rtss_db/src/db_access/draft_data.rs

@@ -458,7 +458,7 @@ mod tests {
                 id: i + 1,
                 name: format!("user{}", i + 1),
                 password: "password".to_string(),
-                roles: serde_json::to_value(&vec![Role::User]).unwrap(),
+                roles: serde_json::to_value(vec![Role::User]).unwrap(),
                 email: None,
                 mobile: None,
                 created_at: Local::now(),

crates/rtss_db/src/db_access/feature.rs

@@ -236,7 +236,7 @@ mod tests {
                 id: i + 1,
                 name: format!("user{}", i + 1),
                 password: "password".to_string(),
-                roles: serde_json::to_value(&vec![Role::User]).unwrap(),
+                roles: serde_json::to_value(vec![Role::User]).unwrap(),
                 email: None,
                 mobile: None,
                 created_at: Local::now(),
@@ -260,7 +260,7 @@ mod tests {
         println!("create feature: {:?}", feature);
         assert_eq!(feature.creator_id, creator_id);
         assert_eq!(feature.updater_id, creator_id);
-        assert_eq!(feature.is_published, true);
+        assert!(feature.is_published);
         assert_eq!(feature.config, create_config);
         let feature_id = feature.id;
         // 获取功能特性测试
@@ -269,7 +269,7 @@ mod tests {
         assert_eq!(feature.id, feature_id);
         assert_eq!(feature.creator_id, creator_id);
         assert_eq!(feature.updater_id, creator_id);
-        assert_eq!(feature.is_published, true);
+        assert!(feature.is_published);
         assert_eq!(feature.config, create_config);
         // 更新测试
         let updater_id = 2;
@@ -279,7 +279,7 @@ mod tests {
             name: "test update".to_string(),
             description: "test update".to_string(),
             config: config.clone(),
-            updater_id: updater_id,
+            updater_id,
         };
         let feature = accessor.update_feature(&feature).await?;
         println!("update feature: {:?}", feature);
@@ -289,7 +289,7 @@ mod tests {
         // 上下架测试
         let feature = accessor.set_feature_published(feature_id, false).await?;
         println!("set feature published: {:?}", feature);
-        assert_eq!(feature.is_published, false);
+        assert!(!feature.is_published);
         // 分页查询测试
         // 创建10个功能特性，5个发布的，5个未发布的
         for i in 0..10 {

crates/rtss_db/src/db_access/mod.rs

@@ -1,14 +1,42 @@
 mod draft_data;
+use std::sync::Mutex;
+
 pub use draft_data::*;
 mod release_data;
 pub use release_data::*;
 mod user;
+use rtss_log::tracing::error;
 pub use user::*;
 mod feature;
 pub use feature::*;
+use lazy_static::lazy_static;
 
 use crate::{model::MqttClientIdSeq, DbAccessError};
 
+lazy_static! {
+    static ref RDA: Mutex<Option<RtssDbAccessor>> = Mutex::new(None);
+}
+
+/// 初始化全局默认数据库访问器
+pub async fn init_default_db_accessor(url: &str) {
+    if RDA.lock().unwrap().is_some() {
+        error!("数据库访问器已初始化，请勿重复初始化");
+        return;
+    }
+    let accessor = get_db_accessor(url).await;
+    let mut rda = RDA.lock().unwrap();
+    *rda = Some(accessor);
+}
+
+/// 获取默认数据库访问器
+pub fn get_default_db_accessor() -> RtssDbAccessor {
+    let rda = RDA.lock().unwrap();
+    if rda.is_none() {
+        panic!("数据库访问器未初始化");
+    }
+    rda.as_ref().unwrap().clone()
+}
+
 #[derive(Clone)]
 pub struct RtssDbAccessor {
     pool: sqlx::PgPool,

crates/rtss_db/src/db_access/release_data.rs

@@ -704,7 +704,7 @@ mod tests {
                 id: i + 1,
                 name: format!("user{}", i + 1),
                 password: "password".to_string(),
-                roles: serde_json::to_value(&vec![Role::User]).unwrap(),
+                roles: serde_json::to_value(vec![Role::User]).unwrap(),
                 email: None,
                 mobile: None,
                 created_at: Local::now(),
@@ -745,7 +745,7 @@ mod tests {
 
         // 发布新版本
         let (release_data, version1) = accessor
-            .release_new_from_draft(draft.id, name, &description, None)
+            .release_new_from_draft(draft.id, name, description, None)
             .await?;
         assert_eq!(release_data.name, name);
         // 检查使用版本是刚刚发布的版本
@@ -759,7 +759,7 @@ mod tests {
         // 检查版本描述
         assert_eq!(version1.description, description);
         // 检查上架状态
-        assert_eq!(release_data.is_published, true);
+        assert!(release_data.is_published);
         // 检查草稿数据的默认发布数据id
         let draft = accessor.query_draft_data_by_id(draft.id).await?;
         assert_eq!(draft.default_release_data_id, Some(release_data.id));
@@ -769,7 +769,7 @@ mod tests {
         let exist = accessor
             .is_release_data_name_exist(rtss_dto::common::DataType::Iscs, name)
             .await?;
-        assert_eq!(exist, true);
+        assert!(exist);
 
         // 修改草稿数据
         let data = "test2".as_bytes();
@@ -798,7 +798,7 @@ mod tests {
         let release_data = accessor
             .set_release_data_published(release_data.id, false)
             .await?;
-        assert_eq!(release_data.is_published, false);
+        assert!(!release_data.is_published);
         assert!(release_data.updated_at > release_data.created_at);
         println!("更新发布数据上架状态测试成功");
 

crates/rtss_db/src/db_access/user.rs

@@ -300,7 +300,7 @@ mod tests {
                 id: 1,
                 name: "test1".to_string(),
                 password: "password".to_string(),
-                roles: serde_json::to_value(&vec![Role::User]).unwrap(),
+                roles: serde_json::to_value(vec![Role::User]).unwrap(),
                 email: None,
                 mobile: None,
                 created_at: Local::now(),
@@ -310,7 +310,7 @@ mod tests {
                 id: 2,
                 name: "test2".to_string(),
                 password: "password".to_string(),
-                roles: serde_json::to_value(&vec![Role::Admin]).unwrap(),
+                roles: serde_json::to_value(vec![Role::Admin]).unwrap(),
                 email: None,
                 mobile: None,
                 created_at: Local::now(),
@@ -342,7 +342,7 @@ mod tests {
                 id: 1,
                 name: "test1".to_string(),
                 password: "password".to_string(),
-                roles: serde_json::to_value(&vec![Role::User]).unwrap(),
+                roles: serde_json::to_value(vec![Role::User]).unwrap(),
                 email: Some("walker@163.com".to_string()),
                 mobile: None,
                 created_at: Local::now() - Duration::from_secs(60),
@@ -352,7 +352,7 @@ mod tests {
                 id: 2,
                 name: "test2".to_string(),
                 password: "password".to_string(),
-                roles: serde_json::to_value(&vec![Role::Admin, Role::User]).unwrap(),
+                roles: serde_json::to_value(vec![Role::Admin, Role::User]).unwrap(),
                 email: None,
                 mobile: Some("123456789".to_string()),
                 created_at: Local::now() - Duration::from_secs(60),
@@ -362,7 +362,7 @@ mod tests {
                 id: 3,
                 name: "test3".to_string(),
                 password: "password".to_string(),
-                roles: serde_json::to_value(&vec![Role::User]).unwrap(),
+                roles: serde_json::to_value(vec![Role::User]).unwrap(),
                 email: None,
                 mobile: None,
                 created_at: Local::now(),

crates/rtss_dto/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2021"
 
 [dependencies]
 prost = "0.13"
+prost-types = "0.13"
 async-graphql = { version = "7.0.7", features = ["chrono"] }
 sqlx = { workspace = true }
 serde = { workspace = true }

crates/rtss_dto/build.rs

@@ -42,6 +42,7 @@ fn main() {
                 "../../rtss-proto-msg/src/em_data.proto",
                 "../../rtss-proto-msg/src/common.proto",
                 "../../rtss-proto-msg/src/iscs_graphic_data.proto",
+                "../../rtss-proto-msg/src/simulation.proto",
             ],
             &["../../rtss-proto-msg/src/"],
         )

crates/rtss_dto/src/lib.rs

@@ -1,3 +1,27 @@
 mod pb;
 
 pub use pb::*;
+
+#[cfg(test)]
+mod tests {
+
+    use crate::simulation::{self, operation, SetSpeedParam};
+
+    #[test]
+    fn test_prost_any() {
+        let op = simulation::Operation {
+            otype: simulation::OperationType::SetSpeed as i32,
+            param: Some(operation::Param::SetSpeedParam(SetSpeedParam {
+                speed: 1.0,
+            })),
+        };
+        println!("{:?}", op);
+        if let Some(param) = op.param {
+            match param {
+                operation::Param::SetSpeedParam(ssp) => {
+                    assert_eq!(ssp.speed, 1.0);
+                }
+            }
+        }
+    }
+}

crates/rtss_dto/src/pb/iscs_graphic_data.rs

@@ -3,11 +3,15 @@
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct IscsGraphicStorage {
     #[prost(message, repeated, tag = "1")]
-    pub cctv_of_station_control_storages: ::prost::alloc::vec::Vec<
+    pub cctv_of_equipment_layout_storages: ::prost::alloc::vec::Vec<
-        CctvOfStationControlStorage,
+        CctvOfEquipmentLayoutStorage,
     >,
     #[prost(message, repeated, tag = "2")]
-    pub fas_platform_alarm_storages: ::prost::alloc::vec::Vec<FasPlatformAlarmStorage>,
+    pub fas_of_platform_alarm_storages: ::prost::alloc::vec::Vec<
+        FasOfPlatformAlarmStorage,
+    >,
+    #[prost(message, repeated, tag = "3")]
+    pub bas_of_escalator_storages: ::prost::alloc::vec::Vec<BasOfEscalatorStorage>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -24,6 +28,22 @@ pub struct UniqueIdOfStationLayout {
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct CommonGraphicStorage {
+    #[prost(message, repeated, tag = "1")]
+    pub arrows: ::prost::alloc::vec::Vec<Arrow>,
+    #[prost(message, repeated, tag = "2")]
+    pub texts: ::prost::alloc::vec::Vec<Text>,
+    #[prost(message, repeated, tag = "3")]
+    pub rects: ::prost::alloc::vec::Vec<Rect>,
+    #[prost(message, repeated, tag = "4")]
+    pub lines: ::prost::alloc::vec::Vec<Line>,
+    #[prost(message, repeated, tag = "5")]
+    pub circles: ::prost::alloc::vec::Vec<Circle>,
+    #[prost(message, repeated, tag = "6")]
+    pub buttons: ::prost::alloc::vec::Vec<Button>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Arrow {
     #[prost(message, optional, tag = "1")]
     pub common: ::core::option::Option<super::common::CommonInfo>,
@@ -32,10 +52,9 @@ pub struct Arrow {
     #[prost(message, repeated, tag = "3")]
     pub points: ::prost::alloc::vec::Vec<super::common::Point>,
 }
-/// Iscs文字
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct IscsText {
+pub struct Text {
     #[prost(message, optional, tag = "1")]
     pub common: ::core::option::Option<super::common::CommonInfo>,
     #[prost(string, tag = "2")]
@@ -72,20 +91,92 @@ pub struct Rect {
     /// 画第一个点的坐标
     #[prost(message, optional, tag = "8")]
     pub point: ::core::option::Option<super::common::Point>,
+    /// 填充色
+    #[prost(string, tag = "9")]
+    pub fill_color: ::prost::alloc::string::String,
+    /// 透明度
+    #[prost(float, tag = "10")]
+    pub alpha: f32,
 }
-/// CCTV按钮
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct CctvButton {
+pub struct Line {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    /// 编号
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+    /// 点列表
+    #[prost(message, repeated, tag = "3")]
+    pub points: ::prost::alloc::vec::Vec<super::common::Point>,
+    /// 是否曲线
+    #[prost(bool, tag = "4")]
+    pub is_curve: bool,
+    /// 曲线分段数
+    #[prost(int32, tag = "5")]
+    pub segments_count: i32,
+    /// 线宽
+    #[prost(int32, tag = "6")]
+    pub line_width: i32,
+    /// 线色
+    #[prost(string, tag = "7")]
+    pub line_color: ::prost::alloc::string::String,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Circle {
     #[prost(message, optional, tag = "1")]
     pub common: ::core::option::Option<super::common::CommonInfo>,
     #[prost(string, tag = "2")]
     pub code: ::prost::alloc::string::String,
-    #[prost(enumeration = "cctv_button::ButtonType", tag = "3")]
+    #[prost(message, optional, tag = "3")]
-    pub button_type: i32,
+    pub position: ::core::option::Option<super::common::Point>,
+    #[prost(float, tag = "4")]
+    pub radius: f32,
+    #[prost(int32, tag = "5")]
+    pub line_width: i32,
+    #[prost(string, tag = "6")]
+    pub line_color: ::prost::alloc::string::String,
+    #[prost(string, tag = "7")]
+    pub fill_color: ::prost::alloc::string::String,
+    #[prost(float, tag = "8")]
+    pub alpha: f32,
 }
-/// Nested message and enum types in `CCTVButton`.
+#[allow(clippy::derive_partial_eq_without_eq)]
-pub mod cctv_button {
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Button {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+    #[prost(string, tag = "3")]
+    pub code_color: ::prost::alloc::string::String,
+    #[prost(int32, tag = "4")]
+    pub code_font_size: i32,
+    /// 所属ISCS子菜单
+    #[prost(string, tag = "5")]
+    pub belong_sub_menu: ::prost::alloc::string::String,
+    /// 类似icon
+    #[prost(enumeration = "button::ButtonType", tag = "6")]
+    pub button_type: i32,
+    /// 宽度
+    #[prost(float, tag = "7")]
+    pub width: f32,
+    /// 高度
+    #[prost(float, tag = "8")]
+    pub height: f32,
+    /// 圆角半径_可变圆
+    #[prost(int32, tag = "9")]
+    pub radius: i32,
+    /// 填充色
+    #[prost(string, tag = "10")]
+    pub fill_color: ::prost::alloc::string::String,
+    /// 透明度
+    #[prost(float, tag = "11")]
+    pub alpha: f32,
+}
+/// Nested message and enum types in `Button`.
+pub mod button {
     #[derive(
         Clone,
         Copy,
@@ -99,11 +190,14 @@ pub mod cctv_button {
     )]
     #[repr(i32)]
     pub enum ButtonType {
-        Rect = 0,
+        /// 没有Icon
+        NoIcon = 0,
+        /// 监控方形
+        CctvRect = 1,
         /// 监控样子的按钮
-        Monitor = 1,
+        CctvMonitor = 2,
         /// 半圆样子的按钮
-        Semicircle = 2,
+        CctvSemicircle = 3,
     }
     impl ButtonType {
         /// String value of the enum field names used in the ProtoBuf definition.
@@ -112,17 +206,19 @@ pub mod cctv_button {
         /// (if the ProtoBuf definition does not change) and safe for programmatic use.
         pub fn as_str_name(&self) -> &'static str {
             match self {
-                ButtonType::Rect => "rect",
+                ButtonType::NoIcon => "noIcon",
-                ButtonType::Monitor => "monitor",
+                ButtonType::CctvRect => "cctvRect",
-                ButtonType::Semicircle => "semicircle",
+                ButtonType::CctvMonitor => "cctvMonitor",
+                ButtonType::CctvSemicircle => "cctvSemicircle",
             }
         }
         /// Creates an enum from field names used in the ProtoBuf definition.
         pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
             match value {
-                "rect" => Some(Self::Rect),
+                "noIcon" => Some(Self::NoIcon),
-                "monitor" => Some(Self::Monitor),
+                "cctvRect" => Some(Self::CctvRect),
-                "semicircle" => Some(Self::Semicircle),
+                "cctvMonitor" => Some(Self::CctvMonitor),
+                "cctvSemicircle" => Some(Self::CctvSemicircle),
                 _ => None,
             }
         }
@@ -175,31 +271,392 @@ pub struct TemperatureDetector {
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct CctvOfStationControlStorage {
+pub struct BasOfEscalatorStorage {
     #[prost(string, tag = "1")]
     pub station_name: ::prost::alloc::string::String,
     #[prost(message, optional, tag = "2")]
     pub canvas: ::core::option::Option<super::common::Canvas>,
-    #[prost(message, repeated, tag = "3")]
+    #[prost(message, optional, tag = "3")]
-    pub arrows: ::prost::alloc::vec::Vec<Arrow>,
+    pub common_graphic_storage: ::core::option::Option<CommonGraphicStorage>,
     #[prost(message, repeated, tag = "4")]
-    pub iscs_texts: ::prost::alloc::vec::Vec<IscsText>,
+    pub escalators: ::prost::alloc::vec::Vec<Escalator>,
     #[prost(message, repeated, tag = "5")]
-    pub rects: ::prost::alloc::vec::Vec<Rect>,
+    pub vertical_elevators: ::prost::alloc::vec::Vec<VerticalElevator>,
-    #[prost(message, repeated, tag = "6")]
+}
-    pub cctv_buttons: ::prost::alloc::vec::Vec<CctvButton>,
+/// 自动扶梯
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Escalator {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 垂直电梯
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct VerticalElevator {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
-pub struct FasPlatformAlarmStorage {
+pub struct CctvOfEquipmentLayoutStorage {
     #[prost(string, tag = "1")]
     pub station_name: ::prost::alloc::string::String,
     #[prost(message, optional, tag = "2")]
     pub canvas: ::core::option::Option<super::common::Canvas>,
-    #[prost(message, repeated, tag = "3")]
+    #[prost(message, optional, tag = "3")]
-    pub arrows: ::prost::alloc::vec::Vec<Arrow>,
+    pub common_graphic_storage: ::core::option::Option<CommonGraphicStorage>,
-    #[prost(message, repeated, tag = "4")]
+    /// 分层
-    pub iscs_texts: ::prost::alloc::vec::Vec<IscsText>,
+    #[prost(enumeration = "cctv_of_equipment_layout_storage::LayerType", tag = "4")]
-    #[prost(message, repeated, tag = "5")]
+    pub layer: i32,
-    pub rects: ::prost::alloc::vec::Vec<Rect>,
+}
+/// Nested message and enum types in `CCTVOfEquipmentLayoutStorage`.
+pub mod cctv_of_equipment_layout_storage {
+    #[derive(
+        Clone,
+        Copy,
+        Debug,
+        PartialEq,
+        Eq,
+        Hash,
+        PartialOrd,
+        Ord,
+        ::prost::Enumeration
+    )]
+    #[repr(i32)]
+    pub enum LayerType {
+        /// 站厅
+        StationHall = 0,
+        /// 站台
+        Platform = 1,
+        /// 云台
+        Ptz = 2,
+    }
+    impl LayerType {
+        /// String value of the enum field names used in the ProtoBuf definition.
+        ///
+        /// The values are not transformed in any way and thus are considered stable
+        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+        pub fn as_str_name(&self) -> &'static str {
+            match self {
+                LayerType::StationHall => "StationHall",
+                LayerType::Platform => "platform",
+                LayerType::Ptz => "PTZ",
+            }
+        }
+        /// Creates an enum from field names used in the ProtoBuf definition.
+        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+            match value {
+                "StationHall" => Some(Self::StationHall),
+                "platform" => Some(Self::Platform),
+                "PTZ" => Some(Self::Ptz),
+                _ => None,
+            }
+        }
+    }
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct FasOfPlatformAlarmStorage {
+    #[prost(string, tag = "1")]
+    pub station_name: ::prost::alloc::string::String,
+    #[prost(message, optional, tag = "2")]
+    pub canvas: ::core::option::Option<super::common::Canvas>,
+    #[prost(message, optional, tag = "3")]
+    pub common_graphic_storage: ::core::option::Option<CommonGraphicStorage>,
+    /// 分区
+    #[prost(string, tag = "4")]
+    pub partition: ::prost::alloc::string::String,
+    #[prost(message, repeated, tag = "5")]
+    pub fas_failure_control_hosts: ::prost::alloc::vec::Vec<FasFailureControlHost>,
+    #[prost(message, repeated, tag = "6")]
+    pub fas_alarms: ::prost::alloc::vec::Vec<FasAlarm>,
+    #[prost(message, repeated, tag = "7")]
+    pub manual_alarm_buttons: ::prost::alloc::vec::Vec<ManualAlarmButton>,
+    #[prost(message, repeated, tag = "8")]
+    pub hydrant_alarm_buttons: ::prost::alloc::vec::Vec<HydrantAlarmButton>,
+    #[prost(message, repeated, tag = "9")]
+    pub gas_extinguishings: ::prost::alloc::vec::Vec<GasExtinguishing>,
+    #[prost(message, repeated, tag = "10")]
+    pub smoke_detectors: ::prost::alloc::vec::Vec<SmokeDetector>,
+    #[prost(message, repeated, tag = "11")]
+    pub temperature_detectors: ::prost::alloc::vec::Vec<TemperatureDetector>,
+    #[prost(message, repeated, tag = "12")]
+    pub fire_shutters: ::prost::alloc::vec::Vec<FireShutter>,
+    #[prost(message, repeated, tag = "13")]
+    pub fire_pumps: ::prost::alloc::vec::Vec<FirePump>,
+    #[prost(message, repeated, tag = "14")]
+    pub spray_pumps: ::prost::alloc::vec::Vec<SprayPump>,
+    #[prost(message, repeated, tag = "15")]
+    pub stabilized_pressure_pumps: ::prost::alloc::vec::Vec<StabilizedPressurePump>,
+    #[prost(message, repeated, tag = "16")]
+    pub acs: ::prost::alloc::vec::Vec<Acs>,
+    #[prost(message, repeated, tag = "17")]
+    pub afc: ::prost::alloc::vec::Vec<Afc>,
+    #[prost(message, repeated, tag = "18")]
+    pub non_fire_power_supplies: ::prost::alloc::vec::Vec<NonFirePowerSupply>,
+    #[prost(message, repeated, tag = "19")]
+    pub water_flow_indicators: ::prost::alloc::vec::Vec<WaterFlowIndicator>,
+    #[prost(message, repeated, tag = "20")]
+    pub signal_butterfly_valves: ::prost::alloc::vec::Vec<SignalButterflyValve>,
+    #[prost(message, repeated, tag = "21")]
+    pub pressure_switches: ::prost::alloc::vec::Vec<PressureSwitch>,
+    #[prost(message, repeated, tag = "22")]
+    pub fault_valves: ::prost::alloc::vec::Vec<FaultValve>,
+    #[prost(message, repeated, tag = "23")]
+    pub start_pump_buttons: ::prost::alloc::vec::Vec<StartPumpButton>,
+    #[prost(message, repeated, tag = "24")]
+    pub temperature_cables: ::prost::alloc::vec::Vec<TemperatureCable>,
+    #[prost(message, repeated, tag = "25")]
+    pub emergency_lightings: ::prost::alloc::vec::Vec<EmergencyLighting>,
+    #[prost(message, repeated, tag = "26")]
+    pub elevator_lift_to_tops: ::prost::alloc::vec::Vec<ElevatorLiftToTop>,
+    #[prost(message, repeated, tag = "27")]
+    pub electric_butterfly_valves: ::prost::alloc::vec::Vec<ElectricButterflyValve>,
+    #[prost(message, repeated, tag = "28")]
+    pub fire_valves: ::prost::alloc::vec::Vec<FireValve>,
+    #[prost(message, repeated, tag = "29")]
+    pub electric_fire_extinguishing_valves: ::prost::alloc::vec::Vec<
+        ElectricFireExtinguishingValve,
+    >,
+    #[prost(message, repeated, tag = "30")]
+    pub fire_intercommunication_signals: ::prost::alloc::vec::Vec<
+        FireIntercommunicationSignal,
+    >,
+}
+/// 火灾故障控制主机
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct FasFailureControlHost {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 警铃
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct FasAlarm {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 防火卷帘
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct FireShutter {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+    #[prost(enumeration = "fire_shutter::ShutterType", tag = "3")]
+    pub r#type: i32,
+}
+/// Nested message and enum types in `FireShutter`.
+pub mod fire_shutter {
+    #[derive(
+        Clone,
+        Copy,
+        Debug,
+        PartialEq,
+        Eq,
+        Hash,
+        PartialOrd,
+        Ord,
+        ::prost::Enumeration
+    )]
+    #[repr(i32)]
+    pub enum ShutterType {
+        /// 隔断型
+        Partition = 0,
+        /// 疏散型
+        Dispersal = 1,
+    }
+    impl ShutterType {
+        /// String value of the enum field names used in the ProtoBuf definition.
+        ///
+        /// The values are not transformed in any way and thus are considered stable
+        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+        pub fn as_str_name(&self) -> &'static str {
+            match self {
+                ShutterType::Partition => "partition",
+                ShutterType::Dispersal => "dispersal",
+            }
+        }
+        /// Creates an enum from field names used in the ProtoBuf definition.
+        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+            match value {
+                "partition" => Some(Self::Partition),
+                "dispersal" => Some(Self::Dispersal),
+                _ => None,
+            }
+        }
+    }
+}
+/// 消防泵
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct FirePump {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 喷淋泵
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct SprayPump {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 稳压泵
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct StabilizedPressurePump {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// ACS
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Acs {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// AFC
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Afc {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 非消防电源
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct NonFirePowerSupply {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 水流指示器
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct WaterFlowIndicator {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 信号蝶阀
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct SignalButterflyValve {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 压力开关
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PressureSwitch {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 故障阀
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct FaultValve {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 启泵按钮
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct StartPumpButton {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 感温电缆
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TemperatureCable {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 应急照明
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct EmergencyLighting {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 电梯归首
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ElevatorLiftToTop {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 电动蝶阀
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ElectricButterflyValve {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 防火阀
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct FireValve {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 电动防烟防火阀
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ElectricFireExtinguishingValve {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
+}
+/// 火灾互联互通信号
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct FireIntercommunicationSignal {
+    #[prost(message, optional, tag = "1")]
+    pub common: ::core::option::Option<super::common::CommonInfo>,
+    #[prost(string, tag = "2")]
+    pub code: ::prost::alloc::string::String,
 }

crates/rtss_dto/src/pb/mod.rs

@@ -1,2 +1,3 @@
 pub mod common;
 pub mod em_data;
+pub mod simulation;

crates/rtss_dto/src/pb/simulation.rs

@@ -0,0 +1,77 @@
+// This file is @generated by prost-build.
+/// 仿真操作
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct Operation {
+    /// 操作类型
+    #[prost(enumeration = "OperationType", tag = "1")]
+    pub otype: i32,
+    /// 操作参数
+    #[prost(oneof = "operation::Param", tags = "2")]
+    pub param: ::core::option::Option<operation::Param>,
+}
+/// Nested message and enum types in `Operation`.
+pub mod operation {
+    /// 操作参数
+    #[allow(clippy::derive_partial_eq_without_eq)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
+    pub enum Param {
+        /// 设置仿真运行速度参数
+        #[prost(message, tag = "2")]
+        SetSpeedParam(super::SetSpeedParam),
+    }
+}
+/// 设置仿真运行速度参数
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct SetSpeedParam {
+    /// 运行速度
+    #[prost(float, tag = "1")]
+    pub speed: f32,
+}
+/// 仿真操作类型
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum OperationType {
+    /// 未知
+    Unknown = 0,
+    /// -------仿真控制操作--------
+    /// 暂停
+    Pause = 1,
+    /// 恢复运行
+    Unpause = 2,
+    /// 重置
+    Reset = 3,
+    /// 设置运行速度
+    SetSpeed = 4,
+    /// 销毁
+    Destroy = 5,
+}
+impl OperationType {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            OperationType::Unknown => "Unknown",
+            OperationType::Pause => "Pause",
+            OperationType::Unpause => "Unpause",
+            OperationType::Reset => "Reset",
+            OperationType::SetSpeed => "SetSpeed",
+            OperationType::Destroy => "Destroy",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "Unknown" => Some(Self::Unknown),
+            "Pause" => Some(Self::Pause),
+            "Unpause" => Some(Self::Unpause),
+            "Reset" => Some(Self::Reset),
+            "SetSpeed" => Some(Self::SetSpeed),
+            "Destroy" => Some(Self::Destroy),
+            _ => None,
+        }
+    }
+}

crates/rtss_mqtt/Cargo.toml

@@ -5,11 +5,10 @@ edition = "2021"
 
 [dependencies]
 rumqttc = { version = "0.24.0", features = ["url"] }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["sync"] }
 async-trait = { workspace = true }
 bytes = { workspace = true }
 lazy_static = { workspace = true }
 thiserror = { workspace = true }
 
-rtss_db = { path = "../rtss_db" }
 rtss_log = { path = "../rtss_log" }

crates/rtss_mqtt/src/lib.rs

@@ -1,3 +1,4 @@
+use core::panic;
 use std::{
     any::TypeId,
     collections::HashMap,
@@ -11,7 +12,7 @@ use std::{
 
 use bytes::Bytes;
 use lazy_static::lazy_static;
-use rtss_log::tracing::{debug, error, info};
+use rtss_log::tracing::{debug, error, info, trace};
 use rumqttc::{
     v5::{
         mqttbytes::{
@@ -29,9 +30,19 @@ use error::MqttClientError;
 
 lazy_static! {
 /// 全局静态MqttClient实例
+/// 使用注意事项:
+/// 每次订阅/发布/请求时都通过get_global_mqtt_client获取新的实例，否则可能会出现死锁
 static ref MQTT_CLIENT: tokio::sync::Mutex<Option<MqttClient>> = tokio::sync::Mutex::new(None);
 }
 
+/// 初始化全局MqttClient实例
+pub async fn init_global_mqtt_client(
+    mut options: MqttClientOptions,
+) -> Result<(), MqttClientError> {
+    let client = options.build();
+    set_global_mqtt_client(client).await
+}
+
 /// 设置全局MqttClient实例
 pub async fn set_global_mqtt_client(client: MqttClient) -> Result<(), MqttClientError> {
     let mut mqtt_client = MQTT_CLIENT.lock().await;
@@ -42,14 +53,20 @@ pub async fn set_global_mqtt_client(client: MqttClient) -> Result<(), MqttClient
     Ok(())
 }
 
-pub async fn get_global_mqtt_client() -> Option<MqttClient> {
+/// 获取全局MqttClient实例
+pub async fn get_global_mqtt_client() -> MqttClient {
     let mqtt_client = MQTT_CLIENT.lock().await;
-    mqtt_client.clone()
+    if let Some(client) = mqtt_client.as_ref() {
+        return client.clone();
+    }
+    panic!("MqttClient未初始化: 使用init_global_mqtt_client初始化，或者在main函数中调用set_global_mqtt_client设置");
 }
 
 pub struct MqttClientOptions {
     id: String,
     options: MqttOptions,
+    /// mqtt客户端请求队列的最大容量
+    max_cap: usize,
     request_timeout: Duration,
 }
 
@@ -59,37 +76,40 @@ impl MqttClientOptions {
             id: id.to_string(),
             options: MqttOptions::parse_url(format!("{}?client_id={}", url, id))
                 .expect("解析mqtt url失败"),
+            max_cap: 30,
             request_timeout: Duration::from_secs(5),
         }
     }
 
-    pub fn set_request_timeout(&mut self, timeout: Duration) -> &mut Self {
+    pub fn set_max_cap(mut self, max_cap: usize) -> Self {
+        self.max_cap = max_cap;
+        self
+    }
+
+    pub fn set_request_timeout(mut self, timeout: Duration) -> Self {
         self.request_timeout = timeout;
         self
     }
 
-    pub fn set_credentials(&mut self, username: &str, password: &str) -> &mut Self {
+    pub fn set_credentials(mut self, username: &str, password: &str) -> Self {
         self.options.set_credentials(username, password);
         self
     }
 
     pub fn build(&mut self) -> MqttClient {
         self.options.set_keep_alive(Duration::from_secs(10));
-        let (client, eventloop) = AsyncClient::new(self.options.clone(), 10);
+        let (client, eventloop) = AsyncClient::new(self.options.clone(), self.max_cap);
 
         let subscriptions = SubscribeHandlerMap::new();
-        let loop_sub = subscriptions.clone();
+        let cli = MqttClient {
-        tokio::spawn(async move {
-            MqttClient::handle_connection_loop(eventloop, loop_sub).await;
-        });
-
-        MqttClient {
             id: self.id.clone(),
             request_timeout: self.request_timeout,
             client,
             request_id: Arc::new(AtomicU64::new(0)),
             subscriptions,
-        }
+        };
+        cli.run_async(eventloop);
+        cli
     }
 }
 
@@ -104,6 +124,7 @@ impl MqttClientOptions {
 #[derive(Clone)]
 pub struct MqttClient {
     id: String,
+    /// 全局的请求超时时间
     request_timeout: Duration,
     client: AsyncClient,
     request_id: Arc<AtomicU64>,
@@ -154,11 +175,11 @@ impl SubscribeHandlerMap {
     }
 
     fn get_handlers(&self, topic: &str) -> Option<Vec<Arc<dyn MessageHandler>>> {
-        if let Some(topic_handlers) = self.sub_handlers.lock().unwrap().get(topic) {
+        self.sub_handlers
-            Some(topic_handlers.values())
+            .lock()
-        } else {
+            .unwrap()
-            None
+            .get(topic)
-        }
+            .map(|handlers| handlers.values())
     }
 
     #[allow(dead_code)]
@@ -261,7 +282,7 @@ where
 }
 
 impl MqttClient {
-    pub async fn close(&self) -> Result<(), MqttClientError> {
+    pub async fn clear(&self) -> Result<(), MqttClientError> {
         self.client.disconnect().await?;
         // 清空订阅处理器
         self.subscriptions.clear();
@@ -319,6 +340,7 @@ impl MqttClient {
     }
 
     /// 发送请求并等待响应
+    /// TODO: 需要测试中请求时的并发情况（多个请求同时等待，以及在请求时订阅和发布是否受影响）
     pub async fn request(
         &self,
         topic: &str,
@@ -335,7 +357,7 @@ impl MqttClient {
             self.register_topic_handler(&response_topic, response_future.clone());
         // 发布请求
         let property = PublishProperties {
-            response_topic: Some(response_topic.clone().into()),
+            response_topic: Some(response_topic.clone()),
             ..Default::default()
         };
         self.client
@@ -349,14 +371,48 @@ impl MqttClient {
         Ok(resp)
     }
 
-    async fn handle_connection_loop(mut eventloop: EventLoop, subscriptions: SubscribeHandlerMap) {
+    /// 发送请求并等待响应，指定响应超时时间
+    /// 响应超时时间为0时表示永不超时
+    pub async fn request_with_timeout(
+        &self,
+        topic: &str,
+        qos: QoS,
+        payload: Vec<u8>,
+        timeout: Duration,
+    ) -> Result<MqttResponse, MqttClientError> {
+        let response_topic = format!("{}/{}/resp/{}", self.id, topic, self.next_request_id());
+        self.subscribe(&response_topic, QoS::ExactlyOnce).await?;
+        let response_future = MqttResponseFuture::new(&response_topic, timeout);
+        let response_handler =
+            self.register_topic_handler(&response_topic, response_future.clone());
+        let property = PublishProperties {
+            response_topic: Some(response_topic.clone()),
+            ..Default::default()
+        };
+        self.client
+            .publish_with_properties(topic, qos, false, payload, property)
+            .await?;
+        let resp = response_future.await;
+        response_handler.unregister();
+        self.unsubscribe(&response_topic).await?;
+        Ok(resp)
+    }
+
+    fn run_async(&self, eventloop: EventLoop) {
+        let cli = self.clone();
+        tokio::spawn(async move {
+            cli.run(eventloop).await;
+        });
+    }
+
+    async fn run(&self, mut eventloop: EventLoop) {
         while let Ok(notification) = eventloop.poll().await {
             match notification {
                 Event::Incoming(Packet::Publish(publish)) => {
-                    debug!("Received message: {:?}", publish);
+                    trace!("Received message: {:?}", publish);
                     let topic: String = String::from_utf8_lossy(&publish.topic).to_string();
 
-                    if let Some(topic_handlers) = subscriptions.get_handlers(&topic) {
+                    if let Some(topic_handlers) = self.subscriptions.get_handlers(&topic) {
                         for handler in topic_handlers {
                             let handler = handler.clone();
                             let p = publish.clone();
@@ -375,7 +431,7 @@ impl MqttClient {
                     break;
                 }
                 _ => {
-                    debug!("Unhandled event: {:?}", notification);
+                    trace!("Unhandled event: {:?}", notification);
                 }
             }
         }
@@ -396,6 +452,12 @@ pub struct MqttResponse {
     response: Arc<Mutex<Bytes>>,
 }
 
+impl Default for MqttResponse {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl MqttResponse {
     pub fn new() -> Self {
         Self {
@@ -460,12 +522,15 @@ impl MqttResponseFuture {
 
     /// 启动超时监控任务逻辑
     fn start_timeout_monitor(&self, rx: oneshot::Receiver<()>) {
+        if self.timeout.as_millis() == 0 {
+            return;
+        }
         let response = self.response.clone();
         let response_topic = self.response_topic.clone();
-        let duration = self.timeout.clone();
+        let duration = self.timeout;
         let waker = self.waker.clone();
         tokio::spawn(async move {
-            if let Err(_) = timeout(duration, rx).await {
+            if (timeout(duration, rx).await).is_err() {
                 error!("Mqtt response timeout: {:?}", response_topic);
                 response.set_timeout();
                 if let Some(waker) = waker.lock().unwrap().take() {
@@ -500,18 +565,24 @@ impl std::future::Future for MqttResponseFuture {
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Self::Output> {
         if self.response.is_waiting() {
-            debug!("Response future poll waiting...");
+            debug!(
+                "topic={} Response future poll waiting...",
+                self.response_topic
+            );
             self.waker.lock().unwrap().replace(cx.waker().clone());
             std::task::Poll::Pending
         } else {
-            debug!("Response future poll ready: {:?}", self.response.get());
+            debug!(
+                "topic={} Response future poll ready: {:?}",
+                self.response_topic, self.response
+            );
             std::task::Poll::Ready(self.response.clone())
         }
     }
 }
 
 pub fn get_publish_response_topic(publish: Option<PublishProperties>) -> Option<String> {
-    publish.map(|p| p.response_topic.clone()).flatten()
+    publish.and_then(|p| p.response_topic.clone())
 }
 
 #[cfg(test)]
@@ -519,14 +590,20 @@ mod tests {
     use super::*;
     use rtss_log::tracing::{info, Level};
 
-    use tokio::time::{sleep, Duration};
+    use tokio::{
+        sync::broadcast,
+        time::{sleep, Duration},
+    };
+
+    fn create_mqtt_options() -> MqttClientOptions {
+        MqttClientOptions::new("rtss_test1", "tcp://localhost:1883")
+            .set_credentials("rtsa", "Joylink@0503")
+    }
 
     #[tokio::test]
     async fn test_subscribe_and_publish() {
-        rtss_log::Logging::default().with_level(Level::DEBUG).init();
+        rtss_log::Logging::default().with_level(Level::TRACE).init();
-        let client = MqttClientOptions::new("rtss_test1", "tcp://localhost:1883")
+        let client = create_mqtt_options().build();
-            .set_credentials("rtss_simulation", "Joylink@0503")
-            .build();
 
         client
             .subscribe("test/topic", QoS::AtMostOnce)
@@ -539,7 +616,7 @@ mod tests {
                 String::from_utf8_lossy(&publish.payload)
             );
         });
-        let h2 = client.register_topic_handler("test/topic", |publish: Publish| {
+        let _ = client.register_topic_handler("test/topic", |publish: Publish| {
             info!(
                 "Handler 2 received: topic={}, payload={:?}",
                 String::from_utf8_lossy(&publish.topic),
@@ -561,52 +638,65 @@ mod tests {
         handler1.unregister();
         assert_eq!(client.topic_handler_count("test/topic"), 1);
 
-        h2.unregister();
+        sleep(Duration::from_millis(200)).await;
-        assert_eq!(client.topic_handler_count("test/topic"), 0);
+        // Test clear
+        client.clear().await.unwrap();
 
-        // Test unsubscribe
+        assert_eq!(client.topic_handler_count("test/topic"), 0);
-        client.close().await.unwrap();
     }
 
     #[tokio::test]
     async fn test_request() {
         rtss_log::Logging::default().with_level(Level::DEBUG).init();
-        let client = MqttClientOptions::new("rtss_test1", "tcp://localhost:1883")
+        init_global_mqtt_client(create_mqtt_options())
-            .set_credentials("rtss_simulation", "Joylink@0503")
+            .await
-            .build();
+            .unwrap();
-        set_global_mqtt_client(client.clone()).await.unwrap();
 
-        if let Some(c) = get_global_mqtt_client().await {
+        let c = get_global_mqtt_client().await;
-            c.subscribe("test/request", QoS::AtMostOnce).await.unwrap();
+        c.subscribe("test/request", QoS::AtMostOnce).await.unwrap();
-            let handler = |p: Publish| {
+        let handler = |p: Publish| {
-                info!(
+            info!(
-                    "Request handler received: topic={}, payload={:?}",
+                "Request handler received: topic={}, payload={:?}",
-                    String::from_utf8_lossy(&p.topic),
+                String::from_utf8_lossy(&p.topic),
-                    String::from_utf8_lossy(&p.payload)
+                String::from_utf8_lossy(&p.payload)
-                );
+            );
-                let response = Bytes::from("Hello, response!");
+            let response = Bytes::from("Hello, response!");
-                let resp_topic = get_publish_response_topic(p.properties.clone());
+            let resp_topic = get_publish_response_topic(p.properties.clone());
-                if let Some(r_topic) = resp_topic {
+            if let Some(r_topic) = resp_topic {
-                    tokio::spawn(async move {
+                tokio::spawn(async move {
-                        if let Some(c) = get_global_mqtt_client().await {
+                    // 此处需要使用全局MqttClient实例，否则会出现死锁
-                            c.publish(&r_topic, QoS::AtMostOnce, response.to_vec())
+                    let c = get_global_mqtt_client().await;
-                                .await
+                    c.publish(&r_topic, QoS::AtMostOnce, response.to_vec())
-                                .unwrap();
+                        .await
-                        }
+                        .unwrap();
-                    });
+                });
-                }
+            }
-            };
+        };
-            let _ = c.register_topic_handler("test/request", handler);
+        let _ = c.register_topic_handler("test/request", handler);
-        }
 
-        if let Some(c) = get_global_mqtt_client().await {
+        let response = c
-            let response = c
+            .request("test/request", QoS::AtMostOnce, b"Hello, request!".to_vec())
-                .request("test/request", QoS::AtMostOnce, b"Hello, request!".to_vec())
+            .await
-                .await
+            .unwrap();
-                .unwrap();
+        info!("Request response: {:?}", response);
-            info!("Request response: {:?}", response);
+    }
-        }
 
-        client.close().await.unwrap();
+    #[tokio::test]
+    async fn test_async_broadcast() {
+        let (tx, mut rx1) = broadcast::channel(16);
+        let mut rx2 = tx.subscribe();
+
+        tokio::spawn(async move {
+            assert_eq!(rx1.recv().await.unwrap(), 10);
+            assert_eq!(rx1.recv().await.unwrap(), 20);
+        });
+
+        tokio::spawn(async move {
+            assert_eq!(rx2.recv().await.unwrap(), 10);
+            assert_eq!(rx2.recv().await.unwrap(), 20);
+        });
+
+        tx.send(10).unwrap();
+        tx.send(20).unwrap();
     }
 }

manager/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "manager"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+rtss_log = { path = "../crates/rtss_log" }
+rtss_api = { path = "crates/rtss_api" }
+rtss_db = { path = "../crates/rtss_db" }
+serde = { workspace = true }
+config = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
+enum_dispatch = { workspace = true }
+anyhow = { workspace = true }

manager/crates/rtss_api/Cargo.toml

@@ -19,9 +19,6 @@ base64 = "0.22.1"
 sysinfo = "0.31.3"
 reqwest = { version = "0.12.7", default-features = false, features = ["rustls-tls", "json"] }
 
-bevy_ecs = { workspace = true }
+rtss_log = { path = "../../../crates/rtss_log" }
-rtss_log = { path = "../rtss_log" }
+rtss_db = { path = "../../../crates/rtss_db" }
-rtss_sim_manage = { path = "../rtss_sim_manage" }
+rtss_dto = { path = "../../../crates/rtss_dto" }
-rtss_trackside = { path = "../rtss_trackside" }
-rtss_db = { path = "../rtss_db" }
-rtss_dto = { path = "../rtss_dto" }

manager/crates/rtss_api/src/apis/mod.rs

@@ -3,9 +3,7 @@ use draft_data::{DraftDataMutation, DraftDataQuery};
 use feature::{FeatureMutation, FeatureQuery};
 use release_data::{ReleaseDataMutation, ReleaseDataQuery};
 
-mod simulation_definition;
 mod sys_info;
-use simulation_definition::*;
 use user::{UserMutation, UserQuery};
 
 mod data_options_def;
@@ -13,7 +11,6 @@ mod draft_data;
 mod feature;
 mod feature_config_def;
 mod release_data;
-mod simulation;
 mod user;
 
 #[derive(Default, MergedObject)]

manager/src/commands/cmd.rs

@@ -1,7 +1,9 @@
 use clap::Parser;
 use enum_dispatch::enum_dispatch;
 
-use crate::{app_config, db::DbSubCommand, CmdExecutor};
+use crate::{app_config, CmdExecutor};
+
+use super::DbSubCommand;
 
 #[derive(Parser, Debug)]
 #[command(name = "rtss-sim", version, author, about, long_about = None)]

manager/src/commands/mod.rs

@@ -1,4 +1,3 @@
-mod app_config;
 mod cmd;
 mod db;
 

manager/src/lib.rs

@@ -0,0 +1,4 @@
+mod app_config;
+mod commands;
+
+pub use commands::*;

manager/src/main.rs

@@ -1,5 +1,5 @@
 use clap::Parser;
-use rtss_simulation::{Cmd, CmdExecutor};
+use manager::{Cmd, CmdExecutor};
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {

rtss-proto-msg

@@ -1 +1 @@
-Subproject commit 236252fc0fa258e6beaae5b5ac0a7e28ebbcb04b
+Subproject commit 1f53057b3f87790ef27c91399a5bb7e890f05549

simulation/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "simulation"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+bevy_core = { workspace = true }
+bevy_ecs = { workspace = true }
+bevy_app = { workspace = true }
+bevy_time = { workspace = true }
+rayon = { workspace = true }
+thiserror = { workspace = true }
+lazy_static = { workspace = true }
+tokio = { workspace = true }
+serde = { workspace = true }
+config = { workspace = true }
+clap = { workspace = true, features = ["derive"] }
+enum_dispatch = { workspace = true }
+anyhow = { workspace = true }
+
+rtss_log = { path = "../crates/rtss_log" }
+rtss_dto = { path = "../crates/rtss_dto" }
+rtss_db = { path = "../crates/rtss_db" }
+rtss_mqtt = { path = "../crates/rtss_mqtt" }

simulation/conf/default.toml

@@ -0,0 +1,9 @@
+[database]
+
+[log]
+level = "debug"
+
+[mqtt]
+url = "http://localhost:8080"
+username = "admin"
+password = "admin"

simulation/conf/dev.toml

@@ -0,0 +1,7 @@
+[database]
+url = "postgresql://joylink:Joylink@0503@localhost:5432/joylink"
+
+[mqtt]
+url = "tcp://localhost:1883"
+username = "rtsa"
+password = "Joylink@0503"

simulation/conf/local_test.toml

@@ -0,0 +1,10 @@
+[database]
+url = "postgresql://joylink:Joylink@0503@10.11.11.2:5432/joylink"
+
+[log]
+level = "debug"
+
+[mqtt]
+url = "tcp://192.168.3.233:1883"
+username = "rtsa"
+password = "Joylink@0503"

simulation/crates/rtss_sim_manage/Cargo.toml

@@ -11,6 +11,6 @@ bevy_time = {workspace = true}
 rayon = {workspace = true}
 thiserror = {workspace = true}
 
-rtss_log = { path = "../rtss_log" }
+rtss_log = { path = "../../crates/rtss_log" }
 rtss_common = { path = "../rtss_common" }
 rtss_trackside = { path = "../rtss_trackside" }

simulation/crates/rtss_sim_manage/src/main.rs

@@ -0,0 +1,4 @@
+
+fn main() {
+    println!("Hello, world!");
+}

simulation/crates/rtss_trackside/Cargo.toml

@@ -9,5 +9,5 @@ bevy_ecs = {workspace = true}
 bevy_app = {workspace = true}
 bevy_time = {workspace = true}
 
-rtss_log = { path = "../rtss_log" }
+rtss_log = { path = "../../crates/rtss_log" }
 rtss_common = { path = "../rtss_common" }

simulation/src/app_config.rs

@@ -0,0 +1,87 @@
+use std::env;
+
+use config::{Config, ConfigError, Environment, File};
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize)]
+#[allow(unused)]
+pub struct Database {
+    pub url: String,
+}
+
+#[derive(Debug, Deserialize)]
+#[allow(unused)]
+pub struct Log {
+    level: String,
+}
+
+impl From<Log> for rtss_log::Logging {
+    fn from(log: Log) -> Self {
+        rtss_log::Logging {
+            level: log.level.parse().unwrap(),
+            ..Default::default()
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[allow(unused)]
+pub struct Mqtt {
+    pub url: String,
+    pub username: String,
+    pub password: String,
+}
+
+#[derive(Debug, Deserialize)]
+#[allow(unused)]
+pub struct AppConfig {
+    pub log: Log,
+    pub database: Database,
+    pub mqtt: Mqtt,
+}
+
+impl AppConfig {
+    pub fn new(dir: &str) -> Result<Self, ConfigError> {
+        let run_mode = env::var("RUN_MODE")
+            .unwrap_or_else(|_| "dev".into())
+            .trim()
+            .to_string();
+        println!("RUN_MODE: {}", run_mode);
+
+        // log 当前目录
+        // println!("Current dir: {:?}", std::env::current_dir().unwrap());
+
+        let s = Config::builder()
+            // Start off by merging in the "default" configuration file
+            .add_source(File::with_name(&format!("{dir}/default")))
+            // Add in a local configuration file
+            // This file shouldn't be checked in to git
+            .add_source(File::with_name(&format!("{dir}/local")).required(false))
+            // Add in the current environment file
+            // Default to 'dev' env
+            // Note that this file is _optional_
+            .add_source(File::with_name(&format!("{dir}/{run_mode}")).required(true))
+            // Add in settings from the environment (with a prefix of RTSS_SIM)
+            // Eg.. `APP_DEBUG=1 ./target/app` would set the `debug` key
+            .add_source(Environment::with_prefix("RTSS_SIM").separator("_"))
+            // You may also programmatically change settings
+            // .set_override("database.url", "postgres://")?
+            // build the configuration
+            .build()?;
+
+        // You can deserialize (and thus freeze) the entire configuration as
+        s.try_deserialize()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_app_config() {
+        std::env::set_var("RUN_MODE", "local_test");
+        let config = AppConfig::new("conf").unwrap();
+        println!("{:?}", config);
+    }
+}

simulation/src/commands/cmd.rs

@@ -0,0 +1,48 @@
+use clap::Parser;
+use enum_dispatch::enum_dispatch;
+
+use crate::app_config;
+
+use super::{CmdExecutor, DbSubCommand};
+
+#[derive(Parser, Debug)]
+#[command(name = "rtss-sim", version, author, about, long_about = None)]
+pub struct Cmd {
+    #[command(subcommand)]
+    pub cmd: SubCommand,
+}
+
+#[derive(Parser, Debug)]
+#[enum_dispatch(CmdExecutor)]
+pub enum SubCommand {
+    #[command(name = "serve")]
+    Serve(ServerOpts),
+    #[command(name = "db", subcommand, about = "Database operations")]
+    Db(DbSubCommand),
+}
+
+#[derive(Parser, Debug)]
+pub struct ServerOpts {
+    #[clap(short, long, required = false, default_value = "conf")]
+    config_path: String,
+}
+
+impl CmdExecutor for ServerOpts {
+    async fn execute(&self) -> anyhow::Result<()> {
+        let app_config =
+            app_config::AppConfig::new(&self.config_path).expect("Failed to load app config");
+        let log: rtss_log::Logging = app_config.log.into();
+        log.init();
+        // 数据库访问器初始化
+        rtss_db::init_default_db_accessor(&app_config.database.url).await;
+        // mqtt客户端初始化
+        let cli_id = rtss_db::get_default_db_accessor()
+            .get_next_mqtt_client_id()
+            .await?;
+        let mqtt_cli_options =
+            rtss_mqtt::MqttClientOptions::new(&format!("rtsa{}", cli_id), &app_config.mqtt.url)
+                .set_credentials(&app_config.mqtt.username, &app_config.mqtt.password);
+        rtss_mqtt::init_global_mqtt_client(mqtt_cli_options).await?;
+        Ok(())
+    }
+}

simulation/src/commands/db.rs

@@ -0,0 +1,29 @@
+use clap::Parser;
+use enum_dispatch::enum_dispatch;
+
+use crate::app_config;
+
+use super::CmdExecutor;
+
+#[derive(Parser, Debug)]
+#[enum_dispatch(CmdExecutor)]
+pub enum DbSubCommand {
+    #[command(name = "migrate", about = "Migrate database")]
+    Migrate(MigrateOpts),
+}
+
+#[derive(Parser, Debug)]
+pub struct MigrateOpts {
+    #[clap(long, required = false, default_value = "conf")]
+    config_path: String,
+    #[clap(long, required = false, default_value = "migrations")]
+    file_path: String,
+}
+
+impl CmdExecutor for MigrateOpts {
+    async fn execute(&self) -> anyhow::Result<()> {
+        let app_config =
+            app_config::AppConfig::new(&self.config_path).expect("Failed to load app config");
+        rtss_db::run_migrations(&app_config.database.url).await
+    }
+}

simulation/src/commands/mod.rs

@@ -0,0 +1,12 @@
+mod cmd;
+mod db;
+
+pub use cmd::*;
+use db::*;
+use enum_dispatch::enum_dispatch;
+
+#[allow(async_fn_in_trait)]
+#[enum_dispatch]
+pub trait CmdExecutor {
+    async fn execute(&self) -> anyhow::Result<()>;
+}

simulation/src/components/mod.rs

@@ -0,0 +1,93 @@
+//! 顶级通用组件
+
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+};
+
+use bevy_ecs::{component::Component, entity::Entity, system::Resource};
+use tokio::sync::broadcast;
+
+#[derive(Resource, Clone, Debug)]
+pub struct SimulationInfo {
+    pub id: String,
+    pub feature_id: String,
+    pub is_paused: bool,
+    pub speed: f32,
+}
+
+impl SimulationInfo {
+    pub fn new(id: &str, feature_id: &str) -> Self {
+        SimulationInfo {
+            id: id.to_string(),
+            feature_id: feature_id.to_string(),
+            is_paused: false,
+            speed: 1.0,
+        }
+    }
+}
+
+// 设备编号组件
+#[derive(Component, Debug, Clone, PartialEq, Eq)]
+pub struct Uid(pub String);
+impl Default for Uid {
+    fn default() -> Self {
+        Uid("".to_string())
+    }
+}
+
+/// 仿真uid与实体映射资源
+#[derive(Resource, Clone, Debug, Default)]
+pub struct SimulationUidEntityMapResource(pub Arc<Mutex<HashMap<String, Entity>>>);
+
+impl SimulationUidEntityMapResource {
+    pub fn get_entity(&self, uid: &str) -> Option<Entity> {
+        self.0.lock().unwrap().get(uid).cloned()
+    }
+
+    pub fn insert_entity(&self, uid: String, entity: Entity) {
+        self.0.lock().unwrap().insert(uid, entity);
+    }
+}
+
+#[derive(Resource, Debug)]
+pub struct TxResource {
+    tx: broadcast::Sender<rtss_dto::simulation::Operation>,
+    rx: Option<broadcast::Receiver<rtss_dto::simulation::Operation>>,
+}
+
+impl TxResource {
+    pub fn new(capacity: usize) -> Self {
+        let (tx, rx) = broadcast::channel(capacity);
+        TxResource { tx, rx: Some(rx) }
+    }
+
+    pub fn get_tx(&self) -> broadcast::Sender<rtss_dto::simulation::Operation> {
+        self.tx.clone()
+    }
+
+    pub fn subscribe(&mut self) -> broadcast::Receiver<rtss_dto::simulation::Operation> {
+        let rx = self.tx.subscribe();
+        if self.rx.is_some() {
+            std::mem::take(&mut self.rx);
+        }
+        rx
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use bevy_ecs::world;
+
+    use super::*;
+
+    #[test]
+    fn it_works() {
+        let simulation_resource = SimulationUidEntityMapResource::default();
+        let mut world = world::World::default();
+        let uid = Uid("1".to_string());
+        let entity = world.spawn(uid.clone()).id();
+        simulation_resource.insert_entity(uid.clone().0, entity);
+        assert_eq!(simulation_resource.get_entity(&uid.0), Some(entity));
+    }
+}

simulation/src/error.rs

@@ -0,0 +1,9 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum SimulationError {
+    #[error("未知的仿真错误")]
+    Unknown,
+    #[error("创建错误: {0}")]
+    CreateError(String),
+}

simulation/src/lib.rs

@@ -0,0 +1,6 @@
+pub mod app_config;
+pub mod commands;
+pub mod components;
+pub mod error;
+pub mod manage;
+pub mod modules;

simulation/src/main.rs

@@ -0,0 +1,9 @@
+use clap::Parser;
+use simulation::commands::{Cmd, CmdExecutor};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let cmd = Cmd::parse();
+    cmd.cmd.execute().await?;
+    Ok(())
+}

simulation/src/manage/data_loading_plugin.rs

@@ -0,0 +1,18 @@
+use bevy_app::{Plugin, Startup};
+use bevy_ecs::system::Res;
+use rtss_log::tracing::debug;
+
+use crate::components::SimulationInfo;
+
+#[derive(Default)]
+pub struct DataLoadingPlugin;
+
+impl Plugin for DataLoadingPlugin {
+    fn build(&self, app: &mut bevy_app::App) {
+        app.add_systems(Startup, loading);
+    }
+}
+
+fn loading(info: Res<SimulationInfo>) {
+    debug!("Loading data: {:?}", info);
+}

simulation/src/manage/mod.rs

@@ -0,0 +1,5 @@
+mod data_loading_plugin;
+mod simulation;
+mod simulation_control_plugin;
+
+pub use simulation::{Simulation, SimulationOptions};

simulation/src/manage/simulation.rs

@@ -0,0 +1,157 @@
+use std::{
+    sync::{Arc, Mutex},
+    time::Duration,
+};
+
+use bevy_app::{prelude::*, MainSchedulePlugin};
+use bevy_ecs::{
+    event::{event_update_system, EventRegistry},
+    prelude::*,
+    reflect::AppTypeRegistry,
+    schedule::ScheduleLabel,
+    system::ResMut,
+};
+use bevy_time::{Fixed, Time, TimePlugin, Virtual};
+use rtss_log::tracing::{debug, error, info};
+use tokio::sync::broadcast;
+
+use crate::{
+    components::{SimulationInfo, SimulationUidEntityMapResource, TxResource},
+    error::SimulationError,
+};
+
+use super::{
+    data_loading_plugin::DataLoadingPlugin, simulation_control_plugin::SimulationControlPlugin,
+};
+
+/// 仿真构造配置项
+pub struct SimulationOptions {
+    /// 仿真ID
+    pub(crate) id: String,
+    /// 仿真功能特性类型
+    pub(crate) feature_id: String,
+    /// 仿真主逻辑循环间隔，详细请查看 [`Time<Fixed>`](bevy_time::fixed::Fixed)
+    pub(crate) loop_duration: Duration,
+}
+
+impl SimulationOptions {
+    pub fn new(id: &str, feature_id: &str) -> Self {
+        SimulationOptions {
+            id: id.to_string(),
+            feature_id: feature_id.to_string(),
+            loop_duration: Duration::from_millis(20),
+        }
+    }
+
+    pub fn loop_duration(mut self, loop_duration: Duration) -> Self {
+        self.loop_duration = loop_duration;
+        self
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct Simulation {
+    app: Arc<Mutex<SubApp>>,
+    tx: broadcast::Sender<rtss_dto::simulation::Operation>,
+}
+
+impl Simulation {
+    pub fn new(options: SimulationOptions) -> Result<Self, SimulationError> {
+        let mut sub = SubApp::new();
+        sub.init_resource::<AppTypeRegistry>()
+            .init_resource::<EventRegistry>();
+        sub.update_schedule = Some(Main.intern());
+        sub.add_plugins(MainSchedulePlugin).add_plugins(TimePlugin);
+        sub.add_systems(
+            First,
+            event_update_system
+                .in_set(bevy_ecs::event::EventUpdates)
+                .run_if(bevy_ecs::event::event_update_condition),
+        );
+        sub.add_systems(
+            PreStartup,
+            move |mut tv: ResMut<Time<Virtual>>, mut tf: ResMut<Time<Fixed>>| {
+                debug!("PreStartup set time");
+                tv.set_max_delta(options.loop_duration);
+                tf.set_timestep(options.loop_duration);
+            },
+        );
+        // 添加全局共享资源
+        let tx_resource = TxResource::new(100);
+        let tx = tx_resource.get_tx();
+        sub.insert_resource(SimulationInfo::new(&options.id, &options.feature_id))
+            .insert_resource(SimulationUidEntityMapResource::default())
+            .insert_resource(tx_resource);
+
+        // 添加通用插件
+        sub.add_plugins(SimulationControlPlugin)
+            .add_plugins(DataLoadingPlugin);
+
+        sub.add_systems(FixedUpdate, hello_world_system);
+
+        Ok(Simulation {
+            app: Arc::new(Mutex::new(sub)),
+            tx,
+        })
+    }
+
+    pub fn send_operation(&self, op: rtss_dto::simulation::Operation) {
+        if let Err(e) = self.tx.send(op) {
+            error!("send operation error: {}", e);
+        }
+    }
+
+    pub fn update(&self) {
+        let mut app = self.app.lock().unwrap();
+        app.update();
+    }
+}
+
+fn hello_world_system(tv: Res<Time<Virtual>>) {
+    info!("hello world: {}", tv.relative_speed());
+}
+
+#[cfg(test)]
+mod tests {
+    use std::thread::{self, sleep};
+
+    use super::*;
+
+    use rtss_dto::simulation::OperationType;
+    use rtss_log::tracing::Level;
+
+    #[test]
+    fn test_new_simulation() {
+        rtss_log::Logging::default().with_level(Level::DEBUG).init();
+        let simulation1 = Simulation::new(SimulationOptions::new("1", "1")).unwrap();
+        assert!(simulation1.app.lock().unwrap().update_schedule.is_some());
+        simulation1.send_operation(rtss_dto::simulation::Operation {
+            otype: rtss_dto::simulation::OperationType::SetSpeed as i32,
+            param: Some(rtss_dto::simulation::operation::Param::SetSpeedParam(
+                rtss_dto::simulation::SetSpeedParam { speed: 2.0 },
+            )),
+        });
+        let clone2 = simulation1.clone();
+        thread::spawn(move || {
+            sleep(Duration::from_millis(100));
+            info!("send pause");
+            clone2.send_operation(rtss_dto::simulation::Operation {
+                otype: OperationType::Pause as i32,
+                param: None,
+            });
+            sleep(Duration::from_millis(200));
+            info!("send unpause");
+            clone2.send_operation(rtss_dto::simulation::Operation {
+                otype: OperationType::Unpause as i32,
+                param: None,
+            });
+        });
+        // let simulation2 = Simulation::new(SimulationOptions::new("2", "1")).unwrap();
+
+        for _ in 0..30 {
+            sleep(Duration::from_millis(20));
+            simulation1.update();
+            // simulation2.update();
+        }
+    }
+}

simulation/src/manage/simulation_control_plugin.rs

@@ -0,0 +1,61 @@
+use bevy_app::prelude::*;
+use bevy_ecs::prelude::*;
+use bevy_time::prelude::*;
+use rtss_dto::simulation::{operation, OperationType, SetSpeedParam};
+use rtss_log::tracing::debug;
+use tokio::sync::broadcast;
+
+use crate::components::{SimulationInfo, TxResource};
+
+#[derive(Default)]
+pub struct SimulationControlPlugin;
+
+impl Plugin for SimulationControlPlugin {
+    fn build(&self, app: &mut bevy_app::App) {
+        app.add_systems(Startup, init_rx_resource)
+            .add_systems(Update, handle_control_operation);
+    }
+}
+
+fn init_rx_resource(mut commands: Commands, mut tx: ResMut<TxResource>) {
+    debug!("init_rx_resource");
+    commands.insert_resource(RxResource::new(tx.subscribe()));
+}
+
+#[derive(Resource, Debug)]
+struct RxResource(pub broadcast::Receiver<rtss_dto::simulation::Operation>);
+
+impl RxResource {
+    fn new(rx: broadcast::Receiver<rtss_dto::simulation::Operation>) -> Self {
+        RxResource(rx)
+    }
+}
+
+fn handle_control_operation(
+    mut rx: ResMut<RxResource>,
+    mut time: ResMut<Time<Virtual>>,
+    info: Res<SimulationInfo>,
+) {
+    if let Ok(op) = rx.0.try_recv() {
+        match OperationType::try_from(op.otype) {
+            Ok(OperationType::Pause) => {
+                debug!("Pausing simulation, id={}", info.id);
+                time.pause();
+            }
+            Ok(OperationType::Unpause) => {
+                debug!("Unpausing simulation, id={}", info.id);
+                time.unpause();
+            }
+            Ok(OperationType::SetSpeed) => {
+                if let Some(operation::Param::SetSpeedParam(SetSpeedParam { speed })) = op.param {
+                    debug!("Update simulation id={} speed to {}", info.id, speed);
+                    time.set_relative_speed(speed);
+                }
+            }
+            Ok(OperationType::Destroy) => {
+                debug!("Exiting simulation, id={}", info.id);
+            }
+            _ => {}
+        }
+    }
+}

simulation/src/modules/ci/mod.rs

@@ -0,0 +1,3 @@
+mod plugin;
+
+pub use plugin::CiPlugin;

simulation/src/modules/ci/plugin.rs

@@ -0,0 +1,10 @@
+use bevy_app::prelude::*;
+
+#[derive(Default)]
+pub struct CiPlugin;
+
+impl Plugin for CiPlugin {
+    fn build(&self, _app: &mut bevy_app::App) {
+        todo!()
+    }
+}

simulation/src/modules/mod.rs

@@ -0,0 +1,2 @@
+pub mod ci;
+pub mod trackside;

simulation/src/modules/trackside/components/bundle.rs

@@ -0,0 +1,21 @@
+use bevy_ecs::bundle::Bundle;
+
+use crate::components::Uid;
+
+use super::{PsdState, TurnoutState, TwoNormalPositionsTransform};
+
+// 道岔设备组件包
+#[derive(Bundle, Default)]
+pub struct TurnoutBundle {
+    pub uid: Uid,
+    pub turnout_state: TurnoutState,
+    pub two_normal_positions_conversion: TwoNormalPositionsTransform,
+}
+
+/// 屏蔽门设备组件包
+#[derive(Bundle, Default)]
+pub struct PsdBundle {
+    pub uid: Uid,
+    pub psd_state: PsdState,
+    pub two_normal_positions_conversion: TwoNormalPositionsTransform,
+}

simulation/src/modules/trackside/components/equipment.rs

@@ -0,0 +1,40 @@
+use bevy_ecs::component::Component;
+
+/// 两常态位置转换组件，用于像道岔位置，屏蔽门位置等
+#[derive(Component, Debug, Clone, PartialEq, Default)]
+pub struct TwoNormalPositionsTransform {
+    // 当前实际位置,百分比值,0-100
+    pub position: f32,
+    // 当前转换速度
+    pub velocity: f32,
+}
+
+/// 道岔设备状态组件
+#[derive(Component, Debug, Clone, PartialEq, Eq, Default)]
+pub struct TurnoutState {
+    // 定位表示继电器状态
+    pub dbj: bool,
+    // 反位表示继电器状态
+    pub fbj: bool,
+    // 是否定位
+    pub dw: bool,
+    // 是否反位
+    pub fw: bool,
+    // 定操继电器状态
+    pub dcj: bool,
+    // 反操继电器状态
+    pub fcj: bool,
+}
+
+/// 屏蔽门设备状态组件
+#[derive(Component, Debug, Clone, PartialEq, Eq, Default)]
+pub struct PsdState {
+    // 门关继电器状态
+    pub mgj: bool,
+    // 关门继电器状态
+    pub gmj: bool,
+    // 开门继电器状态
+    pub kmj: bool,
+    // 门旁路继电器状态(互锁解除)
+    pub mplj: bool,
+}

simulation/src/modules/trackside/components/mod.rs

@@ -0,0 +1,4 @@
+mod bundle;
+mod equipment;
+pub use bundle::*;
+pub use equipment::*;

simulation/src/modules/trackside/events/equipment.rs

@@ -0,0 +1,10 @@
+use bevy_ecs::event::Event;
+
+#[allow(dead_code)]
+#[derive(Event, Debug, Clone, Copy, Eq, PartialEq)]
+pub enum TurnoutControlEvent {
+    // 道岔定操
+    DC,
+    // 道岔反操
+    FC,
+}

simulation/src/modules/trackside/events/mod.rs

@@ -0,0 +1,2 @@
+mod equipment;
+pub use equipment::*;

simulation/src/modules/trackside/mod.rs

@@ -0,0 +1,7 @@
+mod components;
+mod events;
+mod plugin;
+mod resources;
+mod systems;
+
+pub use plugin::TrackSideEquipmentPlugin;

simulation/src/modules/trackside/plugin.rs

@@ -0,0 +1,26 @@
+use bevy_app::{FixedUpdate, Plugin, Startup};
+use bevy_ecs::schedule::IntoSystemConfigs;
+
+use super::{
+    events::TurnoutControlEvent,
+    resources::SimulationConfig,
+    systems::{
+        handle_turnout_control, loading, turnout_state_update, two_normal_position_transform,
+    },
+};
+
+#[derive(Default)]
+pub struct TrackSideEquipmentPlugin;
+
+impl Plugin for TrackSideEquipmentPlugin {
+    fn build(&self, app: &mut bevy_app::App) {
+        app.insert_resource(SimulationConfig::default())
+            .add_event::<TurnoutControlEvent>()
+            .add_systems(Startup, loading)
+            .add_systems(
+                FixedUpdate,
+                (two_normal_position_transform, turnout_state_update).chain(),
+            )
+            .observe(handle_turnout_control);
+    }
+}

simulation/src/modules/trackside/resources/mod.rs

@@ -0,0 +1,15 @@
+use bevy_ecs::system::Resource;
+
+#[derive(Resource, Debug)]
+pub struct SimulationConfig {
+    // 道岔转换总时间,单位ms
+    pub turnout_transform_time: i32,
+}
+
+impl Default for SimulationConfig {
+    fn default() -> Self {
+        SimulationConfig {
+            turnout_transform_time: 3500,
+        }
+    }
+}

simulation/src/modules/trackside/systems/common.rs

@@ -0,0 +1,31 @@
+use bevy_ecs::{entity::Entity, system::Query};
+use rtss_log::tracing::debug;
+
+use crate::{components::Uid, modules::trackside::components::TwoNormalPositionsTransform};
+
+pub const TWO_NORMAL_POSITION_MIN: i32 = 0;
+pub const TWO_NORMAL_POSITION_MAX: i32 = 100;
+// 两常态位置转换系统
+pub fn two_normal_position_transform(
+    mut query: Query<(Entity, &Uid, &mut TwoNormalPositionsTransform)>,
+) {
+    for (entity, uid, mut transform) in &mut query {
+        debug!(
+            "Entity: {:?}, Uid: {:?}, Conversion: {:?}",
+            entity, uid, transform
+        );
+        if transform.velocity == 0f32 {
+            continue;
+        }
+        let p = transform.position + transform.velocity;
+        if p >= TWO_NORMAL_POSITION_MAX as f32 {
+            transform.position = TWO_NORMAL_POSITION_MAX as f32;
+            transform.velocity = TWO_NORMAL_POSITION_MIN as f32;
+        } else if p <= TWO_NORMAL_POSITION_MIN as f32 {
+            transform.position = TWO_NORMAL_POSITION_MIN as f32;
+            transform.velocity = 0 as f32;
+        } else {
+            transform.position = p;
+        }
+    }
+}

simulation/src/modules/trackside/systems/loading.rs

@@ -0,0 +1,17 @@
+use bevy_ecs::{prelude::Commands, system::ResMut};
+use rtss_log::tracing::debug;
+
+use crate::{
+    components::{SimulationUidEntityMapResource, Uid},
+    modules::trackside::components,
+};
+
+pub fn loading(mut commands: Commands, res_uid_mapping: ResMut<SimulationUidEntityMapResource>) {
+    debug!("loading trackside equipment");
+    let uid = Uid("1".to_string());
+    let et = commands.spawn(components::TurnoutBundle {
+        uid: uid.clone(),
+        ..Default::default()
+    });
+    res_uid_mapping.insert_entity(uid.0, et.id());
+}

simulation/src/modules/trackside/systems/mod.rs

@@ -0,0 +1,6 @@
+mod common;
+mod loading;
+mod turnout;
+pub use common::*;
+pub use loading::*;
+pub use turnout::*;