trilog

Author: Caideyipi

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

@@ -76,9 +76,7 @@ public void run(OpcUaClient client) throws Exception {
   }
 
   // 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 UaException {}
 
   /////////////////////////////// Getter ///////////////////////////////
 

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

@@ -98,14 +98,18 @@ public void shutdown() {
   public void transfer(final Tablet tablet, final boolean isTableModel, final OpcUaSink sink)
       throws UaException {
     if (sink.isClientServerModel()) {
-      transferTabletForClientServerModel(tablet, isTableModel, sink);
+      transferTabletForClientServerModel(
+          tablet, isTableModel, sink, this::transferTabletRowForClientServerModel);
     } else {
       transferTabletForPubSubModel(tablet, isTableModel, sink);
     }
   }
 
-  private void transferTabletForClientServerModel(
-      final Tablet tablet, final boolean isTableModel, final OpcUaSink sink) {
+  public static void transferTabletForClientServerModel(
+      final Tablet tablet,
+      final boolean isTableModel,
+      final OpcUaSink sink,
+      final TabletRowConsumer consumer) {
     final List<IMeasurementSchema> schemas = tablet.getSchemas();
     final List<IMeasurementSchema> newSchemas = new ArrayList<>();
     if (!isTableModel) {
@@ -126,8 +130,7 @@ private void transferTabletForClientServerModel(
         }
       }
 
-      transferTabletRowForClientServerModel(
-          tablet.getDeviceId().split("\\."), newSchemas, timestamps, values, sink);
+      consumer.accept(tablet.getDeviceId().split("\\."), newSchemas, timestamps, values, sink);
     } else {
       new PipeTableModelTabletEventSorter(tablet).sortByTimestampIfNecessary();
 
@@ -150,7 +153,7 @@ private void transferTabletForClientServerModel(
         }
 
         final int finalI = i;
-        transferTabletRowForClientServerModel(
+        consumer.accept(
             folderSegments,
             newSchemas,
             Collections.singletonList(tablet.getTimestamp(i)),
@@ -167,6 +170,16 @@ private void transferTabletForClientServerModel(
     }
   }
 
+  @FunctionalInterface
+  public interface TabletRowConsumer {
+    void accept(
+        final String[] segments,
+        final List<IMeasurementSchema> measurementSchemas,
+        final List<Long> timestamps,
+        final List<Object> values,
+        final OpcUaSink sink);
+  }
+
   private void transferTabletRowForClientServerModel(
       final String[] segments,
       final List<IMeasurementSchema> measurementSchemas,