fix

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java

@@ -361,7 +361,13 @@ public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc
     transferByTablet(
         tabletInsertionEvent,
         LOGGER,
-        (tablet, isTableModel) -> nameSpace.transfer(tablet, isTableModel, this));
+        (tablet, isTableModel) -> {
+          if (Objects.nonNull(nameSpace)) {
+            nameSpace.transfer(tablet, isTableModel, this);
+          } else if (Objects.nonNull(client)) {
+            client.transfer(tablet, this);
+          }
+        });
   }
 
   public static void transferByTablet(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java

@@ -35,6 +35,7 @@
 import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
 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.ExpandedNodeId;
 import org.eclipse.milo.opcua.stack.core.types.builtin.ExtensionObject;
 import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
 import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
@@ -43,25 +44,23 @@
 import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
 import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned;
 import org.eclipse.milo.opcua.stack.core.types.enumerated.NodeClass;
-import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
 import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesItem;
 import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResponse;
 import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResult;
-import org.eclipse.milo.opcua.stack.core.types.structured.DeleteNodesItem;
 import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
 import org.eclipse.milo.opcua.stack.core.types.structured.ObjectAttributes;
 import org.eclipse.milo.opcua.stack.core.types.structured.VariableAttributes;
 
-import java.util.Arrays;
-import java.util.Collections;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Predicate;
 
+import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.convertToOpcDataType;
 import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.timestampToUtc;
 
 public class IoTDBOpcUaClient {
-
+  private static final int NAME_SPACE_INDEX = 2;
   private final String nodeUrl;
 
   private final SecurityPolicy securityPolicy;
@@ -101,6 +100,7 @@ private void transferTabletRowForClientServerModel(
     Object value = null;
     long timestamp = 0;
     NodeId nodeId = null;
+    NodeId opcDataType = null;
 
     for (int i = 0; i < measurementSchemas.size(); ++i) {
       if (Objects.isNull(values.get(i))) {
@@ -120,59 +120,21 @@ private void transferTabletRowForClientServerModel(
         throw new UnsupportedOperationException(
             "When the 'with-quality' mode is enabled, the measurement must be either \"value-name\" or \"quality-name\"");
       }
-      nodeId = new NodeId(2, String.join("/", segments));
+      nodeId = new NodeId(NAME_SPACE_INDEX, String.join("/", segments));
 
       final long utcTimestamp = timestampToUtc(timestamps.get(timestamps.size() > 1 ? i : 0));
       value = values.get(i);
       timestamp = utcTimestamp;
+      opcDataType = convertToOpcDataType(type);
     }
+    final Variant variant = new Variant(value);
     final DataValue dataValue =
-        new DataValue(new Variant(value), currentQuality, new DateTime(timestamp), new DateTime());
+        new DataValue(variant, currentQuality, new DateTime(timestamp), new DateTime());
     StatusCode writeStatus = client.writeValue(nodeId, dataValue).get();
 
     if (writeStatus.getValue() == StatusCodes.Bad_NodeIdUnknown) {
       final AddNodesResponse addStatus =
-          client
-              .addNodes(
-                  Arrays.asList(
-                      new AddNodesItem(
-                          Identifiers.ObjectsFolder.expanded(),
-                          Identifiers.Organizes,
-                          new NodeId(2, "root").expanded(),
-                          new QualifiedName(2, "root"),
-                          NodeClass.Object,
-                          ExtensionObject.encode(
-                              client.getStaticSerializationContext(), createFolder0Attributes()),
-                          Identifiers.FolderType.expanded()),
-                      new AddNodesItem(
-                          new NodeId(2, "root").expanded(),
-                          Identifiers.Organizes,
-                          new NodeId(2, "root/sg").expanded(),
-                          new QualifiedName(2, "sg"),
-                          NodeClass.Object,
-                          ExtensionObject.encode(
-                              client.getStaticSerializationContext(), createFolder1Attributes()),
-                          Identifiers.FolderType.expanded()),
-                      new AddNodesItem(
-                          new NodeId(2, "root/sg").expanded(),
-                          Identifiers.Organizes,
-                          new NodeId(2, "root/sg/d1").expanded(),
-                          new QualifiedName(2, "d2"),
-                          NodeClass.Object,
-                          ExtensionObject.encode(
-                              client.getStaticSerializationContext(), createFolder2Attributes()),
-                          Identifiers.FolderType.expanded()),
-                      new AddNodesItem(
-                          new NodeId(2, "root/sg/d1").expanded(),
-                          Identifiers.Organizes,
-                          new NodeId(2, "root/sg/d1/s2").expanded(),
-                          new QualifiedName(2, "s2"),
-                          NodeClass.Variable,
-                          ExtensionObject.encode(
-                              client.getStaticSerializationContext(),
-                              createPressureSensorAttributes()),
-                          Identifiers.BaseDataVariableType.expanded())))
-              .get();
+          client.addNodes(getNodesToAdd(segments, opcDataType, variant)).get();
       for (final AddNodesResult result : addStatus.getResults()) {
         if (!result.getStatusCode().equals(StatusCode.GOOD)
             && !(result.getStatusCode().getValue() == StatusCodes.Bad_NodeIdExists)) {
@@ -193,6 +155,56 @@ private void transferTabletRowForClientServerModel(
     }
   }
 
+  public List<AddNodesItem> getNodesToAdd(
+      final String[] segments, final NodeId opcDataType, final Variant initialValue) {
+    final List<AddNodesItem> addNodesItems = new ArrayList<>();
+    final StringBuilder sb = new StringBuilder(segments[0]);
+    ExpandedNodeId curNodeId = new NodeId(NAME_SPACE_INDEX, segments[0]).expanded();
+    addNodesItems.add(
+        new AddNodesItem(
+            Identifiers.ObjectsFolder.expanded(),
+            Identifiers.Organizes,
+            curNodeId,
+            new QualifiedName(NAME_SPACE_INDEX, segments[0]),
+            NodeClass.Object,
+            ExtensionObject.encode(
+                client.getStaticSerializationContext(), createFolderAttributes(segments[0])),
+            Identifiers.FolderType.expanded()));
+
+    // segments.length >= 3
+    for (int i = 1; i < segments.length - 1; ++i) {
+      sb.append("/").append(segments[i]);
+      final ExpandedNodeId nextId = new NodeId(NAME_SPACE_INDEX, sb.toString()).expanded();
+      addNodesItems.add(
+          new AddNodesItem(
+              curNodeId,
+              Identifiers.Organizes,
+              nextId,
+              new QualifiedName(NAME_SPACE_INDEX, segments[i]),
+              NodeClass.Object,
+              ExtensionObject.encode(
+                  client.getStaticSerializationContext(), createFolderAttributes(segments[i])),
+              Identifiers.FolderType.expanded()));
+      curNodeId = nextId;
+    }
+
+    final String measurementName = segments[segments.length - 1];
+    sb.append("/").append(measurementName);
+    addNodesItems.add(
+        new AddNodesItem(
+            curNodeId,
+            Identifiers.Organizes,
+            new NodeId(NAME_SPACE_INDEX, sb.toString()).expanded(),
+            new QualifiedName(NAME_SPACE_INDEX, measurementName),
+            NodeClass.Variable,
+            ExtensionObject.encode(
+                client.getStaticSerializationContext(),
+                createMeasurementAttributes(measurementName, opcDataType, initialValue)),
+            Identifiers.BaseDataVariableType.expanded()));
+
+    return addNodesItems;
+  }
+
   /////////////////////////////// Getter ///////////////////////////////
 
   String getNodeUrl() {
@@ -211,99 +223,32 @@ IdentityProvider getIdentityProvider() {
     return identityProvider;
   }
 
-  public void runA(OpcUaClient client) throws Exception {
-    // synchronous connect
-    client.connect().get();
-    System.out.println("✅ 连接成功");
-
-    // 读取标签值c
-    NodeId nodeId = new NodeId(2, "root/sg/d1/s2");
-
-    // 1. 先读取当前值确认节点可访问
-    DataValue readValue = client.readValue(0, TimestampsToReturn.Both, nodeId).get();
-    System.out.println("读取当前值: " + readValue.getValue().getValue());
-    System.out.println("读取状态: " + readValue.getStatusCode());
+  /////////////////////////////// Attribute creator ///////////////////////////////
 
-    // 2. 尝试写入新值
-    Variant newValue = new Variant(42.0f);
-    DataValue writeValue = new DataValue(newValue, StatusCode.GOOD, new DateTime(), new DateTime());
-
-    System.out.println("尝试写入值: " + newValue.getValue());
-
-    StatusCode writeStatus = client.writeValue(nodeId, writeValue).get();
-    System.out.println("写入状态: " + writeStatus);
-
-    client.deleteNodes(Collections.singletonList(new DeleteNodesItem(nodeId, true)));
-
-    AddNodesResponse addStatus =
-        client
-            .addNodes(
-                Arrays.asList(
-                    new AddNodesItem(
-                        Identifiers.ObjectsFolder.expanded(),
-                        Identifiers.Organizes,
-                        new NodeId(2, "root").expanded(),
-                        new QualifiedName(2, "root"),
-                        NodeClass.Object,
-                        ExtensionObject.encode(
-                            client.getStaticSerializationContext(), createFolder0Attributes()),
-                        Identifiers.FolderType.expanded()),
-                    new AddNodesItem(
-                        new NodeId(2, "root").expanded(),
-                        Identifiers.Organizes,
-                        new NodeId(2, "root/sg").expanded(),
-                        new QualifiedName(2, "sg"),
-                        NodeClass.Object,
-                        ExtensionObject.encode(
-                            client.getStaticSerializationContext(), createFolder1Attributes()),
-                        Identifiers.FolderType.expanded()),
-                    new AddNodesItem(
-                        new NodeId(2, "root/sg").expanded(),
-                        Identifiers.Organizes,
-                        new NodeId(2, "root/sg/d1").expanded(),
-                        new QualifiedName(2, "d2"),
-                        NodeClass.Object,
-                        ExtensionObject.encode(
-                            client.getStaticSerializationContext(), createFolder2Attributes()),
-                        Identifiers.FolderType.expanded()),
-                    new AddNodesItem(
-                        new NodeId(2, "root/sg/d1").expanded(),
-                        Identifiers.Organizes,
-                        new NodeId(2, "root/sg/d1/s2").expanded(),
-                        new QualifiedName(2, "s2"),
-                        NodeClass.Variable,
-                        ExtensionObject.encode(
-                            client.getStaticSerializationContext(),
-                            createPressureSensorAttributes()),
-                        Identifiers.BaseDataVariableType.expanded())))
-            .get();
-    System.out.println("新增节点状态: " + addStatus);
-    client.disconnect().get();
-  }
-
-  public static VariableAttributes createPressureSensorAttributes() {
+  public static VariableAttributes createMeasurementAttributes(
+      final String name, final NodeId objectType, final Variant initialValue) {
     return new VariableAttributes(
         Unsigned.uint(0xFFFF), // specifiedAttributes
-        LocalizedText.english("s2"),
-        LocalizedText.english("反应釜压力传感器"),
+        LocalizedText.english(name),
+        LocalizedText.english(name),
         Unsigned.uint(0), // writeMask
         Unsigned.uint(0), // userWriteMask
-        new Variant(101.3f), // 初始压力值 101.3 kPa
-        Identifiers.Float, // 浮点数类型
-        ValueRanks.Scalar, // 标量
+        new Variant(initialValue),
+        objectType,
+        ValueRanks.Scalar,
         null, // arrayDimensions
         AccessLevel.toValue(AccessLevel.READ_WRITE),
         AccessLevel.toValue(AccessLevel.READ_WRITE),
-        500.0, // 500ms 采样间隔
-        false // 启用历史记录
+        500.0, // samplingInterval
+        false // historizing
         );
   }
 
-  public static ObjectAttributes createFolder0Attributes() {
+  public static ObjectAttributes createFolderAttributes(final String name) {
     return new ObjectAttributes(
         Unsigned.uint(0xFFFF), // specifiedAttributes
-        LocalizedText.english("root"),
-        LocalizedText.english("反应釜压力传感器"),
+        LocalizedText.english(name),
+        LocalizedText.english(name),
         Unsigned.uint(0), // writeMask
         Unsigned.uint(0), // userWriteMask
         null // 启用历史记录

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java

@@ -490,7 +490,7 @@ private void transferTabletForPubSubModel(
     eventNode.delete();
   }
 
-  private NodeId convertToOpcDataType(final TSDataType type) {
+  public static NodeId convertToOpcDataType(final TSDataType type) {
     switch (type) {
       case BOOLEAN:
         return Identifiers.Boolean;