framework

Author: Caideyipi

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

@@ -20,14 +20,18 @@
 package org.apache.iotdb.db.pipe.sink.protocol.opcua.client;
 
 import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
+import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace;
+import org.apache.iotdb.pipe.api.exception.PipeException;
 
+import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
 import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
 import org.eclipse.milo.opcua.sdk.client.api.identity.IdentityProvider;
 import org.eclipse.milo.opcua.sdk.core.AccessLevel;
 import org.eclipse.milo.opcua.sdk.core.ValueRanks;
 import org.eclipse.milo.opcua.stack.core.Identifiers;
-import org.eclipse.milo.opcua.stack.core.UaException;
+import org.eclipse.milo.opcua.stack.core.StatusCodes;
 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;
@@ -42,24 +46,27 @@
 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.concurrent.atomic.AtomicLong;
+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.timestampToUtc;
+
 public class IoTDBOpcUaClient {
 
   private final String nodeUrl;
 
   private final SecurityPolicy securityPolicy;
   private final IdentityProvider identityProvider;
-
-  private final AtomicLong clientHandles = new AtomicLong(1L);
+  private OpcUaClient client;
 
   public IoTDBOpcUaClient(
       final String nodeUrl,
@@ -70,13 +77,121 @@ public IoTDBOpcUaClient(
     this.identityProvider = identityProvider;
   }
 
-  public void run(OpcUaClient client) throws Exception {
+  public void run(final OpcUaClient client) throws Exception {
     // synchronous connect
+    this.client = client;
     client.connect().get();
   }
 
   // Only support tree model & client-server
-  public void transfer(final Tablet tablet, final OpcUaSink sink) throws UaException {}
+  public void transfer(final Tablet tablet, final OpcUaSink sink) throws Exception {
+    OpcUaNameSpace.transferTabletForClientServerModel(
+        tablet, false, sink, this::transferTabletRowForClientServerModel);
+  }
+
+  private void transferTabletRowForClientServerModel(
+      final String[] segments,
+      final List<IMeasurementSchema> measurementSchemas,
+      final List<Long> timestamps,
+      final List<Object> values,
+      final OpcUaSink sink)
+      throws Exception {
+    StatusCode currentQuality =
+        Objects.isNull(sink.getValueName()) ? StatusCode.GOOD : StatusCode.UNCERTAIN;
+    Object value = null;
+    long timestamp = 0;
+    NodeId nodeId = null;
+
+    for (int i = 0; i < measurementSchemas.size(); ++i) {
+      if (Objects.isNull(values.get(i))) {
+        continue;
+      }
+      final String name = measurementSchemas.get(i).getMeasurementName();
+      final TSDataType type = measurementSchemas.get(i).getType();
+      if (Objects.nonNull(sink.getQualityName()) && sink.getQualityName().equals(name)) {
+        if (!type.equals(TSDataType.BOOLEAN)) {
+          throw new UnsupportedOperationException(
+              "The quality value only supports boolean type, while true == GOOD and false == BAD.");
+        }
+        currentQuality = values.get(i) == Boolean.TRUE ? StatusCode.GOOD : StatusCode.BAD;
+        continue;
+      }
+      if (Objects.nonNull(sink.getValueName()) && !sink.getValueName().equals(name)) {
+        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));
+
+      final long utcTimestamp = timestampToUtc(timestamps.get(timestamps.size() > 1 ? i : 0));
+      value = values.get(i);
+      timestamp = utcTimestamp;
+    }
+    final DataValue dataValue =
+        new DataValue(new Variant(value), 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();
+      for (final AddNodesResult result : addStatus.getResults()) {
+        if (!result.getStatusCode().equals(StatusCode.GOOD)
+            && !(result.getStatusCode().getValue() == StatusCodes.Bad_NodeIdExists)) {
+          throw new PipeException(
+              "Failed to create nodes after transfer data value, write status: "
+                  + writeStatus
+                  + ", creation status: "
+                  + addStatus);
+        }
+      }
+      writeStatus = client.writeValue(nodeId, dataValue).get();
+      if (writeStatus.getValue() != StatusCode.GOOD.getValue()) {
+        throw new PipeException(
+            "Failed to transfer dataValue after successfully created nodes, error: " + writeStatus);
+      }
+    } else {
+      throw new PipeException("Failed to transfer dataValue, error: " + writeStatus);
+    }
+  }
 
   /////////////////////////////// Getter ///////////////////////////////
 

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

@@ -96,7 +96,7 @@ public void shutdown() {
   }
 
   public void transfer(final Tablet tablet, final boolean isTableModel, final OpcUaSink sink)
-      throws UaException {
+      throws Exception {
     if (sink.isClientServerModel()) {
       transferTabletForClientServerModel(
           tablet, isTableModel, sink, this::transferTabletRowForClientServerModel);
@@ -109,7 +109,8 @@ public static void transferTabletForClientServerModel(
       final Tablet tablet,
       final boolean isTableModel,
       final OpcUaSink sink,
-      final TabletRowConsumer consumer) {
+      final TabletRowConsumer consumer)
+      throws Exception {
     final List<IMeasurementSchema> schemas = tablet.getSchemas();
     final List<IMeasurementSchema> newSchemas = new ArrayList<>();
     if (!isTableModel) {
@@ -177,7 +178,8 @@ void accept(
         final List<IMeasurementSchema> measurementSchemas,
         final List<Long> timestamps,
         final List<Object> values,
-        final OpcUaSink sink);
+        final OpcUaSink sink)
+        throws Exception;
   }
 
   private void transferTabletRowForClientServerModel(
@@ -357,7 +359,7 @@ private static Object getTabletObjectValue4Opc(
     }
   }
 
-  private static long timestampToUtc(final long timeStamp) {
+  public static long timestampToUtc(final long timeStamp) {
     return TimestampPrecisionUtils.currPrecision.toNanos(timeStamp) / 100L + 116444736000000000L;
   }