This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 94329e01692 [Pipe] Move OPC UA sink to pipe library (#18464)
94329e01692 is described below
commit 94329e01692f3c9d95830630147053e1c6250fa9
Author: Caideyipi <[email protected]>
AuthorDate: Mon Sep 21 10:25:56 2026 +0800
[Pipe] Move OPC UA sink to pipe library (#18464)
* Move OPC UA sink to pipe library
(cherry picked from commit a9007d652e8ceb2a9ea9813e03b18090bcd8d919)
* [Pipe] Fix OPC UA plugin CI failures
---
LICENSE-binary | 18 ++
distribution/pom.xml | 9 +
distribution/src/assembly/library-pipe.xml | 51 ++++
example/pom.xml | 1 -
integration-test/pom.xml | 7 +
.../it/env/cluster/config/MppConfigNodeConfig.java | 6 +
.../it/env/cluster/config/MppDataNodeConfig.java | 6 +
.../env/remote/config/RemoteConfigNodeConfig.java | 5 +
.../it/env/remote/config/RemoteDataNodeConfig.java | 5 +
.../apache/iotdb/itbase/env/ConfigNodeConfig.java | 2 +
.../apache/iotdb/itbase/env/DataNodeConfig.java | 2 +
.../iotdb/pipe/it/single/AbstractPipeSingleIT.java | 70 ++++-
.../iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java | 17 +-
iotdb-core/datanode/pom.xml | 24 --
.../dataregion/PipeDataRegionSinkConstructor.java | 3 -
.../db/pipe/sink/protocol/opcda/OpcDaSink.java | 9 +-
.../sink/util/PipeTabletEventTransferUtils.java | 97 +++++++
.../subtask/sink/PipeSinkSubtaskManagerTest.java | 3 +-
.../apache/iotdb/db/pipe/sink/PipeSinkTest.java | 93 -------
.../agent/plugin/builtin/BuiltinPipePlugin.java | 5 -
.../agent/plugin/builtin/sink/opcua/OpcUaSink.java | 34 ---
.../pipe/config/constant/PipeSinkConstant.java | 7 +-
.../opc-ua-sink-example}/pom.xml | 21 +-
.../java/org/apache/iotdb/opcua/ClientExample.java | 0
.../apache/iotdb/opcua/ClientExampleRunner.java | 0
.../java/org/apache/iotdb/opcua/ClientTest.java | 0
.../iotdb/opcua/IoTDBKeyStoreLoaderClient.java | 0
library-pipe/opc-ua-sink/README.md | 82 ++++++
library-pipe/opc-ua-sink/pom.xml | 166 ++++++++++++
.../iotdb/pipe/plugin/sink}/opcua/OpcUaSink.java | 91 ++-----
.../plugin/sink}/opcua/client/ClientRunner.java | 2 +-
.../opcua/client/IoTDBKeyStoreLoaderClient.java | 2 +-
.../sink}/opcua/client/IoTDBOpcUaClient.java | 10 +-
.../sink}/opcua/server/OpcUaKeyStoreLoader.java | 2 +-
.../plugin/sink}/opcua/server/OpcUaNameSpace.java | 4 +-
.../sink}/opcua/server/OpcUaServerBuilder.java | 2 +-
.../pipe/plugin/sink/opcua/OpcUaSinkTest.java | 285 +++++++++++++++++++++
.../sink}/opcua/OpcUaSinkTsFileMetadataTest.java | 2 +-
.../opcua/OpcUaSinkTsFilePerformanceTest.java | 6 +-
.../sink}/opcua/client/ClientRunnerTest.java | 2 +-
.../sink}/opcua/client/IoTDBOpcUaClientTest.java | 4 +-
.../opcua/server/OpcUaNameSpaceMetadataTest.java | 4 +-
.../sink}/opcua/server/OpcUaServerBuilderTest.java | 2 +-
.../server/OpcUaTcpNoneCompatibilityTest.java | 2 +-
library-pipe/pom.xml | 68 +++++
licenses/Bouncy Castle Licence | 17 ++
pom.xml | 27 +-
47 files changed, 979 insertions(+), 296 deletions(-)
diff --git a/LICENSE-binary b/LICENSE-binary
index 58c794bd802..9602245b0e2 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -218,6 +218,8 @@ Apache License 2.0
commons-cli:commons-cli:1.5.0
com.google.code.gson:gson:2.13.1
com.google.guava.guava:32.1.2-jre
+com.google.guava:failureaccess:1.0.3
+com.google.guava:guava:33.6.0-jre
com.fasterxml.jackson.core:jackson-annotations:2.16.2
com.fasterxml.jackson.core:jackson-core:2.16.2
com.fasterxml.jackson.core:jackson-databind:2.16.2
@@ -260,6 +262,8 @@ io.netty:netty-transport-native-unix-common:4.1.137.Final
org.osgi:org.osgi.core:7.0.0
org.osgi:osgi.cmpn:7.0.0
org.ops4j.pax.jdbc:pax-jdbc-common:1.5.6
+org.jspecify:jspecify:1.0.0
+org.jctools:jctools-core:4.0.5
org.xerial.snappy:snappy-java:1.1.10.5
io.airlift.airline:0.9
@@ -292,8 +296,22 @@ EPL 2.0
------------
ch.qos.logback:logback-classic:1.5.34
ch.qos.logback:logback-core:1.5.34
+com.digitalpetri.fsm:strict-machine:1.0.0
+com.digitalpetri.netty:netty-channel-fsm:1.0.2
jakarta.annotation:jakarta.annotation-api:3.0.0
jakarta.servlet:jakarta.servlet-api:6.0.0
+org.eclipse.milo:milo-guava-dependencies:1.1.6
+org.eclipse.milo:milo-sdk-client:1.1.6
+org.eclipse.milo:milo-sdk-core:1.1.6
+org.eclipse.milo:milo-sdk-server:1.1.6
+org.eclipse.milo:milo-stack-core:1.1.6
+org.eclipse.milo:milo-transport:1.1.6
+
+Bouncy Castle Licence
+------------
+org.bouncycastle:bcpkix-jdk18on:1.84
+org.bouncycastle:bcprov-jdk18on:1.84
+org.bouncycastle:bcutil-jdk18on:1.84
EDL 1.0
------------
diff --git a/distribution/pom.xml b/distribution/pom.xml
index 94d5ce68d99..147ac1fba06 100644
--- a/distribution/pom.xml
+++ b/distribution/pom.xml
@@ -33,6 +33,13 @@
<client.cpp.package.classifier>${os.classifier}</client.cpp.package.classifier>
</properties>
<dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>opc-ua-sink</artifactId>
+ <version>${project.version}</version>
+ <classifier>jar-with-dependencies</classifier>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-server</artifactId>
@@ -86,6 +93,7 @@
<descriptors>
<descriptor>src/assembly/all.xml</descriptor>
<descriptor>src/assembly/library-udf.xml</descriptor>
+
<descriptor>src/assembly/library-pipe.xml</descriptor>
<descriptor>src/assembly/edge.xml</descriptor>
</descriptors>
<finalName>apache-iotdb-${project.version}</finalName>
@@ -117,6 +125,7 @@
<includes>
<include>apache-iotdb-${project.version}-all-bin.zip</include>
<include>apache-iotdb-${project.version}-library-udf-bin.zip</include>
+
<include>apache-iotdb-${project.version}-library-pipe-bin.zip</include>
<include>apache-iotdb-${project.version}-external-service-impl-bin.zip</include>
<include>apache-iotdb-${project.version}-edge-bin.zip</include>
</includes>
diff --git a/distribution/src/assembly/library-pipe.xml
b/distribution/src/assembly/library-pipe.xml
new file mode 100644
index 00000000000..b10a33d8ea3
--- /dev/null
+++ b/distribution/src/assembly/library-pipe.xml
@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<assembly>
+ <id>library-pipe-bin</id>
+ <formats>
+ <format>dir</format>
+ <format>zip</format>
+ </formats>
+
<baseDirectory>apache-iotdb-${project.version}-library-pipe-bin</baseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>${project.basedir}/../licenses</directory>
+ <outputDirectory>licenses</outputDirectory>
+ </fileSet>
+ </fileSets>
+ <files>
+ <file>
+ <source>${project.basedir}/../LICENSE-binary</source>
+ <outputDirectory>licenses</outputDirectory>
+ <destName>LICENSE</destName>
+ </file>
+ <file>
+ <source>${project.basedir}/../NOTICE-binary</source>
+ <outputDirectory>licenses</outputDirectory>
+ <destName>NOTICE</destName>
+ </file>
+ <file>
+
<source>${maven.multiModuleProjectDirectory}/library-pipe/opc-ua-sink/target/opc-ua-sink-${project.version}-jar-with-dependencies.jar</source>
+ <outputDirectory>ext/pipe</outputDirectory>
+ </file>
+ </files>
+</assembly>
diff --git a/example/pom.xml b/example/pom.xml
index f7767ce6cb5..78054302f71 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -34,7 +34,6 @@
<module>mqtt</module>
<module>mqtt-customize</module>
<module>pipe-count-point-processor</module>
- <module>pipe-opc-ua-sink</module>
<module>rest-java-example</module>
<module>schema</module>
<module>session</module>
diff --git a/integration-test/pom.xml b/integration-test/pom.xml
index f6624fbd65d..960ca4ec2ab 100644
--- a/integration-test/pom.xml
+++ b/integration-test/pom.xml
@@ -257,6 +257,12 @@
<!--We will integrate rest-jar-with-dependencies into lib by
assembly plugin-->
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>opc-ua-sink</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
@@ -336,6 +342,7 @@
<systemPropertyVariables>
<EdgePackage>${integrationTest.edgePackage}</EdgePackage>
<TestEnv>${integrationTest.testEnv}</TestEnv>
+
<OpcUaSinkPluginJar>${maven.multiModuleProjectDirectory}/library-pipe/opc-ua-sink/target/opc-ua-sink-${project.version}-jar-with-dependencies.jar</OpcUaSinkPluginJar>
<RandomSelectWriteNode>${integrationTest.randomSelectWriteNode}</RandomSelectWriteNode>
<ReadAndVerifyWithMultiNode>${integrationTest.readAndVerifyWithMultiNode}</ReadAndVerifyWithMultiNode>
<DataRegionPerDataNode>${integrationTest.dataRegionPerDataNode}</DataRegionPerDataNode>
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
index 8da0f4e02b2..11a339df021 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
@@ -74,6 +74,12 @@ public class MppConfigNodeConfig extends MppBaseConfig
implements ConfigNodeConf
return this;
}
+ @Override
+ public ConfigNodeConfig setRatisLogAppenderBufferSizeMax(int size) {
+ properties.setProperty("config_node_ratis_log_appender_buffer_size_max",
String.valueOf(size));
+ return this;
+ }
+
@Override
public ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs) {
properties.setProperty("metadata_lease_fence_ms",
String.valueOf(metadataLeaseFenceMs));
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java
index f7dd625704c..90174941e54 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java
@@ -80,6 +80,12 @@ public class MppDataNodeConfig extends MppBaseConfig
implements DataNodeConfig {
return this;
}
+ @Override
+ public DataNodeConfig setThriftMaxFrameSize(int thriftMaxFrameSize) {
+ properties.setProperty("dn_thrift_max_frame_size",
String.valueOf(thriftMaxFrameSize));
+ return this;
+ }
+
@Override
public DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes(
long loadTsFileAnalyzeSchemaMemorySizeInBytes) {
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java
index 36fbdcbfc0e..e697b98effe 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java
@@ -44,6 +44,11 @@ public class RemoteConfigNodeConfig implements
ConfigNodeConfig {
return this;
}
+ @Override
+ public ConfigNodeConfig setRatisLogAppenderBufferSizeMax(int size) {
+ return this;
+ }
+
@Override
public ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs) {
return this;
diff --git
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java
index 13eeb343919..95e4b23b3e7 100644
---
a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java
@@ -48,6 +48,11 @@ public class RemoteDataNodeConfig implements DataNodeConfig {
return this;
}
+ @Override
+ public DataNodeConfig setThriftMaxFrameSize(int thriftMaxFrameSize) {
+ return this;
+ }
+
@Override
public DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes(
long loadTsFileAnalyzeSchemaMemorySizeInBytes) {
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java
index aec8f0f2335..a4260202a3a 100644
---
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java
@@ -32,5 +32,7 @@ public interface ConfigNodeConfig {
ConfigNodeConfig setLeaderDistributionPolicy(String policy);
+ ConfigNodeConfig setRatisLogAppenderBufferSizeMax(int size);
+
ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs);
}
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java
index d321053f278..6f65e18cdea 100644
---
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java
@@ -33,6 +33,8 @@ public interface DataNodeConfig {
DataNodeConfig setConnectionTimeoutInMS(int connectionTimeoutInMS);
+ DataNodeConfig setThriftMaxFrameSize(int thriftMaxFrameSize);
+
DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes(
long loadTsFileAnalyzeSchemaMemorySizeInBytes);
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java
index 61d4f0157e4..7cff83623be 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/AbstractPipeSingleIT.java
@@ -19,18 +19,33 @@
package org.apache.iotdb.pipe.it.single;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
+import org.apache.iotdb.db.it.utils.TestUtils;
import org.apache.iotdb.it.env.MultiEnvFactory;
import org.apache.iotdb.itbase.env.BaseEnv;
import org.junit.After;
import org.junit.Before;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+import java.util.stream.Stream;
+
abstract class AbstractPipeSingleIT {
+ private static final String OPC_UA_SINK_CLASS_NAME =
+ "org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink";
+ private static final String OPC_UA_SINK_JAR_PATH_PROPERTY =
"OpcUaSinkPluginJar";
+ private static final String OPC_UA_SINK_JAR_PREFIX = "opc-ua-sink-";
+ private static final String OPC_UA_SINK_JAR_SUFFIX =
"-jar-with-dependencies.jar";
+
protected BaseEnv env;
@Before
- public void setUp() {
+ public void setUp() throws Exception {
MultiEnvFactory.createEnv(1);
env = MultiEnvFactory.getEnv(0);
env.getConfig()
@@ -43,6 +58,59 @@ abstract class AbstractPipeSingleIT {
env.initClusterEnvironment();
}
+ protected final void registerOpcUaSinkPlugin() throws IOException {
+ final String pluginUri = locateOpcUaSinkPluginJar().toUri().toString();
+ registerOpcUaSinkPlugin(PipeSinkConstant.OPC_UA_SINK_NAME, pluginUri);
+ }
+
+ private void registerOpcUaSinkPlugin(final String pluginName, final String
pluginUri) {
+ TestUtils.executeNonQuery(
+ env,
+ String.format(
+ "CREATE PIPEPLUGIN IF NOT EXISTS `%s` AS '%s' USING URI '%s'",
+ pluginName, OPC_UA_SINK_CLASS_NAME, pluginUri));
+ }
+
+ private static Path locateOpcUaSinkPluginJar() throws IOException {
+ final String configuredPluginJar =
System.getProperty(OPC_UA_SINK_JAR_PATH_PROPERTY);
+ if (configuredPluginJar != null && !configuredPluginJar.isEmpty()) {
+ final Path pluginJar =
Paths.get(configuredPluginJar).toAbsolutePath().normalize();
+ if (Files.isRegularFile(pluginJar)) {
+ return pluginJar;
+ }
+ throw new IOException("Cannot locate the OPC UA sink plugin jar at " +
pluginJar + ".");
+ }
+
+ Path current =
Paths.get(System.getProperty("user.dir")).toAbsolutePath().normalize();
+ while (current != null) {
+ final Path targetDir =
+
current.resolve("library-pipe").resolve("opc-ua-sink").resolve("target");
+ if (Files.isDirectory(targetDir)) {
+ try (final Stream<Path> stream = Files.list(targetDir)) {
+ final List<Path> pluginJars =
+ stream
+ .filter(Files::isRegularFile)
+ .filter(path ->
path.getFileName().toString().startsWith(OPC_UA_SINK_JAR_PREFIX))
+ .filter(path ->
path.getFileName().toString().endsWith(OPC_UA_SINK_JAR_SUFFIX))
+ .toList();
+ if (pluginJars.size() == 1) {
+ return pluginJars.get(0);
+ }
+ if (pluginJars.size() > 1) {
+ throw new IOException(
+ "Multiple OPC UA sink plugin jars found in "
+ + targetDir
+ + ". Set -D"
+ + OPC_UA_SINK_JAR_PATH_PROPERTY
+ + " to the expected jar.");
+ }
+ }
+ }
+ current = current.getParent();
+ }
+ throw new IOException("Cannot locate the OPC UA sink plugin jar.");
+ }
+
@After
public final void tearDown() {
env.cleanClusterEnvironment();
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java
index 4054f7fae5f..b9ee7587358 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java
@@ -23,14 +23,14 @@ import
org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.db.it.utils.TestUtils;
-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.it.env.MultiEnvFactory;
import org.apache.iotdb.it.env.cluster.EnvUtils;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.MultiClusterIT1;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.pipe.it.dual.tablemodel.TableModelUtils;
+import org.apache.iotdb.pipe.plugin.sink.opcua.client.ClientRunner;
+import org.apache.iotdb.pipe.plugin.sink.opcua.client.IoTDBOpcUaClient;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.common.conf.TSFileConfig;
@@ -61,16 +61,24 @@ import java.util.Objects;
import java.util.UUID;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_DEFAULT_VALUE;
-import static
org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.timestampToUtc;
+import static
org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace.timestampToUtc;
@RunWith(IoTDBTestRunner.class)
@Category({MultiClusterIT1.class})
public class IoTDBPipeOPCUAIT extends AbstractPipeSingleIT {
+ private static final int PIPE_PLUGIN_JAR_TRANSFER_LIMIT_IN_BYTES = 32 * 1024
* 1024;
+
@Before
- public void setUp() {
+ public void setUp() throws Exception {
MultiEnvFactory.createEnv(1);
env = MultiEnvFactory.getEnv(0);
+ env.getConfig()
+ .getConfigNodeConfig()
+
.setRatisLogAppenderBufferSizeMax(PIPE_PLUGIN_JAR_TRANSFER_LIMIT_IN_BYTES);
+ env.getConfig()
+ .getDataNodeConfig()
+ .setThriftMaxFrameSize(PIPE_PLUGIN_JAR_TRANSFER_LIMIT_IN_BYTES);
env.getConfig()
.getCommonConfig()
.setAutoCreateSchemaEnabled(true)
@@ -81,6 +89,7 @@ public class IoTDBPipeOPCUAIT extends AbstractPipeSingleIT {
.setIsPipeEnableMemoryCheck(false)
.setPipeAutoSplitFullEnabled(false);
env.initClusterEnvironment(1, 1);
+ registerOpcUaSinkPlugin();
}
@Test
diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml
index b615061fc74..c3bc904f5c7 100644
--- a/iotdb-core/datanode/pom.xml
+++ b/iotdb-core/datanode/pom.xml
@@ -185,26 +185,6 @@
<groupId>net.java.dev.jna</groupId>
<artifactId>jna-platform</artifactId>
</dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-stack-core</artifactId>
- </dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-sdk-core</artifactId>
- </dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-transport</artifactId>
- </dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-sdk-client</artifactId>
- </dependency>
- <dependency>
- <groupId>org.bouncycastle</groupId>
- <artifactId>bcprov-jdk18on</artifactId>
- </dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
@@ -217,10 +197,6 @@
<groupId>io.airlift</groupId>
<artifactId>airline</artifactId>
</dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-sdk-server</artifactId>
- </dependency>
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java
index c8e87890afd..7044336f639 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java
@@ -27,7 +27,6 @@ import
org.apache.iotdb.db.pipe.sink.protocol.airgap.IoTDBDataRegionAirGapSink;
import
org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink;
import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink;
import org.apache.iotdb.db.pipe.sink.protocol.opcda.OpcDaSink;
-import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
import
org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
import
org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink;
import org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketSink;
@@ -68,7 +67,6 @@ class PipeDataRegionSinkConstructor extends
PipeSinkConstructor {
IoTDBDataRegionAirGapSink::new);
pluginConstructors.put(
BuiltinPipePlugin.WEBSOCKET_CONNECTOR.getPipePluginName(),
WebSocketSink::new);
-
pluginConstructors.put(BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(),
OpcUaSink::new);
pluginConstructors.put(BuiltinPipePlugin.OPC_DA_CONNECTOR.getPipePluginName(),
OpcDaSink::new);
pluginConstructors.put(
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(),
DoNothingSink::new);
@@ -90,7 +88,6 @@ class PipeDataRegionSinkConstructor extends
PipeSinkConstructor {
BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(),
IoTDBDataRegionAirGapSink::new);
pluginConstructors.put(
BuiltinPipePlugin.WEBSOCKET_SINK.getPipePluginName(),
WebSocketSink::new);
- pluginConstructors.put(BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(),
OpcUaSink::new);
pluginConstructors.put(BuiltinPipePlugin.OPC_DA_SINK.getPipePluginName(),
OpcDaSink::new);
pluginConstructors.put(
BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(),
DoNothingSink::new);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaSink.java
index d13e743bfc9..56025e3fdf9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaSink.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.pipe.sink.protocol.opcda;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
-import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
+import org.apache.iotdb.db.pipe.sink.util.PipeTabletEventTransferUtils;
import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.annotation.TreeModel;
import
org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
@@ -110,8 +110,11 @@ public class OpcDaSink implements PipeConnector {
@Override
public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws
Exception {
- OpcUaSink.transferByTablet(
- tabletInsertionEvent, LOGGER, (tablet, isTableModel) ->
handle.transfer(tablet));
+ PipeTabletEventTransferUtils.transferByTablet(
+ tabletInsertionEvent,
+ OpcDaSink.class.getName(),
+ LOGGER,
+ (tablet, isTableModel) -> handle.transfer(tablet));
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/PipeTabletEventTransferUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/PipeTabletEventTransferUtils.java
new file mode 100644
index 00000000000..719c839e56b
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/PipeTabletEventTransferUtils.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.sink.util;
+
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
+import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
+import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+
+import org.apache.tsfile.write.record.Tablet;
+import org.slf4j.Logger;
+
+public final class PipeTabletEventTransferUtils {
+
+ private PipeTabletEventTransferUtils() {}
+
+ public static void transferByTablet(
+ final TabletInsertionEvent tabletInsertionEvent,
+ final String holderMessage,
+ final Logger logger,
+ final ThrowingBiConsumer<Tablet, Boolean, Exception> transferTablet)
+ throws Exception {
+ // PipeProcessor can change the type of TabletInsertionEvent.
+ if (!(tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent)
+ && !(tabletInsertionEvent instanceof PipeRawTabletInsertionEvent)) {
+ logger.warn(
+ DataNodePipeMessages
+
.THIS_CONNECTOR_ONLY_SUPPORT_PIPEINSERTNODETABLETINSERTIONEVENT_AND_PIPERAWTABLET,
+ tabletInsertionEvent);
+ return;
+ }
+
+ if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
+ transferTabletWrapper(
+ (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent,
holderMessage, transferTablet);
+ } else {
+ transferTabletWrapper(
+ (PipeRawTabletInsertionEvent) tabletInsertionEvent, holderMessage,
transferTablet);
+ }
+ }
+
+ private static void transferTabletWrapper(
+ final PipeInsertNodeTabletInsertionEvent
pipeInsertNodeTabletInsertionEvent,
+ final String holderMessage,
+ final ThrowingBiConsumer<Tablet, Boolean, Exception> transferTablet)
+ throws Exception {
+ if
(!pipeInsertNodeTabletInsertionEvent.increaseReferenceCount(holderMessage)) {
+ return;
+ }
+ try {
+ for (final Tablet tablet :
pipeInsertNodeTabletInsertionEvent.convertToTablets()) {
+ transferTablet.accept(tablet,
pipeInsertNodeTabletInsertionEvent.isTableModelEvent());
+ }
+ } finally {
+ pipeInsertNodeTabletInsertionEvent.decreaseReferenceCount(holderMessage,
false);
+ }
+ }
+
+ private static void transferTabletWrapper(
+ final PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent,
+ final String holderMessage,
+ final ThrowingBiConsumer<Tablet, Boolean, Exception> transferTablet)
+ throws Exception {
+ if (!pipeRawTabletInsertionEvent.increaseReferenceCount(holderMessage)) {
+ return;
+ }
+ try {
+ transferTablet.accept(
+ pipeRawTabletInsertionEvent.convertToTablet(),
+ pipeRawTabletInsertionEvent.isTableModelEvent());
+ } finally {
+ pipeRawTabletInsertionEvent.decreaseReferenceCount(holderMessage, false);
+ }
+ }
+
+ @FunctionalInterface
+ public interface ThrowingBiConsumer<T, U, E extends Exception> {
+ void accept(final T t, final U u) throws E;
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java
index 8c2fd176587..ccd72ffc77a 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManagerTest.java
@@ -165,8 +165,7 @@ public class PipeSinkSubtaskManagerTest {
final Map<String, String> singleThreadAttributes = new HashMap<>();
singleThreadAttributes.put(
PipeSinkConstant.CONNECTOR_SERIALIZE_BY_REGION_KEY,
Boolean.FALSE.toString());
- singleThreadAttributes.put(
- PipeSinkConstant.CONNECTOR_KEY,
BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName());
+ singleThreadAttributes.put(PipeSinkConstant.CONNECTOR_KEY,
PipeSinkConstant.OPC_UA_SINK_NAME);
Assert.assertEquals(
1,
PipeSinkSubtaskManager.calculateSinkSubtaskNum(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
index 1f466e40ea6..ca7ac8d2ba0 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeSinkTest.java
@@ -30,7 +30,6 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import
org.apache.iotdb.db.pipe.event.common.statement.PipeStatementInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink;
-import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
import
org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
import
org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink;
import
org.apache.iotdb.db.pipe.sink.protocol.websocket.WebSocketConnectorServer;
@@ -50,7 +49,6 @@ import org.junit.Test;
import org.mockito.Mockito;
import java.lang.reflect.Field;
-import java.security.SecureRandom;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -446,97 +444,6 @@ public class PipeSinkTest {
}
}
- @Test
- public void testOpcUaSink() {
- final List<IMeasurementSchema> schemaList =
- Arrays.asList(
- new MeasurementSchema("s1", TSDataType.INT64),
- new MeasurementSchema("s2", TSDataType.INT64));
-
- final Tablet tablet = new Tablet("root.db.d1.vector6", schemaList, 100);
-
- long timestamp = System.currentTimeMillis();
- for (long row = 0; row < 100; row++) {
- final int rowSize = tablet.getRowSize();
- tablet.addTimestamp(rowSize, timestamp);
- for (int i = 0; i < 2; i++) {
- tablet.addValue(
- schemaList.get(i).getMeasurementName(), rowSize, new
SecureRandom().nextLong());
- }
- timestamp++;
- }
-
- final List<IMeasurementSchema> opcSchemaList =
- Arrays.asList(
- new MeasurementSchema("value1", TSDataType.INT64),
- new MeasurementSchema("quality1", TSDataType.BOOLEAN));
- final Tablet qualityTablet = new Tablet("root.db.d1.vector6.s3",
opcSchemaList, 100);
-
- timestamp = System.currentTimeMillis();
- for (long row = 0; row < 100; row++) {
- final int rowSize = qualityTablet.getRowSize();
- qualityTablet.addTimestamp(rowSize, timestamp);
- qualityTablet.addValue(
- opcSchemaList.get(0).getMeasurementName(), rowSize, new
SecureRandom().nextLong());
- qualityTablet.addValue(opcSchemaList.get(1).getMeasurementName(),
rowSize, true);
- timestamp++;
- }
-
- try (final OpcUaSink qualityOPC = new OpcUaSink();
- final OpcUaSink normalOPC = new OpcUaSink()) {
- final PipeTaskRuntimeConfiguration configuration =
- new PipeTaskRuntimeConfiguration(new
PipeTaskSinkRuntimeEnvironment("temp", 0, 1));
- qualityOPC.customize(
- new PipeParameters(
- new HashMap<String, String>() {
- {
- put(
- PipeSinkConstant.CONNECTOR_KEY,
- BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName());
- put(PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY,
"true");
- put(PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY,
"value1");
- put(PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY,
"quality1");
- }
- }),
- configuration);
- normalOPC.customize(
- new PipeParameters(
- new HashMap<String, String>() {
- {
- put(
- PipeSinkConstant.CONNECTOR_KEY,
- BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName());
- }
- }),
- configuration);
- final PipeRawTabletInsertionEvent event =
- new PipeRawTabletInsertionEvent(
- false, "root.db", "db", "root.db", tablet, false, "pipe", 0L,
null, null, false);
- event.increaseReferenceCount("");
- normalOPC.transfer(event);
- // Shall not throw
- qualityOPC.transfer(event);
- event.decreaseReferenceCount("", false);
-
- qualityOPC.transfer(
- new PipeRawTabletInsertionEvent(
- false,
- "root.db",
- "db",
- "root.db",
- qualityTablet,
- false,
- "pipe",
- 0L,
- null,
- null,
- false));
-
- } catch (Exception e) {
- Assert.fail();
- }
- }
-
@Test
public void testWriteBackSinkTargetDatabaseValidation() throws Exception {
assertWriteBackSinkTargetDatabaseValid("target");
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
index 7fd99dbc2aa..b9fa3cdef75 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
@@ -39,7 +39,6 @@ import
org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTD
import
org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftSslSink;
import
org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.iotdb.thrift.IoTDBThriftSyncSink;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.opcda.OpcDaSink;
-import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.opcua.OpcUaSink;
import
org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.websocket.WebSocketSink;
import
org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.writeback.WriteBackSink;
import
org.apache.iotdb.commons.pipe.agent.plugin.builtin.source.donothing.DoNothingSource;
@@ -92,7 +91,6 @@ public enum BuiltinPipePlugin {
PIPE_CONSENSUS_ASYNC_CONNECTOR("pipe-consensus-async-connector",
IoTConsensusV2AsyncSink.class),
WEBSOCKET_CONNECTOR("websocket-connector", WebSocketSink.class),
- OPC_UA_CONNECTOR("opc-ua-connector", OpcUaSink.class),
OPC_DA_CONNECTOR("opc-da-connector", OpcDaSink.class),
WRITE_BACK_CONNECTOR("write-back-connector", WriteBackSink.class),
@@ -104,7 +102,6 @@ public enum BuiltinPipePlugin {
IOTDB_LEGACY_PIPE_SINK("iotdb-legacy-pipe-sink", IoTDBLegacyPipeSink.class),
IOTDB_AIR_GAP_SINK("iotdb-air-gap-sink", IoTDBAirGapSink.class),
WEBSOCKET_SINK("websocket-sink", WebSocketSink.class),
- OPC_UA_SINK("opc-ua-sink", OpcUaSink.class),
OPC_DA_SINK("opc-da-sink", OpcDaSink.class),
WRITE_BACK_SINK("write-back-sink", WriteBackSink.class),
SUBSCRIPTION_SINK("subscription-sink", DoNothingSink.class),
@@ -208,7 +205,6 @@ public enum BuiltinPipePlugin {
IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName().toUpperCase(),
IOTDB_AIR_GAP_CONNECTOR.getPipePluginName().toUpperCase(),
WEBSOCKET_CONNECTOR.getPipePluginName().toUpperCase(),
- OPC_UA_CONNECTOR.getPipePluginName().toUpperCase(),
OPC_DA_CONNECTOR.getPipePluginName().toUpperCase(),
WRITE_BACK_CONNECTOR.getPipePluginName().toUpperCase(),
IOT_CONSENSUS_V2_ASYNC_CONNECTOR.getPipePluginName().toUpperCase(),
@@ -218,7 +214,6 @@ public enum BuiltinPipePlugin {
IOTDB_THRIFT_ASYNC_SINK.getPipePluginName().toUpperCase(),
IOTDB_LEGACY_PIPE_SINK.getPipePluginName().toUpperCase(),
WEBSOCKET_SINK.getPipePluginName().toUpperCase(),
- OPC_UA_SINK.getPipePluginName().toUpperCase(),
OPC_DA_SINK.getPipePluginName().toUpperCase(),
SUBSCRIPTION_SINK.getPipePluginName().toUpperCase(),
IOT_CONSENSUS_V2_ASYNC_SINK.getPipePluginName().toUpperCase(),
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/sink/opcua/OpcUaSink.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/sink/opcua/OpcUaSink.java
deleted file mode 100644
index 462fb099f71..00000000000
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/sink/opcua/OpcUaSink.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.opcua;
-
-import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.PlaceholderSink;
-import org.apache.iotdb.pipe.api.annotation.TableModel;
-import org.apache.iotdb.pipe.api.annotation.TreeModel;
-
-/**
- * This class is a placeholder and should not be initialized. It represents
the OPC UA connector.
- * There is a real implementation in the server module but cannot be imported
here. The pipe agent
- * in the server module will replace this class with the real implementation
when initializing the
- * OPC UA connector.
- */
-@TreeModel
-@TableModel
-public class OpcUaSink extends PlaceholderSink {}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
index f2d9ea283c8..f95ed6ce298 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
@@ -41,6 +41,9 @@ public class PipeSinkConstant {
public static final String CONNECTOR_KEY = "connector";
public static final String SINK_KEY = "sink";
+ public static final String OPC_UA_CONNECTOR_NAME = "opc-ua-connector";
+ public static final String OPC_UA_SINK_NAME = "opc-ua-sink";
+
public static final String CONNECTOR_IOTDB_IP_KEY = "connector.ip";
public static final String SINK_IOTDB_IP_KEY = "sink.ip";
public static final String CONNECTOR_IOTDB_HOST_KEY = "connector.host";
@@ -71,8 +74,8 @@ public class PipeSinkConstant {
public static final Set<String> SINGLE_THREAD_DEFAULT_SINK =
new HashSet<>(
Arrays.asList(
- BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(),
- BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(),
+ OPC_UA_SINK_NAME,
+ OPC_UA_CONNECTOR_NAME,
BuiltinPipePlugin.OPC_DA_SINK.getPipePluginName(),
BuiltinPipePlugin.OPC_DA_CONNECTOR.getPipePluginName()));
diff --git a/example/pipe-opc-ua-sink/pom.xml
b/library-pipe/opc-ua-sink-example/pom.xml
similarity index 82%
rename from example/pipe-opc-ua-sink/pom.xml
rename to library-pipe/opc-ua-sink-example/pom.xml
index 0627a854096..c0543538f58 100644
--- a/example/pipe-opc-ua-sink/pom.xml
+++ b/library-pipe/opc-ua-sink-example/pom.xml
@@ -22,22 +22,37 @@
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<parent>
<groupId>org.apache.iotdb</groupId>
- <artifactId>iotdb-examples</artifactId>
+ <artifactId>library-pipe</artifactId>
<version>2.0.11-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pipe-opc-ua-sink-example</artifactId>
- <name>IoTDB: Example: Pipe: OPCUA Sink</name>
+ <name>IoTDB: Library-Pipe: OPC UA Sink Example</name>
<dependencies>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-stack-core</artifactId>
+ </dependency>
<dependency>
<groupId>org.eclipse.milo</groupId>
<artifactId>milo-sdk-client</artifactId>
- <version>${milo.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-transport</artifactId>
</dependency>
<dependency>
<groupId>org.eclipse.milo</groupId>
<artifactId>milo-sdk-server</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcprov-jdk18on</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
</dependencies>
<profiles>
<profile>
diff --git
a/example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExample.java
b/library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientExample.java
similarity index 100%
rename from
example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExample.java
rename to
library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientExample.java
diff --git
a/example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java
b/library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java
similarity index 100%
rename from
example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java
rename to
library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientExampleRunner.java
diff --git
a/example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientTest.java
b/library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientTest.java
similarity index 100%
rename from
example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/ClientTest.java
rename to
library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/ClientTest.java
diff --git
a/example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java
b/library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java
similarity index 100%
rename from
example/pipe-opc-ua-sink/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java
rename to
library-pipe/opc-ua-sink-example/src/main/java/org/apache/iotdb/opcua/IoTDBKeyStoreLoaderClient.java
diff --git a/library-pipe/opc-ua-sink/README.md
b/library-pipe/opc-ua-sink/README.md
new file mode 100644
index 00000000000..90c260a5ac6
--- /dev/null
+++ b/library-pipe/opc-ua-sink/README.md
@@ -0,0 +1,82 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+
+# OPC UA Sink Plugin
+
+`opc-ua-sink` is an external Pipe Sink plugin that publishes IoTDB data
through OPC UA or writes it to an external OPC UA server. Eclipse Milo and the
plugin's other runtime dependencies are bundled in its fat JAR instead of the
default IoTDB server package.
+
+## Build
+
+Run from the repository root:
+
+```bash
+mvn clean package -pl library-pipe/opc-ua-sink -am -DskipTests
+```
+
+The plugin artifact is generated at:
+
+`library-pipe/opc-ua-sink/target/opc-ua-sink-<version>-jar-with-dependencies.jar`
+
+It is also included under `ext/pipe` in the standalone `library-pipe`
distribution ZIP.
+
+## Register
+
+The fat JAR is larger than the default 16 MiB ConfigNode consensus limit.
Before starting the
+nodes, set the following properties in `iotdb-system.properties` so the plugin
can be transferred
+and replicated:
+
+```properties
+config_node_ratis_log_appender_buffer_size_max=33554432
+dn_thrift_max_frame_size=33554432
+```
+
+Upload the fat JAR to a URI accessible to IoTDB and register the plugin before
creating an OPC UA Pipe:
+
+```sql
+CREATE PIPEPLUGIN `opc-ua-sink`
+AS 'org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink'
+USING URI 'file:///path/to/opc-ua-sink-<version>-jar-with-dependencies.jar';
+```
+
+`opc-ua-connector` is the legacy-compatible alias. Register it separately when
existing Pipe definitions use that name:
+
+```sql
+CREATE PIPEPLUGIN `opc-ua-connector`
+AS 'org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink'
+USING URI 'file:///path/to/opc-ua-sink-<version>-jar-with-dependencies.jar';
+```
+
+## Example
+
+The following Pipe starts an OPC UA server with no message security:
+
+```sql
+CREATE PIPE opc_ua_pipe
+WITH SINK (
+ 'sink' = 'opc-ua-sink',
+ 'sink.opcua.model' = 'client-server',
+ 'sink.opcua.security-policy' = 'NONE',
+ 'sink.opcua.tcp.port' = '12686',
+ 'sink.opcua.https.port' = '8443'
+);
+```
+
+The client example is located in `library-pipe/opc-ua-sink-example`.
diff --git a/library-pipe/opc-ua-sink/pom.xml b/library-pipe/opc-ua-sink/pom.xml
new file mode 100644
index 00000000000..314de457e04
--- /dev/null
+++ b/library-pipe/opc-ua-sink/pom.xml
@@ -0,0 +1,166 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>library-pipe</artifactId>
+ <version>2.0.11-SNAPSHOT</version>
+ </parent>
+ <artifactId>opc-ua-sink</artifactId>
+ <name>IoTDB: Library-Pipe: OPC UA Sink</name>
+ <properties>
+ <maven.compiler.source>17</maven.compiler.source>
+ <maven.compiler.target>17</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>node-commons</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>pipe-api</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tsfile</groupId>
+ <artifactId>tsfile</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tsfile</groupId>
+ <artifactId>common</artifactId>
+ <version>${tsfile.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-stack-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-sdk-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-transport</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-sdk-client</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-sdk-server</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcprov-jdk18on</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemProperties>
+
<IOTDB_CONF>${maven.multiModuleProjectDirectory}/iotdb-core/datanode/src/test/resources</IOTDB_CONF>
+ </systemProperties>
+ <reuseForks>false</reuseForks>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ <appendAssemblyId>true</appendAssemblyId>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <configuration>
+ <ignoredDependencies>
+ <!-- The tsfile artifact contains the same common
classes. -->
+
<ignoredDependency>org.apache.tsfile:common</ignoredDependency>
+ </ignoredDependencies>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSink.java
similarity index 91%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
rename to
library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSink.java
index af00b88ffa6..baa46a0ab78 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
+++
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSink.java
@@ -17,19 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua;
+package org.apache.iotdb.pipe.plugin.sink.opcua;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.utils.PathUtils;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
-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.event.common.tsfile.PipeTsFileInsertionEvent;
-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;
+import org.apache.iotdb.db.pipe.sink.util.PipeTabletEventTransferUtils;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.pipe.api.PipeConnector;
@@ -42,6 +37,10 @@ import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.pipe.plugin.sink.opcua.client.ClientRunner;
+import org.apache.iotdb.pipe.plugin.sink.opcua.client.IoTDBOpcUaClient;
+import org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace;
+import org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaServerBuilder;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.common.constant.TsFileConstant;
@@ -52,7 +51,6 @@ import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.TsFileSequenceReader;
import org.apache.tsfile.read.reader.TsFileLastReader;
import org.apache.tsfile.utils.Pair;
-import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider;
@@ -178,9 +176,11 @@ public class OpcUaSink implements PipeConnector {
// Inner server
private @Nullable OpcUaNameSpace nameSpace;
+ private boolean serverRegistered;
// Outer server
private @Nullable IoTDBOpcUaClient client;
+ private boolean clientRegistered;
@Override
public void validate(final PipeParameterValidator validator) throws
Exception {
@@ -387,6 +387,7 @@ public class OpcUaSink implements PipeConnector {
})
.getRight();
SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP.get(serverKey).getLeft().incrementAndGet();
+ serverRegistered = true;
}
}
@@ -467,6 +468,7 @@ public class OpcUaSink implements PipeConnector {
})
.getRight();
CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.get(nodeUrl).getLeft().incrementAndGet();
+ clientRegistered = true;
}
}
@@ -669,8 +671,9 @@ public class OpcUaSink implements PipeConnector {
@Override
public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws
Exception {
- transferByTablet(
+ PipeTabletEventTransferUtils.transferByTablet(
tabletInsertionEvent,
+ OpcUaSink.class.getName(),
LOGGER,
(tablet, isTableModel) -> {
if (Objects.nonNull(nameSpace)) {
@@ -683,75 +686,14 @@ public class OpcUaSink implements PipeConnector {
});
}
- public static void transferByTablet(
- final TabletInsertionEvent tabletInsertionEvent,
- final Logger logger,
- final ThrowingBiConsumer<Tablet, Boolean, Exception> transferTablet)
- throws Exception {
- // PipeProcessor can change the type of TabletInsertionEvent
- if (!(tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent)
- && !(tabletInsertionEvent instanceof PipeRawTabletInsertionEvent)) {
- logger.warn(
- DataNodePipeMessages
-
.THIS_CONNECTOR_ONLY_SUPPORT_PIPEINSERTNODETABLETINSERTIONEVENT_AND_PIPERAWTABLET,
- tabletInsertionEvent);
- return;
- }
-
- if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) {
- transferTabletWrapper(
- (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent,
transferTablet);
- } else {
- transferTabletWrapper((PipeRawTabletInsertionEvent)
tabletInsertionEvent, transferTablet);
- }
- }
-
- private static void transferTabletWrapper(
- final PipeInsertNodeTabletInsertionEvent
pipeInsertNodeTabletInsertionEvent,
- final ThrowingBiConsumer<Tablet, Boolean, Exception> transferTablet)
- throws Exception {
- // We increase the reference count for this event to determine if the
event may be released.
- if
(!pipeInsertNodeTabletInsertionEvent.increaseReferenceCount(OpcUaSink.class.getName()))
{
- return;
- }
- try {
- for (final Tablet tablet :
pipeInsertNodeTabletInsertionEvent.convertToTablets()) {
- transferTablet.accept(tablet,
pipeInsertNodeTabletInsertionEvent.isTableModelEvent());
- }
- } finally {
-
pipeInsertNodeTabletInsertionEvent.decreaseReferenceCount(OpcUaSink.class.getName(),
false);
- }
- }
-
- private static void transferTabletWrapper(
- final PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent,
- final ThrowingBiConsumer<Tablet, Boolean, Exception> transferTablet)
- throws Exception {
- // We increase the reference count for this event to determine if the
event may be released.
- if
(!pipeRawTabletInsertionEvent.increaseReferenceCount(OpcUaSink.class.getName()))
{
- return;
- }
- try {
- transferTablet.accept(
- pipeRawTabletInsertionEvent.convertToTablet(),
- pipeRawTabletInsertionEvent.isTableModelEvent());
- } finally {
-
pipeRawTabletInsertionEvent.decreaseReferenceCount(OpcUaSink.class.getName(),
false);
- }
- }
-
- @FunctionalInterface
- public interface ThrowingBiConsumer<T, U, E extends Exception> {
- void accept(final T t, final U u) throws E;
- }
-
@Override
public void close() throws Exception {
- if (serverKey != null) {
+ if (serverKey != null && serverRegistered) {
synchronized (SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP) {
final Pair<AtomicInteger, OpcUaNameSpace> pair =
SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP.get(serverKey);
if (pair == null) {
+ serverRegistered = false;
return;
}
@@ -762,14 +704,16 @@ public class OpcUaSink implements PipeConnector {
SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP.remove(serverKey);
}
}
+ serverRegistered = false;
}
}
- if (nodeUrl != null) {
+ if (nodeUrl != null && clientRegistered) {
synchronized (CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP) {
final Pair<AtomicInteger, IoTDBOpcUaClient> pair =
CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.get(nodeUrl);
if (pair == null) {
+ clientRegistered = false;
return;
}
@@ -780,6 +724,7 @@ public class OpcUaSink implements PipeConnector {
CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.remove(nodeUrl);
}
}
+ clientRegistered = false;
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunner.java
similarity index 99%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java
rename to
library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunner.java
index 64757596d16..4854d02ce30 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java
+++
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunner.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.client;
+package org.apache.iotdb.pipe.plugin.sink.opcua.client;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.pipe.api.exception.PipeException;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBKeyStoreLoaderClient.java
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBKeyStoreLoaderClient.java
similarity index 98%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBKeyStoreLoaderClient.java
rename to
library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBKeyStoreLoaderClient.java
index 455018695dc..c522f8e7b83 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBKeyStoreLoaderClient.java
+++
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBKeyStoreLoaderClient.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.client;
+package org.apache.iotdb.pipe.plugin.sink.opcua.client;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClient.java
similarity index 98%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
rename to
library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClient.java
index f00ec845047..e898db28508 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
+++
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClient.java
@@ -17,14 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.client;
+package org.apache.iotdb.pipe.plugin.sink.opcua.client;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
-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.iotdb.pipe.plugin.sink.opcua.OpcUaSink;
+import org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace;
import org.apache.tsfile.common.constant.TsFileConstant;
import org.apache.tsfile.enums.TSDataType;
@@ -72,8 +72,8 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutionException;
-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;
+import static
org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace.convertToOpcDataType;
+import static
org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace.timestampToUtc;
import static org.eclipse.milo.opcua.stack.core.StatusCodes.Bad_Timeout;
import static
org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn.Neither;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaKeyStoreLoader.java
similarity index 98%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java
rename to
library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaKeyStoreLoader.java
index 2e27ffc2277..838624403d1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java
+++
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaKeyStoreLoader.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.server;
+package org.apache.iotdb.pipe.plugin.sink.opcua.server;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpace.java
similarity index 99%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
rename to
library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpace.java
index 421ba27f77f..c6649a3222b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
+++
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpace.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.server;
+package org.apache.iotdb.pipe.plugin.sink.opcua.server;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
@@ -25,10 +25,10 @@ import
org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.commons.queryengine.utils.DateTimeUtils;
import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
-import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
import
org.apache.iotdb.db.pipe.sink.util.sorter.PipeTableModelTabletEventSorter;
import
org.apache.iotdb.db.pipe.sink.util.sorter.PipeTreeModelTabletEventSorter;
import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink;
import org.apache.tsfile.common.constant.TsFileConstant;
import org.apache.tsfile.enums.ColumnCategory;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilder.java
similarity index 99%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java
rename to
library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilder.java
index 5b7d820f8f7..1ed4038775b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilder.java
+++
b/library-pipe/opc-ua-sink/src/main/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilder.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.server;
+package org.apache.iotdb.pipe.plugin.sink.opcua.server;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.pipe.api.exception.PipeException;
diff --git
a/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTest.java
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTest.java
new file mode 100644
index 00000000000..4d4a8a8c7e1
--- /dev/null
+++
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTest.java
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.pipe.plugin.sink.opcua;
+
+import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
+import
org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
+import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnvironment;
+import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.security.SecureRandom;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class OpcUaSinkTest {
+
+ @Test
+ public void testValidateClientServerOnlyOptions() {
+ assertValidationFailure(
+ "must be client-server",
+ PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY,
+ "true",
+ PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_KEY,
+ PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_PUB_SUB_VALUE);
+
+ assertValidationFailure(
+ "must be client-server",
+ PipeSinkConstant.CONNECTOR_OPC_UA_NODE_URL_KEY,
+ "opc.tcp://127.0.0.1:12686/iotdb",
+ PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_KEY,
+ PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_PUB_SUB_VALUE);
+ }
+
+ @Test
+ public void testCustomizeQualityParameters() throws Exception {
+ final int[] qualityPorts = findTwoFreePorts();
+ final int[] normalPorts = findTwoFreePorts();
+ try (final OpcUaSink qualitySink = new OpcUaSink();
+ final OpcUaSink normalSink = new OpcUaSink()) {
+ qualitySink.customize(
+ createParameters(
+ PipeSinkConstant.CONNECTOR_KEY,
+ PipeSinkConstant.OPC_UA_SINK_NAME,
+ PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY,
+ "true",
+ PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY,
+ "value1",
+ PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY,
+ "quality1",
+ PipeSinkConstant.CONNECTOR_OPC_UA_DEFAULT_QUALITY_KEY,
+ "BAD",
+ PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_POLICY_KEY,
+ "None",
+ PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY,
+ Integer.toString(qualityPorts[0]),
+ PipeSinkConstant.CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY,
+ Integer.toString(qualityPorts[1]),
+ PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_KEY,
+ createSecurityDir()),
+ createRuntimeConfiguration());
+ normalSink.customize(
+ createOpcUaServerParameters(normalPorts[0], normalPorts[1], "root",
"root"),
+ createRuntimeConfiguration());
+
+ Assert.assertEquals("value1", qualitySink.getValueName());
+ Assert.assertEquals("quality1", qualitySink.getQualityName());
+ Assert.assertEquals(StatusCode.BAD, qualitySink.getDefaultQuality());
+ Assert.assertNull(normalSink.getValueName());
+ Assert.assertNull(normalSink.getQualityName());
+ Assert.assertEquals(StatusCode.GOOD, normalSink.getDefaultQuality());
+ }
+ }
+
+ @Test
+ public void testTransferWithQualityAndNormalTablets() throws Exception {
+ final List<IMeasurementSchema> schemaList =
+ Arrays.asList(
+ new MeasurementSchema("s1", TSDataType.INT64),
+ new MeasurementSchema("s2", TSDataType.INT64));
+ final Tablet tablet = new Tablet("root.db.d1.vector6", schemaList, 100);
+ long timestamp = System.currentTimeMillis();
+ for (long row = 0; row < 100; row++) {
+ final int rowSize = tablet.getRowSize();
+ tablet.addTimestamp(rowSize, timestamp++);
+ for (int i = 0; i < 2; i++) {
+ tablet.addValue(
+ schemaList.get(i).getMeasurementName(), rowSize, new
SecureRandom().nextLong());
+ }
+ }
+
+ final List<IMeasurementSchema> opcSchemaList =
+ Arrays.asList(
+ new MeasurementSchema("value1", TSDataType.INT64),
+ new MeasurementSchema("quality1", TSDataType.BOOLEAN));
+ final Tablet qualityTablet = new Tablet("root.db.d1.vector6.s3",
opcSchemaList, 100);
+ timestamp = System.currentTimeMillis();
+ for (long row = 0; row < 100; row++) {
+ final int rowSize = qualityTablet.getRowSize();
+ qualityTablet.addTimestamp(rowSize, timestamp++);
+ qualityTablet.addValue(
+ opcSchemaList.get(0).getMeasurementName(), rowSize, new
SecureRandom().nextLong());
+ qualityTablet.addValue(opcSchemaList.get(1).getMeasurementName(),
rowSize, true);
+ }
+
+ final int[] qualityPorts = findTwoFreePorts();
+ final int[] normalPorts = findTwoFreePorts();
+ try (final OpcUaSink qualitySink = new OpcUaSink();
+ final OpcUaSink normalSink = new OpcUaSink()) {
+ qualitySink.customize(
+ createQualityServerParameters(qualityPorts[0], qualityPorts[1]),
+ createRuntimeConfiguration());
+ normalSink.customize(
+ createOpcUaServerParameters(normalPorts[0], normalPorts[1], "root",
"root"),
+ createRuntimeConfiguration());
+
+ final PipeRawTabletInsertionEvent event =
+ new PipeRawTabletInsertionEvent(
+ false, "root.db", "db", "root.db", tablet, false, "pipe", 0L,
null, null, false);
+ event.increaseReferenceCount("");
+ normalSink.transfer(event);
+ qualitySink.transfer(event);
+ event.decreaseReferenceCount("", false);
+
+ qualitySink.transfer(
+ new PipeRawTabletInsertionEvent(
+ false,
+ "root.db",
+ "db",
+ "root.db",
+ qualityTablet,
+ false,
+ "pipe",
+ 0L,
+ null,
+ null,
+ false));
+ }
+ }
+
+ @Test
+ public void testSharedServerLifecycle() throws Exception {
+ final int[] ports = findTwoFreePorts();
+ final PipeTaskRuntimeConfiguration configuration =
createRuntimeConfiguration();
+ final PipeParameters parameters =
+ createOpcUaServerParameters(ports[0], ports[1], "root", "root");
+ final PipeParameters conflictingParameters =
+ createOpcUaServerParameters(ports[0], ports[1], "root", "conflict");
+
+ try (final OpcUaSink firstSink = new OpcUaSink();
+ final OpcUaSink secondSink = new OpcUaSink()) {
+ firstSink.customize(parameters, configuration);
+ secondSink.customize(parameters, configuration);
+
+ assertCustomizeFailure(conflictingParameters, configuration);
+ secondSink.close();
+ secondSink.close();
+ assertCustomizeFailure(conflictingParameters, configuration);
+ }
+ }
+
+ private static void assertCustomizeFailure(
+ final PipeParameters parameters, final PipeTaskRuntimeConfiguration
configuration) {
+ try (final OpcUaSink conflictingSink = new OpcUaSink()) {
+ final PipeException exception =
+ Assert.assertThrows(
+ PipeException.class, () -> conflictingSink.customize(parameters,
configuration));
+ Assert.assertTrue(exception.getMessage(),
exception.getMessage().contains("reject reusing"));
+ } catch (final Exception e) {
+ throw new AssertionError(e);
+ }
+ }
+
+ private static void assertValidationFailure(
+ final String expectedMessagePart, final String... keyValues) {
+ final PipeParameterNotValidException exception =
+ Assert.assertThrows(
+ PipeParameterNotValidException.class,
+ () ->
+ new OpcUaSink().validate(new
PipeParameterValidator(createParameters(keyValues))));
+ Assert.assertTrue(exception.getMessage(),
exception.getMessage().contains(expectedMessagePart));
+ }
+
+ private static PipeParameters createQualityServerParameters(
+ final int tcpPort, final int httpsPort) {
+ final Map<String, String> attributes =
+ new HashMap<>(
+ createOpcUaServerParameters(tcpPort, httpsPort, "root",
"root").getAttribute());
+ attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_KEY, "true");
+ attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY, "value1");
+ attributes.put(PipeSinkConstant.CONNECTOR_OPC_UA_QUALITY_NAME_KEY,
"quality1");
+ return new PipeParameters(attributes);
+ }
+
+ private static PipeParameters createOpcUaServerParameters(
+ final int tcpPort, final int httpsPort, final String user, final String
password) {
+ return createParameters(
+ PipeSinkConstant.CONNECTOR_KEY,
+ PipeSinkConstant.OPC_UA_SINK_NAME,
+ PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY,
+ Integer.toString(tcpPort),
+ PipeSinkConstant.CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY,
+ Integer.toString(httpsPort),
+ PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_POLICY_KEY,
+ "None",
+ PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY,
+ user,
+ PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY,
+ password,
+ PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_DIR_KEY,
+ createSecurityDir(),
+ PipeSinkConstant.CONNECTOR_OPC_UA_DEBOUNCE_TIME_MS_KEY,
+ "1");
+ }
+
+ private static PipeParameters createParameters(final String... keyValues) {
+ final Map<String, String> attributes = new HashMap<>();
+ for (int i = 0; i < keyValues.length; i += 2) {
+ attributes.put(keyValues[i], keyValues[i + 1]);
+ }
+ return new PipeParameters(attributes);
+ }
+
+ private static PipeTaskRuntimeConfiguration createRuntimeConfiguration() {
+ return new PipeTaskRuntimeConfiguration(new
PipeTaskSinkRuntimeEnvironment("temp", 0, 1));
+ }
+
+ private static String createSecurityDir() {
+ return new File(
+ "target"
+ + File.separatorChar
+ + "opc-ua-sink-test"
+ + File.separatorChar
+ + UUID.randomUUID())
+ .getAbsolutePath();
+ }
+
+ private static int[] findTwoFreePorts() throws IOException {
+ final int firstPort = findFreePort();
+ int secondPort;
+ do {
+ secondPort = findFreePort();
+ } while (secondPort == firstPort);
+ return new int[] {firstPort, secondPort};
+ }
+
+ private static int findFreePort() throws IOException {
+ try (final ServerSocket socket = new ServerSocket(0)) {
+ return socket.getLocalPort();
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFileMetadataTest.java
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTsFileMetadataTest.java
similarity index 99%
rename from
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFileMetadataTest.java
rename to
library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTsFileMetadataTest.java
index ad26debe416..541d74576c1 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFileMetadataTest.java
+++
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTsFileMetadataTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua;
+package org.apache.iotdb.pipe.plugin.sink.opcua;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.common.constant.TsFileConstant;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFilePerformanceTest.java
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTsFilePerformanceTest.java
similarity index 98%
rename from
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFilePerformanceTest.java
rename to
library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTsFilePerformanceTest.java
index 59fdd76f73a..2a6fdbc5cf4 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSinkTsFilePerformanceTest.java
+++
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/OpcUaSinkTsFilePerformanceTest.java
@@ -16,18 +16,18 @@
* limitations under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua;
+package org.apache.iotdb.pipe.plugin.sink.opcua;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.scan.TsFileInsertionEventScanParser;
-import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace;
-import
org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.TabletRowConsumer;
import org.apache.iotdb.db.utils.ManualPerformanceTestUtils;
import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement;
import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary;
+import org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace;
+import
org.apache.iotdb.pipe.plugin.sink.opcua.server.OpcUaNameSpace.TabletRowConsumer;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunnerTest.java
similarity index 98%
rename from
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java
rename to
library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunnerTest.java
index 9ccffbfc450..5f9c027866e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunnerTest.java
+++
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/ClientRunnerTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.client;
+package org.apache.iotdb.pipe.plugin.sink.opcua.client;
import org.apache.iotdb.pipe.api.exception.PipeException;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClientTest.java
similarity index 99%
rename from
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java
rename to
library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClientTest.java
index d327155f32d..102974d415a 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClientTest.java
+++
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/client/IoTDBOpcUaClientTest.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.client;
+package org.apache.iotdb.pipe.plugin.sink.opcua.client;
-import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpaceMetadataTest.java
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpaceMetadataTest.java
similarity index 98%
rename from
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpaceMetadataTest.java
rename to
library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpaceMetadataTest.java
index a7410aea71a..cb5687483d8 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpaceMetadataTest.java
+++
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaNameSpaceMetadataTest.java
@@ -16,9 +16,9 @@
* limitations under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.server;
+package org.apache.iotdb.pipe.plugin.sink.opcua.server;
-import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
+import org.apache.iotdb.pipe.plugin.sink.opcua.OpcUaSink;
import org.apache.tsfile.common.constant.TsFileConstant;
import org.apache.tsfile.enums.TSDataType;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilderTest.java
similarity index 99%
rename from
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java
rename to
library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilderTest.java
index 8c21c6632ce..1942f86515e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaServerBuilderTest.java
+++
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaServerBuilderTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.server;
+package org.apache.iotdb.pipe.plugin.sink.opcua.server;
import org.apache.iotdb.pipe.api.exception.PipeException;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaTcpNoneCompatibilityTest.java
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaTcpNoneCompatibilityTest.java
similarity index 98%
rename from
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaTcpNoneCompatibilityTest.java
rename to
library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaTcpNoneCompatibilityTest.java
index fabee999e2e..a199447774c 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaTcpNoneCompatibilityTest.java
+++
b/library-pipe/opc-ua-sink/src/test/java/org/apache/iotdb/pipe/plugin/sink/opcua/server/OpcUaTcpNoneCompatibilityTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.pipe.sink.protocol.opcua.server;
+package org.apache.iotdb.pipe.plugin.sink.opcua.server;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider;
diff --git a/library-pipe/pom.xml b/library-pipe/pom.xml
new file mode 100644
index 00000000000..3f4749cb9d1
--- /dev/null
+++ b/library-pipe/pom.xml
@@ -0,0 +1,68 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-parent</artifactId>
+ <version>2.0.11-SNAPSHOT</version>
+ </parent>
+ <artifactId>library-pipe</artifactId>
+ <name>IoTDB: Library-Pipe</name>
+ <packaging>pom</packaging>
+ <properties>
+ <milo.version>1.1.6</milo.version>
+ </properties>
+ <modules>
+ <module>opc-ua-sink</module>
+ <module>opc-ua-sink-example</module>
+ </modules>
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-stack-core</artifactId>
+ <version>${milo.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-sdk-core</artifactId>
+ <version>${milo.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-transport</artifactId>
+ <version>${milo.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-sdk-client</artifactId>
+ <version>${milo.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.milo</groupId>
+ <artifactId>milo-sdk-server</artifactId>
+ <version>${milo.version}</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+</project>
diff --git a/licenses/Bouncy Castle Licence b/licenses/Bouncy Castle Licence
new file mode 100644
index 00000000000..24506f04ccd
--- /dev/null
+++ b/licenses/Bouncy Castle Licence
@@ -0,0 +1,17 @@
+Copyright (c) 2000-2023 The Legion of the Bouncy Castle Inc.
(https://www.bouncycastle.org)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
of this software
+and associated documentation files (the "Software"), to deal in the Software
without restriction,
+including without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense,
+and/or sell copies of the Software, and to permit persons to whom the Software
is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
copies or substantial
+portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED,
+INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
PARTICULAR
+PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
CONTRACT, TORT OR
+OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
OR OTHER
+DEALINGS IN THE SOFTWARE.
diff --git a/pom.xml b/pom.xml
index 468a9760c6a..84efc0898ef 100644
--- a/pom.xml
+++ b/pom.xml
@@ -44,6 +44,7 @@
<module>iotdb-client</module>
<module>iotdb-core</module>
<module>iotdb-protocol</module>
+ <module>library-pipe</module>
<module>distribution</module>
<module>example</module>
<module>library-udf</module>
@@ -109,7 +110,6 @@
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<micrometer.version>1.11.4</micrometer.version>
- <milo.version>1.1.6</milo.version>
<!-- It seems that powermock is having issues with the newest mockito
versions -->
<mockito.version>2.23.4</mockito.version>
<!--mockito.version>4.11.0</mockito.version-->
@@ -371,26 +371,6 @@
<artifactId>units</artifactId>
<version>${airlift-units.version}</version>
</dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-stack-core</artifactId>
- <version>${milo.version}</version>
- </dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-sdk-core</artifactId>
- <version>${milo.version}</version>
- </dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-transport</artifactId>
- <version>${milo.version}</version>
- </dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-sdk-client</artifactId>
- <version>${milo.version}</version>
- </dependency>
<!-- TODO: Deprecated: Use Airline 2 or Picocli instead -->
<dependency>
<groupId>io.airlift</groupId>
@@ -403,11 +383,6 @@
</exclusion>
</exclusions>
</dependency>
- <dependency>
- <groupId>org.eclipse.milo</groupId>
- <artifactId>milo-sdk-server</artifactId>
- <version>${milo.version}</version>
- </dependency>
<dependency>
<groupId>org.reflections</groupId>
<artifactId>reflections</artifactId>
