fix(opcua): reconnect when subscription becomes invalid
This commit is contained in:
parent
0893c9783c
commit
63bcf679c2
|
|
@ -297,31 +297,19 @@ impl ConnectionManager {
|
|||
};
|
||||
|
||||
// 检查订阅状态 - 仅当有 subscription_id 时才检查
|
||||
let subscription_valid = {
|
||||
// 这里使用 set_publishing_mode 直接让服务器校验订阅 ID，避免仅靠会话读请求误判为健康。
|
||||
let subscription_id = {
|
||||
let status = manager.status.read().await;
|
||||
if let Some(conn_status) = status.get(&source_id) {
|
||||
if let Some(_subscription_id) = conn_status.subscription_id {
|
||||
// 尝试读取订阅的属性来验证订阅是否真正有效
|
||||
let node_id = NodeId::new(0, 2253); // ServerServiceLevel节点
|
||||
let read_request = ReadValueId {
|
||||
node_id,
|
||||
attribute_id: AttributeId::Value as u32,
|
||||
index_range: NumericRange::None,
|
||||
data_encoding: Default::default(),
|
||||
};
|
||||
|
||||
match session.read(&[read_request], TimestampsToReturn::Neither, 0f64).await {
|
||||
Ok(_) => true,
|
||||
Err(_) => false,
|
||||
}
|
||||
} else {
|
||||
// 没有 subscription_id 时，认为订阅状态有效（不需要检查）
|
||||
true
|
||||
}
|
||||
} else {
|
||||
// 没有连接状态时，认为订阅状态无效
|
||||
false
|
||||
status.get(&source_id).and_then(|conn_status| conn_status.subscription_id)
|
||||
};
|
||||
let subscription_valid = if let Some(subscription_id) = subscription_id {
|
||||
match session.set_publishing_mode(&[subscription_id], true).await {
|
||||
Ok(results) => results.first().map(|status| status.is_good()).unwrap_or(false),
|
||||
Err(_) => false,
|
||||
}
|
||||
} else {
|
||||
// 没有 subscription_id 时，认为订阅状态有效（不需要检查）
|
||||
true
|
||||
};
|
||||
|
||||
(session_valid, subscription_valid)
|
||||
|
|
@ -1102,6 +1090,8 @@ impl ConnectionManager {
|
|||
let manager = self.clone();
|
||||
let current_source_id = source_id;
|
||||
|
||||
let status_manager = manager.clone();
|
||||
let data_manager = manager.clone();
|
||||
match session
|
||||
.create_subscription(
|
||||
Duration::from_secs(1),
|
||||
|
|
@ -1110,36 +1100,51 @@ impl ConnectionManager {
|
|||
0,
|
||||
0,
|
||||
true,
|
||||
opcua::client::DataChangeCallback::new(move |dv, item| {
|
||||
let client_handle = item.client_handle();
|
||||
let val = dv.value;
|
||||
let timex = Some(Utc::now());
|
||||
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);
|
||||
opcua::client::SubscriptionCallbacks::new(
|
||||
move |notification| {
|
||||
if notification.status.is_bad() {
|
||||
tracing::warn!(
|
||||
"Subscription status changed to {:?} for source {}, triggering reconnection",
|
||||
notification.status,
|
||||
current_source_id
|
||||
);
|
||||
if let Some(tx) = status_manager.reconnect_tx.as_ref() {
|
||||
let _ = tx.send(current_source_id);
|
||||
}
|
||||
}
|
||||
},
|
||||
move |dv, item| {
|
||||
let client_handle = item.client_handle();
|
||||
let val = dv.value;
|
||||
let timex = Some(Utc::now());
|
||||
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);
|
||||
|
||||
if let Some(event_manager) = &manager.event_manager {
|
||||
let _ = event_manager.send(crate::event::ReloadEvent::PointNewValue(
|
||||
crate::telemetry::PointNewValue {
|
||||
source_id: current_source_id,
|
||||
point_id: None,
|
||||
client_handle,
|
||||
value: unified_value,
|
||||
value_type: unified_value_type,
|
||||
value_text: unified_value_text,
|
||||
quality,
|
||||
protocol: "opcua".to_string(),
|
||||
timestamp: timex,
|
||||
scan_mode: ScanMode::Subscribe,
|
||||
}));
|
||||
}
|
||||
}),
|
||||
if let Some(event_manager) = &data_manager.event_manager {
|
||||
let _ = event_manager.send(crate::event::ReloadEvent::PointNewValue(
|
||||
crate::telemetry::PointNewValue {
|
||||
source_id: current_source_id,
|
||||
point_id: None,
|
||||
client_handle,
|
||||
value: unified_value,
|
||||
value_type: unified_value_type,
|
||||
value_text: unified_value_text,
|
||||
quality,
|
||||
protocol: "opcua".to_string(),
|
||||
timestamp: timex,
|
||||
scan_mode: ScanMode::Subscribe,
|
||||
}));
|
||||
}
|
||||
},
|
||||
|_event_fields, _item| {},
|
||||
),
|
||||
)
|
||||
.await
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in New Issue