优化连接流程：connect_from_source自动处理订阅，简化main.rs

migrations/20260224070000_remove_scan_interval_s.sql

@@ -0,0 +1,2 @@
+-- 移除 scan_interval_s 字段，因为现在使用统一的轮询任务
+ALTER TABLE point DROP COLUMN scan_interval_s;

src/connection.rs

@@ -59,8 +59,10 @@ struct PointWriteTarget {
     external_id: String,
 }
 
+#[derive(Debug, Clone)]
 pub struct PollPointInfo {
-    handle: JoinHandle<()>,
+    pub point_id: Uuid,
+    pub external_id: String,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -93,7 +95,8 @@ pub struct ConnectionStatus {
     pub next_client_handle: u32,
     pub client_handle_map: HashMap<u32, Uuid>, // client_handle -> point_id
     pub monitored_item_map: HashMap<Uuid, u32>, // point_id -> monitored_item_id
-    pub poll_points: HashMap<Uuid, PollPointInfo>, // 正在轮询的点集合
+    pub poll_points: Vec<PollPointInfo>, // 正在轮询的点集合
+    poll_handle: Option<JoinHandle<()>>, // 统一的轮询任务句柄
 }
 
 #[derive(Clone)]
@@ -102,7 +105,6 @@ pub struct ConnectionManager {
     point_monitor_data: Arc<RwLock<HashMap<Uuid, PointMonitorInfo>>>,
     point_history_data: Arc<RwLock<HashMap<Uuid, VecDeque<PointMonitorInfo>>>>,
     point_write_target_cache: Arc<RwLock<HashMap<Uuid, PointWriteTarget>>>,
-    pool: Option<sqlx::PgPool>,
     event_manager: Option<std::sync::Arc<crate::event::EventManager>>,
 }
 
@@ -152,10 +154,9 @@ impl ConnectionManager {
         }
     }
 
-    pub fn new_with_pool(pool: sqlx::PgPool) -> Self {
+    pub fn new() -> Self {
         Self {
             status: Arc::new(RwLock::new(HashMap::new())),
-            pool: Some(pool),
             point_monitor_data: Arc::new(RwLock::new(HashMap::new())),
             point_history_data: Arc::new(RwLock::new(HashMap::new())),
             point_write_target_cache: Arc::new(RwLock::new(HashMap::new())),
@@ -216,101 +217,109 @@ impl ConnectionManager {
         self.point_monitor_data.read().await
     }
 
-    async fn start_polling_for_point(
+    async fn start_unified_poll_task(&self, source_id: Uuid, session: Arc<Session>) {
-        &self,
+        let event_manager = match self.event_manager.clone() {
-        source_id: Uuid,
+            Some(em) => em,
-        point: PointSubscriptionInfo,
+            None => {
-        session: Arc<Session>,
+                tracing::warn!("Event manager is not initialized, cannot start unified poll task");
-    ) -> Result<(), String> {
+                return;
-        let interval_s = point.scan_interval_s;
+            }
-        if interval_s <= 0 {
+        };
-            return Err(format!(
-                "Point {} has invalid scan_interval_s {}",
-                point.point_id, point.scan_interval_s
-            ));
-        }
-
-        let node_id = NodeId::from_str(&point.external_id)
-            .map_err(|e| format!("Invalid node id {}: {}", point.external_id, e))?;
-
-        let event_manager = self
-            .event_manager
-            .clone()
-            .ok_or_else(|| "Event manager is not initialized".to_string())?;
 
+        // 停止旧的轮询任务
         {
-            let status = self.status.read().await;
+            let mut status = self.status.write().await;
-            if let Some(conn_status) = status.get(&source_id) {
+            if let Some(conn_status) = status.get_mut(&source_id) {
-                if conn_status.poll_points.contains_key(&point.point_id) {
+                if let Some(handle) = conn_status.poll_handle.take() {
-                    return Ok(());
+                    handle.abort();
                 }
             }
         }
 
-        let point_id = point.point_id;
+        tracing::info!(
-        let external_id = point.external_id.clone();
+            "Starting unified poll task for source {}",
-        let interval_sec_u64 = u64::try_from(interval_s)
+            source_id
-            .map_err(|_| format!("Invalid scan_interval_s {} for point {}", interval_s, point_id))?;
+        );
 
+        // 克隆 status 引用，以便在异步任务中使用
+        let status_ref = self.status.clone();
+
+        // 启动新的轮询任务
         let handle = tokio::spawn(async move {
-            let mut ticker = tokio::time::interval(Duration::from_secs(interval_sec_u64));
+            let mut ticker = tokio::time::interval(Duration::from_secs(1));
             ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
             loop {
                 ticker.tick().await;
 
-                let read_request = ReadValueId {
+                // 在任务内部获取轮询点列表
-                    node_id: node_id.clone(),
+                let poll_points = {
-                    attribute_id: AttributeId::Value as u32,
+                    let status = status_ref.read().await;
-                    index_range: NumericRange::None,
+                    status.get(&source_id)
-                    data_encoding: Default::default(),
+                        .map(|conn_status| conn_status.poll_points.clone())
+                        .unwrap_or_default()
                 };
 
-                match session
+                if poll_points.is_empty() {
-                    .read(&[read_request], TimestampsToReturn::Both, 0f64)
+                    continue;
-                    .await
+                }
-                {
-                    Ok(result) if !result.is_empty() => {
-                        let dv = &result[0];
-                        let val = dv.value.clone();
-                        let unified_value =
-                            val.as_ref().map(crate::telemetry::opcua_variant_to_data);
-                        let unified_value_type =
-                            val.as_ref().map(crate::telemetry::opcua_variant_type);
-                        let unified_value_text = val.as_ref().map(|v| v.to_string());
-                        let quality = dv
-                            .status
-                            .as_ref()
-                            .map(crate::telemetry::PointQuality::from_status_code)
-                            .unwrap_or(crate::telemetry::PointQuality::Unknown);
 
-                        let _ = event_manager.send(crate::event::ReloadEvent::PointNewValue(
+                // 构建批量读取请求
-                            crate::telemetry::PointNewValue {
+                let read_requests: Vec<ReadValueId> = poll_points
-                                source_id,
+                    .iter()
-                                point_id: Some(point_id),
+                    .filter_map(|p| {
-                                client_handle: 0,
+                        NodeId::from_str(&p.external_id).ok().map(|node_id| ReadValueId {
-                                value: unified_value,
+                            node_id,
-                                value_type: unified_value_type,
+                            attribute_id: AttributeId::Value as u32,
-                                value_text: unified_value_text,
+                            index_range: NumericRange::None,
-                                quality,
+                            data_encoding: Default::default(),
-                                protocol: "opcua".to_string(),
+                        })
-                                timestamp: Some(Utc::now()),
+                    })
-                                scan_mode: ScanMode::Poll,
+                    .collect();
-                            },
+
-                        ));
+                if read_requests.is_empty() {
-                    }
+                    continue;
-                    Ok(_) => {
+                }
-                        tracing::warn!(
+
-                            "Poll read returned empty result for point {} node {}",
+                // 执行批量读取
-                            point_id,
+                match session.read(&read_requests, TimestampsToReturn::Both, 0f64).await {
-                            external_id
+                    Ok(results) => {
-                        );
+                        for (i, result) in results.iter().enumerate() {
+                            if i >= poll_points.len() {
+                                break;
+                            }
+
+                            let poll_point = &poll_points[i];
+                            let dv = result;
+                            let val = dv.value.clone();
+                            let unified_value = val.as_ref().map(crate::telemetry::opcua_variant_to_data);
+                            let unified_value_type = val.as_ref().map(crate::telemetry::opcua_variant_type);
+                            let unified_value_text = val.as_ref().map(|v| v.to_string());
+                            let quality = dv.status
+                                .as_ref()
+                                .map(crate::telemetry::PointQuality::from_status_code)
+                                .unwrap_or(crate::telemetry::PointQuality::Unknown);
+
+                            let _ = event_manager.send(crate::event::ReloadEvent::PointNewValue(
+                                crate::telemetry::PointNewValue {
+                                    source_id,
+                                    point_id: Some(poll_point.point_id),
+                                    client_handle: 0,
+                                    value: unified_value,
+                                    value_type: unified_value_type,
+                                    value_text: unified_value_text,
+                                    quality,
+                                    protocol: "opcua".to_string(),
+                                    timestamp: Some(Utc::now()),
+                                    scan_mode: ScanMode::Poll,
+                                },
+                            ));
+                        }
                     }
                     Err(e) => {
                         tracing::warn!(
-                            "Poll read failed for point {} node {}: {:?}",
+                            "Unified poll read failed for source {}: {:?}",
-                            point_id,
+                            source_id,
-                            external_id,
                             e
                         );
                     }
@@ -318,75 +327,42 @@ impl ConnectionManager {
             }
         });
 
-        {
+        // 保存轮询任务句柄
-            let mut status = self.status.write().await;
-            if let Some(conn_status) = status.get_mut(&source_id) {
-                conn_status.poll_points.insert(
-                    point_id,
-                    PollPointInfo { handle },
-                );
-            }
-        }
-
-        Ok(())
-    }
-
-    async fn stop_polling_for_point(&self, point_id: Uuid) {
         let mut status = self.status.write().await;
-        for conn_status in status.values_mut() {
+        if let Some(conn_status) = status.get_mut(&source_id) {
-            if let Some(poll_info) = conn_status.poll_points.remove(&point_id) {
+            conn_status.poll_handle = Some(handle);
-                poll_info.handle.abort();
-            }
         }
     }
 
-    async fn stop_polling_for_source(&self, source_id: Uuid) {
+    // 将点添加到轮询列表
-        let poll_infos = {
+    async fn add_points_to_poll_list(
-            let mut status = self.status.write().await;
-            status
-                .get_mut(&source_id)
-                .map(|conn_status| conn_status.poll_points.drain().collect::<Vec<_>>())
-                .unwrap_or_default()
-        };
-
-        if poll_infos.is_empty() {
-            return;
-        }
-
-        for (_, poll_info) in poll_infos {
-            poll_info.handle.abort();
-        }
-    }
-
-    async fn start_polling_for_points(
         &self,
         source_id: Uuid,
         points: &[PointSubscriptionInfo],
-        session: Arc<Session>,
     ) -> usize {
         let mut started = 0usize;
-        for point in points.iter().cloned() {
+
-            match self
+        // 添加新的轮询点
-                .start_polling_for_point(source_id, point.clone(), session.clone())
+        {
-                .await
+            let mut status = self.status.write().await;
-            {
+            if let Some(conn_status) = status.get_mut(&source_id) {
-                Ok(()) => {
+                for point in points {
-                    started += 1;
+                    // 检查点是否已经在轮询列表中
-                    tracing::info!(
+                    if !conn_status.poll_points.iter().any(|p| p.point_id == point.point_id) {
-                        "Point {} switched to poll mode with scan_interval_s {}",
+                        conn_status.poll_points.push(PollPointInfo {
-                        point.point_id,
+                            point_id: point.point_id,
-                        point.scan_interval_s
+                            external_id: point.external_id.clone(),
-                    );
+                        });
-                }
+                        started += 1;
-                Err(e) => {
+                        tracing::info!(
-                    tracing::warn!(
+                            "Point {} switched to poll mode",
-                        "Point {} cannot switch to poll mode: {}",
+                            point.point_id
-                        point.point_id,
+                        );
-                        e
+                    }
-                    );
                 }
             }
         }
+
         started
     }
 
@@ -419,7 +395,13 @@ impl ConnectionManager {
             source.username.as_deref(),
             source.password.as_deref(),
         )
-        .await
+        .await?;
+
+        // Subscribe to points for this source
+        self.subscribe_points_from_source(source_id, None, pool)
+            .await?;
+
+        Ok(())
     }
 
     pub async fn connect(
@@ -493,7 +475,7 @@ impl ConnectionManager {
         status.insert(
             source_id,
             ConnectionStatus {
-                session: Some(session),
+                session: Some(session.clone()),
                 is_connected: true,
                 last_error: None,
                 last_time: Utc::now(),
@@ -501,9 +483,14 @@ impl ConnectionManager {
                 next_client_handle: 1000,
                 client_handle_map: HashMap::new(),
                 monitored_item_map: HashMap::new(),
-                poll_points: HashMap::new(),
+                poll_points: Vec::new(),
+                poll_handle: None,
             },
         );
+        drop(status); // 显式释放锁，在调用 start_unified_poll_task 之前
+
+        // 启动统一的轮询任务
+        self.start_unified_poll_task(source_id, session).await;
 
         tracing::info!("Successfully connected to source {}", source_id);
         Ok(())
@@ -522,12 +509,22 @@ impl ConnectionManager {
                 client_handle_map: HashMap::new(),
                 monitored_item_map: HashMap::new(),
                 next_client_handle: 1000,
-                poll_points: HashMap::new(),
+                poll_points: Vec::new(),
+                poll_handle: None,
             },
         );
     }
     pub async fn disconnect(&self, source_id: Uuid) -> Result<(), String> {
-        self.stop_polling_for_source(source_id).await;
+        // 停止轮询任务并清空轮询点列表
+        {
+            let mut status = self.status.write().await;
+            if let Some(conn_status) = status.get_mut(&source_id) {
+                conn_status.poll_points.clear();
+                if let Some(handle) = conn_status.poll_handle.take() {
+                    handle.abort();
+                }
+            }
+        }
 
         let conn_status = self.status.write().await.remove(&source_id);
         if let Some(conn_status) = conn_status {
@@ -546,7 +543,16 @@ impl ConnectionManager {
         let source_ids: Vec<Uuid> = self.status.read().await.keys().copied().collect();
 
         for source_id in source_ids {
-            self.stop_polling_for_source(source_id).await;
+            // 停止轮询任务并清空轮询点列表
+            {
+                let mut status = self.status.write().await;
+                if let Some(conn_status) = status.get_mut(&source_id) {
+                    conn_status.poll_points.clear();
+                    if let Some(handle) = conn_status.poll_handle.take() {
+                        handle.abort();
+                    }
+                }
+            }
 
             let conn_status = self.status.write().await.remove(&source_id);
             if let Some(conn_status) = conn_status {
@@ -560,28 +566,13 @@ impl ConnectionManager {
     }
 
     pub async fn get_session(&self, source_id: Uuid) -> Option<Arc<Session>> {
-        // comment fixed
+        let status = self.status.read().await;
-        {
+        if let Some(conn_status) = status.get(&source_id) {
-            let status = self.status.read().await;
+            if conn_status.is_connected {
-            if let Some(conn_status) = status.get(&source_id) {
+                return conn_status.session.clone();
-                if conn_status.is_connected {
-                    return conn_status.session.clone();
-                }
             }
         }
-
+        None
-        // comment fixed
-        if let Some(pool) = &self.pool {
-            if let Ok(()) = self.connect_from_source(pool, source_id).await {
-                // comment fixed
-                let status = self.status.read().await;
-                status.get(&source_id).and_then(|s| s.session.clone())
-            } else {
-                None
-            }
-        } else {
-            None
-        }
     }
 
     pub async fn get_status(&self, source_id: Uuid) -> Option<ConnectionStatusView> {
@@ -930,7 +921,7 @@ impl ConnectionManager {
 
         if subscription_id.is_none() {
             let polled_count = self
-                .start_polling_for_points(source_id, &points, session.clone())
+                .add_points_to_poll_list(source_id, &points)
                 .await;
             return Ok(Self::subscription_result(0, polled_count));
         }
@@ -998,7 +989,7 @@ impl ConnectionManager {
                 }
 
                 let polled_count = self
-                    .start_polling_for_points(source_id, &item_points, session.clone())
+                    .add_points_to_poll_list(source_id, &item_points)
                     .await;
                 return Ok(Self::subscription_result(0, polled_count));
             }
@@ -1020,8 +1011,9 @@ impl ConnectionManager {
                     conn_status
                         .monitored_item_map
                         .insert(point.point_id, monitored_item_result.result.monitored_item_id);
+                    // 从轮询列表中移除该点
+                    conn_status.poll_points.retain(|p| p.point_id != point.point_id);
                 }
-                self.stop_polling_for_point(point.point_id).await;
             } else {
                 tracing::error!(
                     "Failed to create monitored item for point {}: {:?}",
@@ -1053,7 +1045,7 @@ impl ConnectionManager {
         }
 
         let polled_count = self
-            .start_polling_for_points(source_id, &failed_points, session.clone())
+            .add_points_to_poll_list(source_id, &failed_points)
             .await;
 
         Ok(Self::subscription_result(
@@ -1144,9 +1136,6 @@ impl ConnectionManager {
                 .retain(|_, point_id| !removed_set.contains(point_id));
         }
 
-        for point_id in &removed_point_ids {
-            self.stop_polling_for_point(*point_id).await;
-        }
         let _ = self
             .remove_point_write_target_cache_by_point_ids(&removed_point_ids)
             .await;
@@ -1163,9 +1152,27 @@ impl ConnectionManager {
                 history_data.remove(point_id);
             }
         }
+
+        // 从轮询列表中移除传入的点，并记录移除的轮询点数量
+        let polling_removed_count = {
+            let mut status = self.status.write().await;
+            if let Some(conn_status) = status.get_mut(&source_id) {
+                let before_count = conn_status.poll_points.len();
+                conn_status.poll_points.retain(|p| !target_ids.contains(&p.point_id));
+                let after_count = conn_status.poll_points.len();
+                before_count - after_count
+            } else {
+                0
+            }
+        };
+
+        // 计算从订阅点和轮询点移除的总数
+        let total_removed = removed_point_ids.len() + polling_removed_count;
         tracing::info!(
-            "Unsubscribed {} points from source {}",
+            "Unsubscribed {} points (subscription: {}, polling: {}) from source {}",
+            total_removed,
             removed_point_ids.len(),
+            polling_removed_count,
             source_id
         );
         Ok(removed_point_ids.len())

src/event.rs

@@ -12,10 +12,6 @@ pub enum ReloadEvent {
     SourceDelete {
         source_id: Uuid,
     },
-    PointCreate {
-        source_id: Uuid,
-        point_id: Uuid,
-    },
     PointCreateBatch {
         source_id: Uuid,
         point_ids: Vec<Uuid>,
@@ -58,29 +54,6 @@ impl EventManager {
                             tracing::error!("Failed to disconnect from source {}: {}", source_id, e);
                         }
                     }
-                    ReloadEvent::PointCreate { source_id, point_id } => {
-                        match connection_manager
-                            .subscribe_points_from_source(source_id, Some(vec![point_id]), &pool)
-                            .await
-                        {
-                            Ok(stats) => {
-                                let subscribed = *stats.get("subscribed").unwrap_or(&0);
-                                let polled = *stats.get("polled").unwrap_or(&0);
-                                let total = *stats.get("total").unwrap_or(&0);
-                                tracing::info!(
-                                    "PointCreate subscribe finished for source {} point {}: subscribed={}, polled={}, total={}",
-                                    source_id,
-                                    point_id,
-                                    subscribed,
-                                    polled,
-                                    total
-                                );
-                            }
-                            Err(e) => {
-                                tracing::error!("Failed to subscribe to point {}: {}", point_id, e);
-                            }
-                        }
-                    }
                     ReloadEvent::PointCreateBatch { source_id, point_ids } => {
                         let requested_count = point_ids.len();
                         match connection_manager

src/main.rs

@@ -38,7 +38,7 @@ async fn main() {
     let config = AppConfig::from_env().expect("Failed to load configuration");
     let pool = init_database(&config.database_url).await.expect("Failed to initialize database");
 
-    let mut connection_manager = ConnectionManager::new_with_pool(pool.clone());
+    let mut connection_manager = ConnectionManager::new();
     let ws_manager = Arc::new(websocket::WebSocketManager::new());
     let event_manager = Arc::new(EventManager::new(
         pool.clone(),
@@ -48,41 +48,32 @@ async fn main() {
     connection_manager.set_event_manager(event_manager.clone());
     let connection_manager = Arc::new(connection_manager);
 
-    // Connect to all enabled sources
+    // Connect to all enabled sources concurrently
     let sources = service::get_all_enabled_sources(&pool)
         .await
     .expect("Failed to fetch sources");
 
+    // Spawn a task for each source to connect and subscribe concurrently
+    let mut tasks = Vec::new();
     for source in sources {
-        tracing::info!("Connecting to source: {} ({})", source.name, source.endpoint);
+        let cm = connection_manager.clone();
-        match connection_manager.connect_from_source(&pool, source.id).await {
+        let p = pool.clone();
-            Ok(_) => {
+        let source_name = source.name.clone();
-                tracing::info!("Successfully connected to source: {}", source.name);
+        let source_id = source.id;
-                // Subscribe to points for this source
+
-                match connection_manager
+        let task = tokio::spawn(async move {
-                    .subscribe_points_from_source(source.id, None, &pool)
+            if let Err(e) = cm.connect_from_source(&p, source_id).await {
-                    .await
+                tracing::error!("Failed to connect to source {}: {}", source_name, e);
-                {
-                    Ok(stats) => {
-                        let subscribed = *stats.get("subscribed").unwrap_or(&0);
-                        let polled = *stats.get("polled").unwrap_or(&0);
-                        let total = *stats.get("total").unwrap_or(&0);
-                        tracing::info!(
-                            "Point subscribe setup for source {}: subscribed={}, polled={}, total={}",
-                            source.name,
-                            subscribed,
-                            polled,
-                            total
-                        );
-                    }
-                    Err(e) => {
-                        tracing::error!("Failed to subscribe to points for source {}: {}", source.name, e);
-                    }
-                }
-            }
-            Err(e) => {
-                tracing::error!("Failed to connect to source {}: {}", source.name, e);
             }
+        });
+
+        tasks.push(task);
+    }
+
+    // Wait for all connection tasks to complete
+    for task in tasks {
+        if let Err(e) = task.await {
+            tracing::error!("Source connection task failed: {:?}", e);
         }
     }
 

src/model.rs

@@ -92,7 +92,6 @@ pub struct Point {
     pub name: String,
     pub description: Option<String>,
     pub unit: Option<String>,
-    pub scan_interval_s: i32, // s
     pub tag_id: Option<Uuid>,
     #[serde(serialize_with = "utc_to_local_str")]
     pub created_at: DateTime<Utc>,
@@ -104,7 +103,6 @@ pub struct Point {
 pub struct PointSubscriptionInfo {
     pub point_id: Uuid,
     pub external_id: String,
-    pub scan_interval_s: i32,
 }
 
 #[derive(Debug, Serialize, Deserialize, FromRow, Clone)]

src/service.rs

@@ -30,8 +30,7 @@ pub async fn get_points_grouped_by_source(
         SELECT
             p.id as point_id,
             n.source_id,
-            n.external_id,
+            n.external_id
-            p.scan_interval_s
         FROM point p
         INNER JOIN node n ON p.node_id = n.id
         WHERE p.id = ANY($1)
@@ -54,7 +53,6 @@ pub async fn get_points_grouped_by_source(
         let info = PointSubscriptionInfo {
             point_id,
             external_id: row.get("external_id"),
-            scan_interval_s: row.get("scan_interval_s"),
         };
 
         result.entry(source_id).or_default().push(info);
@@ -73,8 +71,7 @@ pub async fn get_points_with_ids(
             r#"
             SELECT
                 p.id as point_id,
-                n.external_id,
+                n.external_id
-                p.scan_interval_s
             FROM point p
             INNER JOIN node n ON p.node_id = n.id
             WHERE n.source_id = $1
@@ -89,8 +86,7 @@ pub async fn get_points_with_ids(
             r#"
             SELECT
                 p.id as point_id,
-                n.external_id,
+                n.external_id
-                p.scan_interval_s
             FROM point p
             INNER JOIN node n ON p.node_id = n.id
             WHERE n.source_id = $1
@@ -110,7 +106,6 @@ pub async fn get_points_with_ids(
         .map(|row| PointSubscriptionInfo {
             point_id: row.get("point_id"),
             external_id: row.get("external_id"),
-            scan_interval_s: row.get("scan_interval_s"),
         })
         .collect())
 }