From 2b95391d16e79dad76d907eb6fe27f78280478c8 Mon Sep 17 00:00:00 2001 From: Arnaud Rocher Date: Mon, 16 Oct 2023 11:35:24 +0000 Subject: [PATCH 1/4] fix: remove unneeded clone --- src/db.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/db.rs b/src/db.rs index 6c1d7ffa..c98759d8 100644 --- a/src/db.rs +++ b/src/db.rs @@ -122,10 +122,7 @@ impl MongoDB { doc! { "$addFields": updates_doc }, ]; let query = doc! { "_id": message.partner_id }; - if let Err(err) = collection - .update_one(query.clone(), update, options.clone()) - .await - { + if let Err(err) = collection.update_one(query, update, options.clone()).await { error!(when = "updating document", %err); } } From 4acc19b7801c1518a56c2d94d2ebb6be1dcc2f82 Mon Sep 17 00:00:00 2001 From: Arnaud Rocher Date: Mon, 16 Oct 2023 11:35:50 +0000 Subject: [PATCH 2/4] chore: refactor useless match --- src/opcua/session.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/src/opcua/session.rs b/src/opcua/session.rs index 1619aeb8..5de33e2e 100644 --- a/src/opcua/session.rs +++ b/src/opcua/session.rs @@ -58,13 +58,11 @@ pub(super) fn create_session( ) .into(); - let session = match client.connect_to_endpoint(endpoint, user_identity_token) { - Ok(session) => session, - Err(err) => { + let session = client + .connect_to_endpoint(endpoint, user_identity_token) + .map_err(|err| { error!(kind = "endpoint connection", %err); - return Err(()); - } - }; + })?; { let mut session = session.try_write_for(SESSION_LOCK_TIMEOUT).ok_or_else(|| { From c811755e03f2c1eac962828db482cb57781a83cb Mon Sep 17 00:00:00 2001 From: Arnaud Rocher Date: Mon, 16 Oct 2023 11:37:13 +0000 Subject: [PATCH 3/4] feat: chunk monitored items creation Prevents request timing out. --- src/opcua/subscription.rs | 20 +++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/src/opcua/subscription.rs b/src/opcua/subscription.rs index ab71507d..6fd78bb9 100644 --- a/src/opcua/subscription.rs +++ b/src/opcua/subscription.rs @@ -104,15 +104,17 @@ where let session = session .try_read_for(SESSION_LOCK_TIMEOUT) .ok_or_else(|| error!(kind = "session lock timeout"))?; - session - .create_monitored_items( - subscription_id, - TimestampsToReturn::Source, - &items_to_create, - ) - .map_err(|err| { - error!(kind = "monitored items creation", %err); - })? + let mut create_results = Vec::with_capacity(items_to_create.len()); + for (chunk_index, items_chunk) in items_to_create.chunks(50).enumerate() { + let results = session + .create_monitored_items(subscription_id, TimestampsToReturn::Source, items_chunk) + .map_err(|err| { + let chunk_size = items_chunk.len(); + error!(kind = "monitored items creation", %err, chunk_index, chunk_size); + })?; + create_results.extend(results); + } + create_results }; for (i, MonitoredItemCreateResult { status_code, .. }) in results.iter().enumerate() { From de48daab2c80e2557c743009417469651529766f Mon Sep 17 00:00:00 2001 From: Arnaud Rocher Date: Tue, 17 Oct 2023 08:11:31 +0200 Subject: [PATCH 4/4] feat: implement recursive tag container --- README.md | 3 +- integration/config-api/config.json | 5 ++ integration/initial-data.mongodb | 6 +- integration/tests.mongodb | 62 +++++-------- src/opcua/tag_set.rs | 135 +++++++++++++++++------------ 5 files changed, 116 insertions(+), 95 deletions(-) diff --git a/README.md b/README.md index 20932947..103b826c 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,7 @@ Each element of `tags` array is an object with a `type` property and is of one o #### Nodes container -This format allows to collect tags from an OPC-UA container node which has [`HasComponent`][hascomponent] forward reference(s). Each referenced node will be added to the tag set to monitor, with its [`DisplayName`][displayname] as tag name. +This format allows to recursively collect tags from an OPC-UA container node which has [`HasComponent`][hascomponent] or [`Organizes`][organizes] forward reference(s). Each referenced node will be added to the tag set to monitor, with its [`DisplayName`][displayname] as tag name. In case of recursion, the tag name will be a chain of the `DisplayName`s, joined with dots. ```jsonc { @@ -89,6 +89,7 @@ This format directly refers to a single tag. _\* identifier type will be inferred from JSON type._ [hascomponent]: https://reference.opcfoundation.org/Core/Part3/v105/docs/7.7 +[organizes]: https://reference.opcfoundation.org/Core/Part3/docs/7.11 [displayname]: https://reference.opcfoundation.org/Core/Part3/5.2.5/ [nodeid]: https://reference.opcfoundation.org/v104/Core/docs/Part3/8.2.1/ diff --git a/integration/config-api/config.json b/integration/config-api/config.json index a9465bc3..2e746279 100644 --- a/integration/config-api/config.json +++ b/integration/config-api/config.json @@ -39,6 +39,11 @@ "name": "slowNumberOfUpdates", "namespaceUri": "http://microsoft.com/Opc/OpcPlc/", "nodeIdentifier": "SlowNumberOfUpdates" + }, + { + "type": "container", + "namespaceUri": "http://microsoft.com/Opc/OpcPlc/Boiler", + "nodeIdentifier": 5017 } ] } diff --git a/integration/initial-data.mongodb b/integration/initial-data.mongodb index be5d2520..40198857 100644 --- a/integration/initial-data.mongodb +++ b/integration/initial-data.mongodb @@ -1,6 +1,8 @@ use("opcua"); db.data.insertOne({ - _id: "integration-tests", - alreadyPresent: true + _id: "novalue", + val: { + alreadyPresent: true, + }, }); diff --git a/integration/tests.mongodb b/integration/tests.mongodb index 936c78e9..97971513 100644 --- a/integration/tests.mongodb +++ b/integration/tests.mongodb @@ -1,54 +1,40 @@ const assert = require("node:assert/strict"); -db = connect("mongodb://localhost/opcua"); +const db = connect("mongodb://localhost/opcua"); -result = db.data.findOne({ _id: "novalue" }); +(function testNoValue() { + const result = db.data.findOne({ _id: "novalue" }); -print(EJSON.stringify(result)); + print(EJSON.stringify(result)); -assert.ok(!Object.hasOwn(result, "alreadyPresent")); + assert.equal(Object.keys(result.val).length, 0); -assert.equal(Object.keys(result.val).length, 0); + assert.equal(Object.keys(result.ts).length, 0); -assert.equal(Object.keys(result.ts).length, 0); + assert.equal(result.updatedAt.toISOString(), "1969-12-31T23:59:59.000Z"); +})(); -assert.equal(result.updatedAt.toISOString(), "1969-12-31T23:59:59.000Z"); +(function testIntegration() { + const result = db.data.findOne({ _id: "integration-tests" }); -result = db.data.findOne( - { _id: "integration-tests" }, - { - state: "$val.State", - slowNumberOfUpdates: "$val.slowNumberOfUpdates", - alternatingBoolean: "$val.AlternatingBoolean", - randomUnsignedInt32: "$val.RandomUnsignedInt32", - timeDiff: { - $abs: { - $subtract: [ - { $dateFromString: { dateString: "$val.CurrentTime" } }, - { $dateFromString: { dateString: "$val.StartTime" } }, - ], - }, - }, - timestampDiff: { - $dateDiff: { - startDate: "$ts.State", - endDate: "$$NOW", - unit: "millisecond", - }, - }, - } -); + print(EJSON.stringify(result)); -print(EJSON.stringify(result)); + assert.equal(result.val.State, 0); -assert.equal(result.state, 0); + assert.equal(result.val.slowNumberOfUpdates, -1); -assert.equal(result.slowNumberOfUpdates, -1); + assert.equal(typeof result.val.AlternatingBoolean, "boolean"); -assert.equal(typeof result.alternatingBoolean, "boolean"); + assert.ok(result.val.RandomUnsignedInt32 > 0, "RandomUnsignedInt32"); -assert.ok(result.randomUnsignedInt32 > 0); + const timeDiff = + Date.parse(result.val.CurrentTime) - Date.parse(result.val.StartTime); + assert.ok(timeDiff > 0, `timeDiff = ${timeDiff}`); -assert.ok(result.timeDiff > 0); + const timestampDiff = Date.now() - result.ts.State; + assert.ok(timestampDiff > 0, `timestampDiff = ${timestampDiff}`); -assert.ok(result.timestampDiff > 0); + assert.equal(result.val.ParameterSet.BaseTemperature, 10) + + assert.equal(result.val.ParameterSet.TargetTemperature, 80) +})(); diff --git a/src/opcua/tag_set.rs b/src/opcua/tag_set.rs index 3c622644..9cc1f73a 100644 --- a/src/opcua/tag_set.rs +++ b/src/opcua/tag_set.rs @@ -52,60 +52,8 @@ where error!(kind = "namespace not found", namespace_uri); })?; let node_id = NodeId::new(*namespace, node_identifier.clone()); - let result_mask = (BrowseDescriptionResultMask::RESULT_MASK_DISPLAY_NAME - | BrowseDescriptionResultMask::RESULT_MASK_REFERENCE_TYPE) - .bits(); - let browse_description = BrowseDescription { - node_id: node_id.clone(), - browse_direction: BrowseDirection::Forward, - reference_type_id: ReferenceTypeId::HierarchicalReferences.into(), - include_subtypes: true, - node_class_mask: NodeClassMask::VARIABLE.bits(), - result_mask, - }; - let browse_result = { - let session = session.try_read_for(SESSION_LOCK_TIMEOUT).ok_or_else(|| { - error!(kind = "session lock timeout"); - })?; - session - .browse(&[browse_description]) - .map_err(|err| { - error!(kind="Browse request", %err); - })? - .unwrap() - .pop() - .ok_or_else(|| { - error!(kind = "empty Browse results"); - })? - }; - if !browse_result.status_code.is_good() { - let status_code = browse_result.status_code; - error!(kind = "BrowseResult", %status_code); - return Err(()); - } - if !browse_result.continuation_point.is_null() { - error!(kind = "unimplemented ContinuationPoint"); - return Err(()); - } - let references = browse_result.references.ok_or_else(|| { - error!(kind = "NodeId is missing forward reference", %node_id); - })?; - for ReferenceDescription { - node_id, - display_name, - .. - } in references.into_iter().filter(|ref_description| { - use ReferenceTypeId::*; - matches!( - ref_description.reference_type_id.as_reference_type_id(), - Ok(HasComponent | Organizes) - ) - }) { - tag_set.push(Tag { - name: display_name.to_string(), - node_id: node_id.node_id, - }); - } + let cloned_session = Arc::clone(&session); + add_browsed_nodes(&mut tag_set, cloned_session, node_id, "")?; } TagsConfigGroup::Tag { name, @@ -127,6 +75,85 @@ where Ok(tag_set) } +fn add_browsed_nodes( + tag_set: &mut Vec, + session: Arc>, + node_id: NodeId, + name_chain: &str, +) -> Result<(), ()> +where + T: ViewService, +{ + const NODE_CLASS_MASK: u32 = NodeClassMask::VARIABLE.bits() | NodeClassMask::OBJECT.bits(); + const RESULT_MASK: u32 = BrowseDescriptionResultMask::RESULT_MASK_DISPLAY_NAME.bits() + | BrowseDescriptionResultMask::RESULT_MASK_REFERENCE_TYPE.bits() + | BrowseDescriptionResultMask::RESULT_MASK_TYPE_DEFINITION.bits(); + let browse_description = BrowseDescription { + node_id: node_id.clone(), + browse_direction: BrowseDirection::Forward, + reference_type_id: ReferenceTypeId::HierarchicalReferences.into(), + include_subtypes: true, + node_class_mask: NODE_CLASS_MASK, + result_mask: RESULT_MASK, + }; + let browse_result = { + let session = session.try_read_for(SESSION_LOCK_TIMEOUT).ok_or_else(|| { + error!(kind = "session lock timeout"); + })?; + session + .browse(&[browse_description]) + .map_err(|err| { + error!(kind="Browse request", %err); + })? + .unwrap() + .pop() + .ok_or_else(|| { + error!(kind = "empty Browse results"); + })? + }; + if !browse_result.status_code.is_good() { + let status_code = browse_result.status_code; + error!(kind = "BrowseResult", %status_code); + return Err(()); + } + if !browse_result.continuation_point.is_null() { + error!(kind = "unimplemented ContinuationPoint"); + return Err(()); + } + let references = browse_result.references.ok_or_else(|| { + error!(kind = "NodeId is missing forward reference", %node_id); + })?; + for ReferenceDescription { + node_id, + display_name, + type_definition, + .. + } in references.into_iter().filter(|ref_description| { + use ReferenceTypeId::*; + matches!( + ref_description.reference_type_id.as_reference_type_id(), + Ok(HasComponent | Organizes) + ) + }) { + let new_name_chain = if name_chain.is_empty() { + display_name.to_string() + } else { + format!("{name_chain}.{display_name}") + }; + if type_definition.node_id == VariableTypeId::BaseDataVariableType.into() { + tag_set.push(Tag { + name: new_name_chain, + node_id: node_id.node_id, + }); + } else { + let cloned_session = Arc::clone(&session); + add_browsed_nodes(tag_set, cloned_session, node_id.node_id, &new_name_chain)?; + } + } + + Ok(()) +} + #[cfg(test)] mod tests { use super::*;