Recursive tags container

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/
 

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
         }
       ]
     }

integration/initial-data.mongodb

@@ -1,6 +1,8 @@
 use("opcua");
 
 db.data.insertOne({
-    _id: "integration-tests",
-    alreadyPresent: true
+    _id: "novalue",
+    val: {
+        alreadyPresent: true,
+    },
 });

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)
+})();

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);
                     }
                 }

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(|| {

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() {

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<T>(
+    tag_set: &mut Vec<Tag>,
+    session: Arc<RwLock<T>>,
+    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::*;