fix

Author: Caideyipi

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

@@ -24,6 +24,7 @@
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.ClientRunner;
 import org.apache.iotdb.db.pipe.sink.protocol.opcua.client.IoTDBOpcUaClient;
 import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace;
 import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaServerBuilder;
@@ -321,6 +322,7 @@ private void customizeClient(final String nodeUrl, final PipeParameters paramete
                 parameters.getStringByKeys(CONNECTOR_IOTDB_PASSWORD_KEY, SINK_IOTDB_PASSWORD_KEY))
             : new AnonymousProvider();
     client = new IoTDBOpcUaClient(nodeUrl, policy, provider);
+    new ClientRunner(client).run();
   }
 
   @Override

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

@@ -68,74 +68,6 @@ public IoTDBOpcUaClient(
   public void run(OpcUaClient client) throws Exception {
     // synchronous connect
     client.connect().get();
-
-    // create a subscription and a monitored item
-    final UaSubscription subscription =
-        client.getSubscriptionManager().createSubscription(200.0).get();
-
-    final ReadValueId readValueId =
-        new ReadValueId(
-            Identifiers.Server, AttributeId.EventNotifier.uid(), null, QualifiedName.NULL_VALUE);
-
-    // client handle must be unique per item
-    final UInteger clientHandle = uint(clientHandles.getAndIncrement());
-
-    final EventFilter eventFilter =
-        new EventFilter(
-            new SimpleAttributeOperand[] {
-              new SimpleAttributeOperand(
-                  Identifiers.BaseEventType,
-                  new QualifiedName[] {new QualifiedName(0, "Time")},
-                  AttributeId.Value.uid(),
-                  null),
-              new SimpleAttributeOperand(
-                  Identifiers.BaseEventType,
-                  new QualifiedName[] {new QualifiedName(0, "Message")},
-                  AttributeId.Value.uid(),
-                  null),
-              new SimpleAttributeOperand(
-                  Identifiers.BaseEventType,
-                  new QualifiedName[] {new QualifiedName(0, "SourceName")},
-                  AttributeId.Value.uid(),
-                  null),
-              new SimpleAttributeOperand(
-                  Identifiers.BaseEventType,
-                  new QualifiedName[] {new QualifiedName(0, "SourceNode")},
-                  AttributeId.Value.uid(),
-                  null)
-            },
-            new ContentFilter(null));
-
-    final MonitoringParameters parameters =
-        new MonitoringParameters(
-            clientHandle,
-            0.0,
-            ExtensionObject.encode(client.getStaticSerializationContext(), eventFilter),
-            uint(10000),
-            true);
-
-    final MonitoredItemCreateRequest request =
-        new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, parameters);
-
-    final List<UaMonitoredItem> items =
-        subscription
-            .createMonitoredItems(TimestampsToReturn.Both, Collections.singletonList(request))
-            .get();
-
-    // do something with the value updates
-    final UaMonitoredItem monitoredItem = items.get(0);
-
-    final AtomicInteger eventCount = new AtomicInteger(0);
-
-    monitoredItem.setEventConsumer(
-        (item, vs) -> {
-          eventCount.incrementAndGet();
-          System.out.println("Event Received from " + item.getReadValueId().getNodeId());
-
-          for (int i = 0; i < vs.length; i++) {
-            System.out.println(("\tvariant[" + i + "]: " + vs[i].getValue()));
-          }
-        });
   }
 
   /////////////////////////////// Getter ///////////////////////////////