fix: Support abstract data types in OPC-UA adapter (#4178)

Author: dominikriemer

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaAdapter.java

@@ -99,7 +99,7 @@ private void prepareAdapter() throws AdapterException {
       this.connectedClient = clientProvider.getClient(this.opcUaAdapterConfig);
       OpcUaNodeBrowser browserClient =
           new OpcUaNodeBrowser(this.connectedClient.getClient(), this.opcUaAdapterConfig);
-      this.nodeProvider = browserClient.makeNodeProvider(List.of());
+      this.nodeProvider = browserClient.makeNodeProvider();
       this.allNodes = nodeProvider.getNodes();
 
       if (opcUaAdapterConfig.inPullMode()) {

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaNodeBrowser.java

@@ -60,10 +60,10 @@ public OpcUaNodeBrowser(
     this.spOpcConfig = spOpcUaClientConfig;
   }
 
-  public OpcUaNodeProvider makeNodeProvider(List<String> runtimeNameFilters) throws UaException {
+  public OpcUaNodeProvider makeNodeProvider() throws UaException {
     var opcNodes = new ArrayList<OpcUaNode>();
     for (String selectedNodeName : this.spOpcConfig.getSelectedNodeNames()) {
-      opcNodes.add(toOpcNode(selectedNodeName, runtimeNameFilters));
+      opcNodes.add(toOpcNode(selectedNodeName));
     }
 
     return new OpcUaNodeProvider(opcNodes);
@@ -79,8 +79,7 @@ public List<TreeInputNode> buildNodeTreeFromOrigin(String nextBaseNodeToResolve)
     return findChildren(client, currentNodeId);
   }
 
-  private OpcUaNode toOpcNode(String nodeName,
-                              List<String> runtimeNamesToDelete) throws UaException {
+  private OpcUaNode toOpcNode(String nodeName) throws UaException {
     AddressSpace addressSpace = getAddressSpace();
 
     NodeId nodeId;
@@ -109,8 +108,9 @@ private OpcUaNode toOpcNode(String nodeName,
     );
 
     if (node instanceof VariableNode) {
+      var dataValue = ((VariableNode) node).getValue();
       var nodeInfo = new BasicVariableNodeInfo((VariableNode) node, spOpcConfig.getNamingStrategy());
-      return OpcUaNodeFactory.createOpcUaNode(nodeInfo, runtimeNamesToDelete);
+      return OpcUaNodeFactory.createOpcUaNode(nodeInfo, dataValue);
     }
 
     LOG.warn("Node {} not of type VariableNode", node.getDisplayName());

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaNodeMetadataExtractor.java

@@ -25,6 +25,7 @@
 import org.eclipse.milo.opcua.stack.core.UaException;
 import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
 import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime;
+import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
 import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
 
 import java.util.HashMap;
@@ -57,10 +58,12 @@ public Map<String, Object> extract() {
         var dataTypeNode = client.getAddressSpace().getNode(dataTypeNodeId);
         var value = client.readValue(0, TimestampsToReturn.Both, node.getNodeId()).get();
 
+        extractNodeId((UaVariableNode) node);
         extractSourceTime(value);
         extractServerTime(value);
         extractStatusCode(value);
         extractDataType(dataTypeNode);
+        extractDataTypeNodeId(dataTypeNodeId);
       } catch (UaException | ExecutionException | InterruptedException e) {
         throw new RuntimeException(e);
       }
@@ -69,20 +72,36 @@ public Map<String, Object> extract() {
     return metadata;
   }
 
+  public void extractNodeId(UaVariableNode node) {
+    if (node != null && node.getNodeId() != null) {
+      add("Node ID", node.getNodeId().toParseableString());
+    } else {
+      add("Node ID", "N/A");
+    }
+  }
+
+  public void extractDataTypeNodeId(NodeId dataTypeNodeId) {
+    if (dataTypeNodeId != null) {
+      add("Data Type Node ID", dataTypeNodeId.toParseableString());
+    } else {
+      add("Data Type Node ID", "N/A");
+    }
+  }
+
 
   public void extractDescription() {
     if (node.getDescription() != null) {
       add("Description", node.getDescription().getText());
     } else {
-      add("Description", "");
+      add("Description", "N/A");
     }
   }
 
   public void extractNamespaceIndex() {
     if (node.getNodeId() != null) {
       add("NamespaceIndex", node.getNodeId().getNamespaceIndex().toString());
     } else {
-      add("NamespaceIndex", "");
+      add("NamespaceIndex", "N/A");
     }
   }
 
@@ -98,15 +117,15 @@ public void extractBrowseName() {
     if (node.getBrowseName() != null) {
       add("BrowseName", node.getBrowseName().getName());
     } else {
-      add("BrowseName", "");
+      add("BrowseName", "N/A");
     }
   }
 
   public void extractDisplayName() {
     if (node.getDisplayName() != null) {
       add("DisplayName", node.getDisplayName().getText());
     } else {
-      add("DisplayName", "");
+      add("DisplayName", "N/A");
     }
   }
 

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/adapter/OpcUaSchemaProvider.java

@@ -27,17 +27,13 @@
 import org.apache.streampipes.extensions.connectors.opcua.model.node.OpcUaNode;
 import org.apache.streampipes.model.connect.guess.FieldStatusInfo;
 import org.apache.streampipes.model.connect.guess.SampleData;
-import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.model.schema.EventSchema;
-import org.apache.streampipes.sdk.builder.adapter.GuessSchemaBuilder;
 import org.apache.streampipes.sdk.builder.adapter.SampleDataBuilder;
 
 import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
 import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
 import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
 import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
 
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -56,11 +52,8 @@ public SampleData getSampleData(OpcUaClientProvider clientProvider,
                                    IAdapterParameterExtractor extractor,
                                    IStreamPipesClient streamPipesClient)
       throws AdapterException, ParseException {
-    var builder = GuessSchemaBuilder.create();
-    EventSchema eventSchema = new EventSchema();
     Map<String, Object> eventPreview = new HashMap<>();
     Map<String, FieldStatusInfo> fieldStatusInfos = new HashMap<>();
-    List<EventProperty> allProperties = new ArrayList<>();
 
     var opcUaConfig = SpOpcUaConfigExtractor.extractAdapterConfig(
         extractor.getStaticPropertyExtractor(),
@@ -70,15 +63,9 @@ public SampleData getSampleData(OpcUaClientProvider clientProvider,
       var connectedClient = clientProvider.getClient(opcUaConfig);
       OpcUaNodeBrowser nodeBrowser =
           new OpcUaNodeBrowser(connectedClient.getClient(), opcUaConfig);
-      var nodeProvider = nodeBrowser.makeNodeProvider(List.of());
+      var nodeProvider = nodeBrowser.makeNodeProvider();
       var selectedNodes = nodeProvider.getNodes();
 
-      if (!selectedNodes.isEmpty()) {
-        for (OpcUaNode opcNode : selectedNodes) {
-          opcNode.addToSchema(connectedClient.getClient(), allProperties);
-        }
-      }
-
       var nodeIds = selectedNodes.stream()
           .map(node -> node.nodeInfo().getNodeId())
           .collect(Collectors.toList());
@@ -96,12 +83,10 @@ public SampleData getSampleData(OpcUaClientProvider clientProvider,
       clientProvider.releaseClient(opcUaConfig);
     }
 
-    var sampleData = SampleDataBuilder.create()
+    return SampleDataBuilder.create()
                      .sample(eventPreview)
+        .fieldStatusInfos(fieldStatusInfos)
                      .build();
-
-
-    return sampleData;
   }
 
   private static void makeEventPreview(

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/model/OpcUaNodeFactory.java

@@ -22,18 +22,89 @@
 import org.apache.streampipes.extensions.connectors.opcua.model.node.ExtensionObjectOpcUaNode;
 import org.apache.streampipes.extensions.connectors.opcua.model.node.OpcUaNode;
 import org.apache.streampipes.extensions.connectors.opcua.model.node.PrimitiveOpcUaNode;
-import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaTypes;
 
-import java.util.List;
+import org.eclipse.milo.opcua.stack.core.BuiltinDataType;
+import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
+import org.eclipse.milo.opcua.stack.core.types.builtin.ExtensionObject;
+import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
+import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
+
+import java.lang.reflect.Array;
+import java.util.Objects;
 
 public class OpcUaNodeFactory {
 
-  public static OpcUaNode createOpcUaNode(BasicVariableNodeInfo nodeInfo,
-                                          List<String> runtimeNamesToDelete) {
-    if (OpcUaTypes.isExtensionOrCustom(nodeInfo.getNode())) {
-      return new ExtensionObjectOpcUaNode(nodeInfo, runtimeNamesToDelete);
-    } else {
-      return new PrimitiveOpcUaNode(nodeInfo, runtimeNamesToDelete);
+  public static OpcUaNode createOpcUaNode(
+      BasicVariableNodeInfo nodeInfo,
+      DataValue dataValue
+  ) {
+    var hasVariant = hasVariant(dataValue);
+    if (hasVariant) {
+      var byValue = isExtensionByValue(dataValue);
+        return byValue
+            ? new ExtensionObjectOpcUaNode(nodeInfo)
+            : new PrimitiveOpcUaNode(nodeInfo);
+    }
+
+    return isExtensionByDataType(nodeInfo)
+        ? new ExtensionObjectOpcUaNode(nodeInfo)
+        : new PrimitiveOpcUaNode(nodeInfo);
+  }
+
+  private static boolean hasVariant(DataValue dataValue) {
+    if (dataValue == null) {
+      return false;
     }
+
+    StatusCode sc = dataValue.getStatusCode();
+    if (sc == null || !sc.isGood()) {
+      return false;
+    }
+
+    return dataValue.getValue() != null;
+  }
+
+  /**
+   * @return TRUE  -> value is (or contains) ExtensionObject
+   *         FALSE -> value is present and not ExtensionObject
+   */
+  private static boolean isExtensionByValue(DataValue dv) {
+    Object v = dv.getValue().getValue();
+    if (v == null) {
+      return false;
+    }
+
+    if (v instanceof ExtensionObject) {
+      return true;
+    }
+
+    // Handle arrays of any kind (Object[] or primitive arrays)
+    Class<?> c = v.getClass();
+    if (c.isArray()) {
+      int len = Array.getLength(v);
+      for (int i = 0; i < len; i++) {
+        Object el = Array.get(v, i);
+        if (el instanceof ExtensionObject) {
+          return true;
+        }
+      }
+      return false;
+    }
+
+    return false;
+  }
+
+  /**
+   * Conservative fallback based on declared DataType only.
+   * Treat only ExtensionObject itself as "extension" here.
+   *
+   * Why so conservative? Because abstract standard types like Integer/Number
+   * are NOT builtins but are still "primitive-ish" and should not be treated
+   * as custom/extension.
+   */
+  private static boolean isExtensionByDataType(BasicVariableNodeInfo nodeInfo) {
+    NodeId dt = nodeInfo.getNode().getDataType();
+    return Objects.equals(dt, BuiltinDataType.ExtensionObject.getNodeId());
   }
 }
+

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/model/node/ExtensionObjectOpcUaNode.java

@@ -19,29 +19,22 @@
 package org.apache.streampipes.extensions.connectors.opcua.model.node;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.extensions.connectors.opcua.utils.OpcUaTypes;
 import org.apache.streampipes.model.connect.guess.FieldStatus;
 import org.apache.streampipes.model.connect.guess.FieldStatusInfo;
-import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.sdk.builder.PrimitivePropertyBuilder;
 
 import org.eclipse.milo.opcua.binaryschema.Struct;
 import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
 import org.eclipse.milo.opcua.stack.core.types.builtin.ExtensionObject;
 import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
 
-import java.util.List;
 import java.util.Map;
 
 public class ExtensionObjectOpcUaNode implements OpcUaNode {
 
   private final BasicVariableNodeInfo nodeInfo;
-  private final List<String> runtimeNamesToDelete;
 
-  public ExtensionObjectOpcUaNode(BasicVariableNodeInfo nodeInfo,
-                                  List<String> runtimeNamesToDelete) {
+  public ExtensionObjectOpcUaNode(BasicVariableNodeInfo nodeInfo) {
     this.nodeInfo = nodeInfo;
-    this.runtimeNamesToDelete = runtimeNamesToDelete;
   }
 
   @Override
@@ -52,26 +45,7 @@ public BasicVariableNodeInfo nodeInfo() {
   @Override
   public int getNumberOfEventProperties(OpcUaClient client) {
     var struct = extractStruct(client, nodeInfo.getNode().getValue().getValue());
-    return (int) struct.getMembers().entrySet().stream()
-        .filter(entry -> {
-          var nodeName = nodeInfo.getDesiredName(entry.getKey());
-          return !runtimeNamesToDelete.contains(nodeName);
-        })
-        .count();
-  }
-
-  @Override
-  public void addToSchema(OpcUaClient client,
-                          List<EventProperty> eventProperties) {
-    var struct = extractStruct(client, nodeInfo.getNode().getValue().getValue());
-    struct.getMembers().forEach((key, member) -> {
-      var nodeName = nodeInfo.getDesiredName(key);
-      eventProperties.add(
-          PrimitivePropertyBuilder.create(OpcUaTypes.getTypeFromValue(member.getValue()), nodeName)
-              .label(nodeName)
-              .build()
-      );
-    });
+    return struct.getMembers().size();
   }
 
   @Override
@@ -82,9 +56,7 @@ public void addToEvent(OpcUaClient client,
 
     struct.getMembers().forEach((key, member) -> {
       var nodeName = nodeInfo.getDesiredName(key);
-      if (!runtimeNamesToDelete.contains(nodeName)) {
-        event.put(nodeName, member.getValue());
-      }
+      event.put(nodeName, member.getValue());
     });
   }
 

streampipes-extensions/streampipes-connectors-opcua/src/main/java/org/apache/streampipes/extensions/connectors/opcua/model/node/OpcUaNode.java

@@ -19,12 +19,10 @@
 package org.apache.streampipes.extensions.connectors.opcua.model.node;
 
 import org.apache.streampipes.model.connect.guess.FieldStatusInfo;
-import org.apache.streampipes.model.schema.EventProperty;
 
 import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
 import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
 
-import java.util.List;
 import java.util.Map;
 
 public interface OpcUaNode {
@@ -33,9 +31,6 @@ public interface OpcUaNode {
 
   int getNumberOfEventProperties(OpcUaClient client);
 
-  void addToSchema(OpcUaClient client,
-                   List<EventProperty> eventProperties);
-
   void addToEvent(OpcUaClient client,
                   Map<String, Object> event,
                   Variant variant);