You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/04/21 09:33:31 UTC
(iotdb) branch opc-fix-master created (now d99577b54c0)
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a change to branch opc-fix-master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at d99577b54c0 fix
This branch includes the following new commits:
new 5f934d44ccf Update OpcUaNameSpace.java
new d99577b54c0 fix
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/02: Update OpcUaNameSpace.java
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch opc-fix-master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5f934d44ccfdbbbbdbe8c5f0a60aa5c8c2d277c0
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue Apr 21 14:13:49 2026 +0800
Update OpcUaNameSpace.java
---
.../apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
index d1182e59b52..da4480458c5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
@@ -90,7 +90,7 @@ public class OpcUaNameSpace extends ManagedNamespaceWithLifecycle {
// Debounce task cache: used to merge updates within a short period of time, avoiding unnecessary
// duplicate pushes
private final ConcurrentMap<NodeId, ScheduledFuture<?>> debounceTasks = new ConcurrentHashMap<>();
- // Debounce interval: within 10ms, the same node is updated multiple times, and only the last one
+ // Debounce interval: within 50ms, the same node is updated multiple times, and only the last one
// will be pushed (can be adjusted according to your site delay requirements, the minimum can be
// set to 1ms)
private final long debounceIntervalMs;
(iotdb) 02/02: fix
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch opc-fix-master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d99577b54c034a2a63b05f04777910e320cd2445
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue Apr 21 17:33:14 2026 +0800
fix
---
.../protocol/opcua/client/IoTDBOpcUaClient.java | 65 +++++++++++++++++-----
.../sink/protocol/opcua/server/OpcUaNameSpace.java | 12 ++--
2 files changed, 56 insertions(+), 21 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
index 977d672df73..a9aa95cacdf 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
import org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace;
import org.apache.iotdb.pipe.api.exception.PipeException;
+import org.apache.tsfile.common.constant.TsFileConstant;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
@@ -56,9 +57,10 @@ import org.eclipse.milo.opcua.stack.core.types.structured.VariableAttributes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.nio.file.Paths;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
@@ -130,7 +132,6 @@ public class IoTDBOpcUaClient {
StatusCode currentQuality = sink.getDefaultQuality();
Object value = null;
long timestamp = 0;
- NodeId nodeId = null;
NodeId opcDataType = null;
for (int i = 0; i < measurementSchemas.size(); ++i) {
@@ -153,17 +154,43 @@ public class IoTDBOpcUaClient {
"When the 'with-quality' mode is enabled, the measurement must be either \"value-name\" or \"quality-name\"");
continue;
}
- nodeId = new NodeId(NAME_SPACE_INDEX, String.join("/", segments));
final long utcTimestamp = timestampToUtc(timestamps.get(timestamps.size() > 1 ? i : 0));
- value = values.get(i);
- timestamp = utcTimestamp;
- opcDataType = convertToOpcDataType(type);
+ if (Objects.isNull(sink.getValueName())) {
+ writeValue(
+ values.get(i),
+ utcTimestamp,
+ convertToOpcDataType(type),
+ currentQuality,
+ segments,
+ name);
+ } else {
+ value = values.get(i);
+ timestamp = utcTimestamp;
+ opcDataType = convertToOpcDataType(type);
+ }
}
if (Objects.isNull(value)) {
return;
}
+ writeValue(value, timestamp, opcDataType, currentQuality, segments, null);
+ }
+
+ private void writeValue(
+ final Object value,
+ final long timestamp,
+ final NodeId opcDataType,
+ final StatusCode currentQuality,
+ final String[] segments,
+ final @Nullable String name)
+ throws Exception {
+ final NodeId nodeId =
+ new NodeId(
+ NAME_SPACE_INDEX,
+ Objects.nonNull(name)
+ ? String.join("/", segments) + "/" + name
+ : String.join("/", segments));
final Variant variant = new Variant(value);
final DataValue dataValue =
new DataValue(variant, currentQuality, new DateTime(timestamp), new DateTime());
@@ -171,36 +198,41 @@ public class IoTDBOpcUaClient {
if (writeStatus.getValue() == StatusCodes.Bad_NodeIdUnknown) {
final AddNodesResponse addStatus =
- client.addNodes(getNodesToAdd(segments, opcDataType, variant)).get();
+ client.addNodes(getNodesToAdd(segments, name, opcDataType, variant)).get();
for (final AddNodesResult result : addStatus.getResults()) {
if (!result.getStatusCode().equals(StatusCode.GOOD)
&& !(result.getStatusCode().getValue() == StatusCodes.Bad_NodeIdExists)) {
throw new PipeException(
"Failed to create nodes after transfer data value, creation status: "
+ addStatus
- + getErrorString(segments, opcDataType, value, writeStatus));
+ + getErrorString(segments, name, opcDataType, value, writeStatus));
}
}
writeStatus = client.writeValue(nodeId, dataValue).get();
if (writeStatus.getValue() != StatusCode.GOOD.getValue()) {
throw new PipeException(
"Failed to transfer dataValue after successfully created nodes"
- + getErrorString(segments, opcDataType, value, writeStatus));
+ + getErrorString(segments, name, opcDataType, value, writeStatus));
}
} else if (writeStatus.getValue() != StatusCode.GOOD.getValue()) {
throw new PipeException(
"Failed to transfer dataValue"
- + getErrorString(segments, opcDataType, value, writeStatus));
+ + getErrorString(segments, name, opcDataType, value, writeStatus));
}
}
private static String getErrorString(
final String[] segments,
+ final @Nullable String name,
final NodeId dataType,
final Object value,
final StatusCode writeStatus) {
- return ", segments: "
- + Arrays.toString(segments)
+ return ", measurement: "
+ + (Objects.nonNull(name)
+ ? String.join(TsFileConstant.PATH_SEPARATOR, segments)
+ + TsFileConstant.PATH_SEPARATOR
+ + name
+ : String.join(TsFileConstant.PATH_SEPARATOR, segments))
+ ", dataType: "
+ dataType
+ ", value: "
@@ -210,7 +242,10 @@ public class IoTDBOpcUaClient {
}
public List<AddNodesItem> getNodesToAdd(
- final String[] segments, final NodeId opcDataType, final Variant initialValue) {
+ final String[] segments,
+ final @Nullable String name,
+ final NodeId opcDataType,
+ final Variant initialValue) {
final List<AddNodesItem> addNodesItems = new ArrayList<>();
final StringBuilder sb = new StringBuilder(segments[0]);
ExpandedNodeId curNodeId = new NodeId(NAME_SPACE_INDEX, segments[0]).expanded();
@@ -226,7 +261,7 @@ public class IoTDBOpcUaClient {
Identifiers.FolderType.expanded()));
// segments.length >= 3
- for (int i = 1; i < segments.length - 1; ++i) {
+ for (int i = 1; i < (Objects.nonNull(name) ? segments.length : segments.length - 1); ++i) {
sb.append("/").append(segments[i]);
final ExpandedNodeId nextId = new NodeId(NAME_SPACE_INDEX, sb.toString()).expanded();
addNodesItems.add(
@@ -242,7 +277,7 @@ public class IoTDBOpcUaClient {
curNodeId = nextId;
}
- final String measurementName = segments[segments.length - 1];
+ final String measurementName = Objects.nonNull(name) ? name : segments[segments.length - 1];
sb.append("/").append(measurementName);
addNodesItems.add(
new AddNodesItem(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
index da4480458c5..bfbbddc7636 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
@@ -294,14 +294,14 @@ public class OpcUaNameSpace extends ManagedNamespaceWithLifecycle {
}
final UaVariableNode measurementNode;
final long utcTimestamp = timestampToUtc(timestamps.get(timestamps.size() > 1 ? i : 0));
- final DataValue dataValue =
- new DataValue(
- new Variant(values.get(i)),
- currentQuality,
- new DateTime(utcTimestamp),
- new DateTime());
if (Objects.isNull(sink.getValueName())) {
+ final DataValue dataValue =
+ new DataValue(
+ new Variant(values.get(i)),
+ currentQuality,
+ new DateTime(utcTimestamp),
+ new DateTime());
measurementNode = addNode(name, currentFolder, folderNode, dataValue, type);
if (Objects.isNull(measurementNode.getValue())
|| Objects.isNull(measurementNode.getValue().getSourceTime())