migrations/20260224070000_remove_scan_interval_s.sql

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

src/connection.rs

@@ -59,10 +59,8 @@ struct PointWriteTarget {
     external_id: String,
 }
 
-#[derive(Debug, Clone)]
 pub struct PollPointInfo {
-    pub point_id: Uuid,
+    handle: JoinHandle<()>,
-    pub external_id: String,
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -95,8 +93,7 @@ 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: Vec<PollPointInfo>, // 正在轮询的点集合
+    pub poll_points: HashMap<Uuid, PollPointInfo>, // 正在轮询的点集合
-    poll_handle: Option<JoinHandle<()>>, // 统一的轮询任务句柄
 }
 
 #[derive(Clone)]
@@ -105,6 +102,7 @@ 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>>,
 }
 
@@ -154,9 +152,10 @@ impl ConnectionManager {
         }
     }
 
-    pub fn new() -> Self {
+    pub fn new_with_pool(pool: sqlx::PgPool) -> 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())),
@@ -217,109 +216,101 @@ impl ConnectionManager {
         self.point_monitor_data.read().await
     }
 
-    async fn start_unified_poll_task(&self, source_id: Uuid, session: Arc<Session>) {
+    async fn start_polling_for_point(
-        let event_manager = match self.event_manager.clone() {
+        &self,
-            Some(em) => em,
+        source_id: Uuid,
-            None => {
+        point: PointSubscriptionInfo,
-                tracing::warn!("Event manager is not initialized, cannot start unified poll task");
+        session: Arc<Session>,
-                return;
+    ) -> Result<(), String> {
-            }
+        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 mut status = self.status.write().await;
+            let status = self.status.read().await;
-            if let Some(conn_status) = status.get_mut(&source_id) {
+            if let Some(conn_status) = status.get(&source_id) {
-                if let Some(handle) = conn_status.poll_handle.take() {
+                if conn_status.poll_points.contains_key(&point.point_id) {
-                    handle.abort();
+                    return Ok(());
                 }
             }
         }
 
-        tracing::info!(
+        let point_id = point.point_id;
-            "Starting unified poll task for source {}",
+        let external_id = point.external_id.clone();
-            source_id
+        let interval_sec_u64 = u64::try_from(interval_s)
-        );
+            .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(1));
+            let mut ticker = tokio::time::interval(Duration::from_secs(interval_sec_u64));
             ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
             loop {
                 ticker.tick().await;
 
-                // 在任务内部获取轮询点列表
+                let read_request = ReadValueId {
-                let poll_points = {
+                    node_id: node_id.clone(),
-                    let status = status_ref.read().await;
+                    attribute_id: AttributeId::Value as u32,
-                    status.get(&source_id)
+                    index_range: NumericRange::None,
-                        .map(|conn_status| conn_status.poll_points.clone())
+                    data_encoding: Default::default(),
-                        .unwrap_or_default()
                 };
 
-                if poll_points.is_empty() {
+                match session
-                    continue;
+                    .read(&[read_request], TimestampsToReturn::Both, 0f64)
-                }
+                    .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(
-                let read_requests: Vec<ReadValueId> = poll_points
+                            crate::telemetry::PointNewValue {
-                    .iter()
+                                source_id,
-                    .filter_map(|p| {
+                                point_id: Some(point_id),
-                        NodeId::from_str(&p.external_id).ok().map(|node_id| ReadValueId {
+                                client_handle: 0,
-                            node_id,
+                                value: unified_value,
-                            attribute_id: AttributeId::Value as u32,
+                                value_type: unified_value_type,
-                            index_range: NumericRange::None,
+                                value_text: unified_value_text,
-                            data_encoding: Default::default(),
+                                quality,
-                        })
+                                protocol: "opcua".to_string(),
-                    })
+                                timestamp: Some(Utc::now()),
-                    .collect();
+                                scan_mode: ScanMode::Poll,
-
+                            },
-                if read_requests.is_empty() {
+                        ));
-                    continue;
+                    }
-                }
+                    Ok(_) => {
-
+                        tracing::warn!(
-                // 执行批量读取
+                            "Poll read returned empty result for point {} node {}",
-                match session.read(&read_requests, TimestampsToReturn::Both, 0f64).await {
+                            point_id,
-                    Ok(results) => {
+                            external_id
-                        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!(
-                            "Unified poll read failed for source {}: {:?}",
+                            "Poll read failed for point {} node {}: {:?}",
-                            source_id,
+                            point_id,
+                            external_id,
                             e
                         );
                     }
@@ -327,42 +318,75 @@ impl ConnectionManager {
             }
         });
 
-        // 保存轮询任务句柄
-        let mut status = self.status.write().await;
-        if let Some(conn_status) = status.get_mut(&source_id) {
-            conn_status.poll_handle = Some(handle);
-        }
-    }
-
-    // 将点添加到轮询列表
-    async fn add_points_to_poll_list(
-        &self,
-        source_id: Uuid,
-        points: &[PointSubscriptionInfo],
-    ) -> usize {
-        let mut started = 0usize;
-
-        // 添加新的轮询点
         {
             let mut status = self.status.write().await;
             if let Some(conn_status) = status.get_mut(&source_id) {
-                for point in points {
+                conn_status.poll_points.insert(
-                    // 检查点是否已经在轮询列表中
+                    point_id,
-                    if !conn_status.poll_points.iter().any(|p| p.point_id == point.point_id) {
+                    PollPointInfo { handle },
-                        conn_status.poll_points.push(PollPointInfo {
+                );
-                            point_id: point.point_id,
-                            external_id: point.external_id.clone(),
-                        });
-                        started += 1;
-                        tracing::info!(
-                            "Point {} switched to poll mode",
-                            point.point_id
-                        );
-                    }
-                }
             }
         }
 
+        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(poll_info) = conn_status.poll_points.remove(&point_id) {
+                poll_info.handle.abort();
+            }
+        }
+    }
+
+    async fn stop_polling_for_source(&self, source_id: Uuid) {
+        let poll_infos = {
+            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
+            {
+                Ok(()) => {
+                    started += 1;
+                    tracing::info!(
+                        "Point {} switched to poll mode with scan_interval_s {}",
+                        point.point_id,
+                        point.scan_interval_s
+                    );
+                }
+                Err(e) => {
+                    tracing::warn!(
+                        "Point {} cannot switch to poll mode: {}",
+                        point.point_id,
+                        e
+                    );
+                }
+            }
+        }
         started
     }
 
@@ -395,13 +419,7 @@ 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(
@@ -475,7 +493,7 @@ impl ConnectionManager {
         status.insert(
             source_id,
             ConnectionStatus {
-                session: Some(session.clone()),
+                session: Some(session),
                 is_connected: true,
                 last_error: None,
                 last_time: Utc::now(),
@@ -483,14 +501,9 @@ impl ConnectionManager {
                 next_client_handle: 1000,
                 client_handle_map: HashMap::new(),
                 monitored_item_map: HashMap::new(),
-                poll_points: Vec::new(),
+                poll_points: HashMap::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(())
@@ -509,22 +522,12 @@ impl ConnectionManager {
                 client_handle_map: HashMap::new(),
                 monitored_item_map: HashMap::new(),
                 next_client_handle: 1000,
-                poll_points: Vec::new(),
+                poll_points: HashMap::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 {
@@ -543,16 +546,7 @@ 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 {
@@ -566,13 +560,28 @@ impl ConnectionManager {
     }
 
     pub async fn get_session(&self, source_id: Uuid) -> Option<Arc<Session>> {
-        let status = self.status.read().await;
+        // comment fixed
-        if let Some(conn_status) = status.get(&source_id) {
+        {
-            if conn_status.is_connected {
+            let status = self.status.read().await;
-                return conn_status.session.clone();
+            if let Some(conn_status) = status.get(&source_id) {
+                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> {
@@ -921,7 +930,7 @@ impl ConnectionManager {
 
         if subscription_id.is_none() {
             let polled_count = self
-                .add_points_to_poll_list(source_id, &points)
+                .start_polling_for_points(source_id, &points, session.clone())
                 .await;
             return Ok(Self::subscription_result(0, polled_count));
         }
@@ -989,7 +998,7 @@ impl ConnectionManager {
                 }
 
                 let polled_count = self
-                    .add_points_to_poll_list(source_id, &item_points)
+                    .start_polling_for_points(source_id, &item_points, session.clone())
                     .await;
                 return Ok(Self::subscription_result(0, polled_count));
             }
@@ -1011,9 +1020,8 @@ 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 {}: {:?}",
@@ -1045,7 +1053,7 @@ impl ConnectionManager {
         }
 
         let polled_count = self
-            .add_points_to_poll_list(source_id, &failed_points)
+            .start_polling_for_points(source_id, &failed_points, session.clone())
             .await;
 
         Ok(Self::subscription_result(
@@ -1136,6 +1144,9 @@ 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;
@@ -1152,27 +1163,9 @@ 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 (subscription: {}, polling: {}) from source {}",
+            "Unsubscribed {} points from source {}",
-            total_removed,
             removed_point_ids.len(),
-            polling_removed_count,
             source_id
         );
         Ok(removed_point_ids.len())

src/event.rs

@@ -12,6 +12,10 @@ pub enum ReloadEvent {
     SourceDelete {
         source_id: Uuid,
     },
+    PointCreate {
+        source_id: Uuid,
+        point_id: Uuid,
+    },
     PointCreateBatch {
         source_id: Uuid,
         point_ids: Vec<Uuid>,
@@ -54,6 +58,29 @@ 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();
+    let mut connection_manager = ConnectionManager::new_with_pool(pool.clone());
     let ws_manager = Arc::new(websocket::WebSocketManager::new());
     let event_manager = Arc::new(EventManager::new(
         pool.clone(),
@@ -48,32 +48,41 @@ async fn main() {
     connection_manager.set_event_manager(event_manager.clone());
     let connection_manager = Arc::new(connection_manager);
 
-    // Connect to all enabled sources concurrently
+    // Connect to all enabled sources
     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 {
-        let cm = connection_manager.clone();
+        tracing::info!("Connecting to source: {} ({})", source.name, source.endpoint);
-        let p = pool.clone();
+        match connection_manager.connect_from_source(&pool, source.id).await {
-        let source_name = source.name.clone();
+            Ok(_) => {
-        let source_id = source.id;
+                tracing::info!("Successfully connected to source: {}", source.name);
-
+                // Subscribe to points for this source
-        let task = tokio::spawn(async move {
+                match connection_manager
-            if let Err(e) = cm.connect_from_source(&p, source_id).await {
+                    .subscribe_points_from_source(source.id, None, &pool)
-                tracing::error!("Failed to connect to source {}: {}", source_name, e);
+                    .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!(
+                            "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,6 +92,7 @@ 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>,
@@ -103,6 +104,7 @@ 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,7 +30,8 @@ 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)
@@ -53,6 +54,7 @@ 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);
@@ -71,7 +73,8 @@ 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
@@ -86,7 +89,8 @@ 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
@@ -106,6 +110,7 @@ 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())
 }