This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new ae3fd1e88d fix(plc4j/opcua): [Bug]: OPC-UA returns multiple identical
PlcSubscriptionHandles, one for every subscribed tag (Actually fixed this
phenomenon for all drivers supporting subscriptions)
ae3fd1e88d is described below
commit ae3fd1e88dc7886240bf9392feb67f8b23f48a00
Author: Christofer Dutz <[email protected]>
AuthorDate: Fri Aug 14 14:44:35 2026 +0200
fix(plc4j/opcua): [Bug]: OPC-UA returns multiple identical
PlcSubscriptionHandles, one for every subscribed tag (Actually fixed this
phenomenon for all drivers supporting subscriptions)
Fixes: #1896
---
.../apache/plc4x/java/opcua/OpcuaConnection.java | 199 +++++++++++----------
.../opcua/protocol/OpcuaSubscriptionHandle.java | 116 ++++++------
.../plc4x/java/opcua/OpcuaPlcDriverTest.java | 31 ++++
.../messages/DefaultPlcSubscriptionResponse.java | 9 +-
.../DefaultPlcSubscriptionResponseTest.java | 139 ++++++++++++++
5 files changed, 347 insertions(+), 147 deletions(-)
diff --git
a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/OpcuaConnection.java
b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/OpcuaConnection.java
index 632944549d..62f6d3c081 100644
---
a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/OpcuaConnection.java
+++
b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/OpcuaConnection.java
@@ -89,6 +89,9 @@ import java.util.stream.Collectors;
public class OpcuaConnection extends ConnectionBase<OpcuaConfiguration>
implements OpcuaWire {
private static final Logger LOGGER =
LoggerFactory.getLogger(OpcuaConnection.class);
+
+ /** Publishing interval used when a subscription tag doesn't ask for one.
*/
+ private static final Duration DEFAULT_CYCLE_TIME = Duration.ofMillis(1000);
public static final PascalString NULL_STRING = new PascalString(null);
private static final ExpandedNodeId NULL_EXPANDED_NODEID = new
ExpandedNodeId(false,
false,
@@ -551,7 +554,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
ReadRequest request = new
ReadRequest(conversation.createRequestHeader(), 0.0d,
TimestampsToReturn.timestampsToReturnNeither,
Collections.singletonList(dataTypeRead));
return conversation.submit(request,
ReadResponse.class).thenCompose(response -> {
- NodeId dataTypeNodeId =
dataTypeNodeIdOf(response.getResults().get(0));
+ NodeId dataTypeNodeId =
dataTypeNodeIdOf(response.getResults().getFirst());
if (dataTypeNodeId == null) {
return CompletableFuture.completedFuture(null);
}
@@ -566,7 +569,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
ReadRequest request = new
ReadRequest(conversation.createRequestHeader(), 0.0d,
TimestampsToReturn.timestampsToReturnNeither,
Collections.singletonList(definitionRead));
return conversation.submit(request,
ReadResponse.class).thenApply(response -> {
- DataValue value = response.getResults().get(0);
+ DataValue value = response.getResults().getFirst();
if (!value.getValueSpecified() || !(value.getValue() instanceof
VariantExtensionObject)) {
// Server doesn't expose the DataTypeDefinition attribute
(e.g. pre-1.04 servers
// return BadAttributeIdInvalid); the layout must come from
the legacy type
@@ -579,7 +582,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
}
// The DataTypeDefinition is a well-known standard type
(ns=0;i=101), so it is parsed
// into a typed BinaryExtensionObjectWithMask whose body is the
StructureDefinition.
- ExtensionObjectDefinition body = extensionObjects.get(0).getBody();
+ ExtensionObjectDefinition body =
extensionObjects.getFirst().getBody();
return (body instanceof StructureDefinition) ?
(StructureDefinition) body : null;
});
}
@@ -640,11 +643,11 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
new ViewDescription(new NodeId(new NodeIdTwoByte((short) 0)), 0L,
0L), 0L,
Collections.singletonList(description));
return conversation.submit(request,
BrowseResponse.class).thenApply(response -> {
- BrowseResult result = response.getResults().get(0);
+ BrowseResult result = response.getResults().getFirst();
if (result.getReferences() == null ||
result.getReferences().isEmpty()) {
return null;
}
- ExpandedNodeId target = result.getReferences().get(0).getNodeId();
+ ExpandedNodeId target =
result.getReferences().getFirst().getNodeId();
return target == null ? null : new NodeId(target.getNodeId());
});
}
@@ -654,7 +657,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
ReadValueId readValueId = new ReadValueId(node, attributeId,
NULL_STRING, new QualifiedName(0, NULL_STRING));
ReadRequest request = new
ReadRequest(conversation.createRequestHeader(), 0.0d,
TimestampsToReturn.timestampsToReturnNeither,
Collections.singletonList(readValueId));
- return conversation.submit(request,
ReadResponse.class).thenApply(response -> response.getResults().get(0));
+ return conversation.submit(request,
ReadResponse.class).thenApply(response -> response.getResults().getFirst());
}
private static String stringValueOf(DataValue dataValue) {
@@ -662,7 +665,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
return null;
}
List<PascalString> values = ((VariantString)
dataValue.getValue()).getValue();
- return (values == null || values.isEmpty()) ? null :
values.get(0).getStringValue();
+ return (values == null || values.isEmpty()) ? null :
values.getFirst().getStringValue();
}
private static byte[] byteStringValueOf(DataValue dataValue) {
@@ -673,7 +676,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
if (arrays == null || arrays.isEmpty()) {
return null;
}
- List<Short> bytes = arrays.get(0).getValue();
+ List<Short> bytes = arrays.getFirst().getValue();
byte[] result = new byte[bytes.size()];
for (int i = 0; i < bytes.size(); i++) {
result[i] = bytes.get(i).byteValue();
@@ -715,7 +718,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
Set<String> lengthFields = new HashSet<>();
for (int i = 0; i < fieldNodes.getLength(); i++) {
String lengthField = ((Element)
fieldNodes.item(i)).getAttribute("LengthField");
- if (lengthField != null && !lengthField.isEmpty()) {
+ if (!lengthField.isEmpty()) {
lengthFields.add(lengthField);
}
}
@@ -748,25 +751,24 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
return null;
}
String local = typeName.contains(":") ?
typeName.substring(typeName.indexOf(':') + 1) : typeName;
- switch (local) {
- case "Boolean": return 1;
- case "SByte": return 2;
- case "Byte": return 3;
- case "Int16": return 4;
- case "UInt16": return 5;
- case "Int32": return 6;
- case "UInt32": return 7;
- case "Int64": return 8;
- case "UInt64": return 9;
- case "Float": return 10;
- case "Double": return 11;
- case "String":
- case "CharArray": return 12;
- case "DateTime": return 13;
- case "Guid": return 14;
- case "ByteString": return 15;
- default: return null;
- }
+ return switch (local) {
+ case "Boolean" -> 1;
+ case "SByte" -> 2;
+ case "Byte" -> 3;
+ case "Int16" -> 4;
+ case "UInt16" -> 5;
+ case "Int32" -> 6;
+ case "UInt32" -> 7;
+ case "Int64" -> 8;
+ case "UInt64" -> 9;
+ case "Float" -> 10;
+ case "Double" -> 11;
+ case "String", "CharArray" -> 12;
+ case "DateTime" -> 13;
+ case "Guid" -> 14;
+ case "ByteString" -> 15;
+ default -> null;
+ };
}
/** Turns a struct Variant (single or array of ExtensionObjects) into a
PlcStruct / PlcList of PlcStruct. */
@@ -836,23 +838,22 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
int id = builtInDataTypeId == null ? -1 : builtInDataTypeId.intValue();
// Each OPC UA built-in is read with the smallest buffer reader that
covers its bit width
// (readUnsignedByte only handles 1-7 bits, so 8-bit and unsigned
values step up a size).
- switch (id) {
- case 1: return new PlcBOOL(buffer.readUnsignedShort(8) != 0);
- case 2: return new PlcSINT(buffer.readSignedByte(8));
- case 3: return new PlcUSINT(buffer.readUnsignedShort(8));
- case 4: return new PlcINT(buffer.readSignedShort(16));
- case 5: return new PlcUINT(buffer.readUnsignedInt(16));
- case 6: return new PlcDINT(buffer.readSignedInt(32));
- case 7: return new PlcUDINT(buffer.readUnsignedLong(32));
- case 8: return new PlcLINT(buffer.readSignedLong(64));
- case 9: return new PlcULINT(buffer.readUnsignedBigInteger(64));
- case 10: return new PlcREAL(buffer.readFloat(32));
- case 11: return new PlcLREAL(buffer.readDouble(64));
- case 12: return new
PlcSTRING(PascalString.staticParse(buffer).getStringValue());
- default:
- throw new PlcRuntimeException("Unsupported struct field data
type id " + id
- + " (nested structs, enums and non-builtin types are not
yet decoded)");
- }
+ return switch (id) {
+ case 1 -> new PlcBOOL(buffer.readUnsignedShort(8) != 0);
+ case 2 -> new PlcSINT(buffer.readSignedByte(8));
+ case 3 -> new PlcUSINT(buffer.readUnsignedShort(8));
+ case 4 -> new PlcINT(buffer.readSignedShort(16));
+ case 5 -> new PlcUINT(buffer.readUnsignedInt(16));
+ case 6 -> new PlcDINT(buffer.readSignedInt(32));
+ case 7 -> new PlcUDINT(buffer.readUnsignedLong(32));
+ case 8 -> new PlcLINT(buffer.readSignedLong(64));
+ case 9 -> new PlcULINT(buffer.readUnsignedBigInteger(64));
+ case 10 -> new PlcREAL(buffer.readFloat(32));
+ case 11 -> new PlcLREAL(buffer.readDouble(64));
+ case 12 -> new
PlcSTRING(PascalString.staticParse(buffer).getStringValue());
+ default -> throw new PlcRuntimeException("Unsupported struct field
data type id " + id
+ + " (nested structs, enums and non-builtin types are not yet
decoded)");
+ };
}
/** Canonical key for a NodeId (namespace + identifier) used to cache
resolved type layouts. */
@@ -952,7 +953,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
new ViewDescription(new NodeId(new NodeIdTwoByte((short) 0)), 0L,
0L), 0L,
Collections.singletonList(description));
return conversation.submit(request,
BrowseResponse.class).thenApply(response -> {
- List<ReferenceDescription> references =
response.getResults().get(0).getReferences();
+ List<ReferenceDescription> references =
response.getResults().getFirst().getReferences();
if (references == null || references.isEmpty()) {
return null;
}
@@ -961,7 +962,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
return reference.getNodeId();
}
}
- return references.get(0).getNodeId();
+ return references.getFirst().getNodeId();
});
}
@@ -974,7 +975,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
Collections.singletonList(extensionObject));
}
- /** Serialises a PlcStruct into the OPC UA binary body described by its
StructureDefinition. */
+ /** Serializes a PlcStruct into the OPC UA binary body described by its
StructureDefinition. */
private static byte[] encodeStructBody(PlcValue value, StructureDefinition
definition) {
try {
byte[] body = new byte[structBodySize(value, definition)];
@@ -1049,15 +1050,14 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
private static int scalarSize(Long builtInDataTypeId, PlcValue value) {
int id = builtInDataTypeId == null ? -1 : builtInDataTypeId.intValue();
- switch (id) {
- case 1: case 2: case 3: return 1;
- case 4: case 5: return 2;
- case 6: case 7: case 10: return 4;
- case 8: case 9: case 11: return 8;
- case 12: return new
PascalString(value.getString()).getLengthInBytes();
- default:
- throw new PlcRuntimeException("Unsupported struct field data
type id " + id + " for writing");
- }
+ return switch (id) {
+ case 1, 2, 3 -> 1;
+ case 4, 5 -> 2;
+ case 6, 7, 10 -> 4;
+ case 8, 9, 11 -> 8;
+ case 12 -> new PascalString(value.getString()).getLengthInBytes();
+ default -> throw new PlcRuntimeException("Unsupported struct field
data type id " + id + " for writing");
+ };
}
// The standard "Objects" folder — the usual entry point into a server's
address space.
@@ -1184,14 +1184,14 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
0L, // requestedMaxReferencesPerNode 0 ->
server decides
Collections.singletonList(description));
resultFuture = conversation.submit(request, BrowseResponse.class)
- .thenApply(response -> response.getResults().get(0));
+ .thenApply(response -> response.getResults().getFirst());
} else {
BrowseNextRequest request = new BrowseNextRequest(
conversation.createRequestHeader(),
false, // don't release the continuation point,
we want the next batch
Collections.singletonList(continuationPoint));
resultFuture = conversation.submit(request,
BrowseNextResponse.class)
- .thenApply(response -> response.getResults().get(0));
+ .thenApply(response -> response.getResults().getFirst());
}
return resultFuture.thenCompose(result -> {
if (result.getReferences() != null) {
@@ -1249,7 +1249,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
}
return new DefaultPlcBrowseItem(tag, name, readable, writable,
- Collections.<PlcSubscriptionType>emptySet(), false, arrayInfo,
children, options);
+ Collections.emptySet(), false, arrayInfo, children, options);
}
// The node attributes read for every variable child during a browse so
items carry a real
@@ -1437,7 +1437,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
return null;
}
List<NodeId> nodeIds = ((VariantNodeId)
dataValue.getValue()).getValue();
- return (nodeIds == null || nodeIds.isEmpty()) ? null : nodeIds.get(0);
+ return (nodeIds == null || nodeIds.isEmpty()) ? null :
nodeIds.getFirst();
}
/**
@@ -1501,7 +1501,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
return null;
}
List<Short> values = ((VariantByte) dataValue.getValue()).getValue();
- return (values == null || values.isEmpty()) ? null : values.get(0);
+ return (values == null || values.isEmpty()) ? null : values.getFirst();
}
/** Extracts the first Int32 from a Variant, or null if not an int32
scalar. */
@@ -1510,7 +1510,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
return null;
}
List<Integer> values = ((VariantInt32)
dataValue.getValue()).getValue();
- return (values == null || values.isEmpty()) ? null : values.get(0);
+ return (values == null || values.isEmpty()) ? null : values.getFirst();
}
/** Extracts the UInt32 list from a Variant, or null if not a uint32
value. */
@@ -1557,19 +1557,17 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
}
NodeIdTypeDefinition nodeId = expandedNodeId.getNodeId();
String identifier = nodeId.getIdentifier();
- if (nodeId instanceof NodeIdTwoByte) {
- return "ns=0;i=" + identifier;
- } else if (nodeId instanceof NodeIdFourByte fourByte) {
- return "ns=" + fourByte.getNamespaceIndex() + ";i=" + identifier;
- } else if (nodeId instanceof NodeIdNumeric numeric) {
- return "ns=" + numeric.getNamespaceIndex() + ";i=" + identifier;
- } else if (nodeId instanceof NodeIdString string) {
- return "ns=" + string.getNamespaceIndex() + ";s=" + identifier;
- }
- // GUID and opaque (ByteString) node identifiers are not round-tripped
in Phase 1:
- // NodeIdGuid#getIdentifier() returns the raw byte-array toString(),
which the tag
- // parser can't turn back into a usable node id. Such nodes are
skipped for now.
- return null;
+ return switch (nodeId) {
+ case NodeIdTwoByte ignored -> "ns=0;i=" + identifier;
+ case NodeIdFourByte fourByte -> "ns=" +
fourByte.getNamespaceIndex() + ";i=" + identifier;
+ case NodeIdNumeric numeric -> "ns=" + numeric.getNamespaceIndex()
+ ";i=" + identifier;
+ case NodeIdString string -> "ns=" + string.getNamespaceIndex() +
";s=" + identifier;
+ default ->
+ // GUID and opaque (ByteString) node identifiers are not
round-tripped in Phase 1:
+ // NodeIdGuid#getIdentifier() returns the raw byte-array
toString(), which the tag
+ // parser can't turn back into a usable node id. Such nodes
are skipped for now.
+ null;
+ };
}
private static String localizedTextValue(LocalizedText localizedText) {
@@ -1595,7 +1593,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
case null -> {
return new PlcNull();
}
- case VariantNull variantNull -> {
+ case VariantNull ignored -> {
return new PlcNull();
}
case VariantBoolean variantBoolean -> {
@@ -2036,9 +2034,9 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
} else if (serverAttributes != null &&
serverAttributes.unsigned()) {
// Node is the abstract UInteger type: no concrete built-in to
resolve to, so pick
// an unsigned type sized to the value (an abstract node
accepts any subtype).
- dataType = inferUnsignedWriteType(plcValueList.get(0));
+ dataType = inferUnsignedWriteType(plcValueList.getFirst());
} else {
- dataType = inferWriteType(plcValueList.get(0));
+ dataType = inferWriteType(plcValueList.getFirst());
}
}
int length = valueObject.getLength();
@@ -2390,12 +2388,24 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
@Override
protected CompletableFuture<PlcSubscriptionResponse>
onSubscribe(PlcSubscriptionRequest subscriptionRequest) {
List<String> tagNames = new
ArrayList<>(subscriptionRequest.getTagNames());
- long cycleTime =
(subscriptionRequest.getTag(tagNames.get(0))).getDuration().orElse(Duration.ofMillis(1000)).toMillis();
+ // A subscription has a single publishing interval, so take the
fastest rate any of its tags
+ // asked for - taking the first tag's rate would starve every faster
tag behind it. The
+ // individual sampling rates are set per monitored item (see
onSubscribeCreateMonitoredItemsRequest).
+ long cycleTime = tagNames.stream()
+ .map(subscriptionRequest::getTag)
+ .map(tag -> tag.getDuration().orElse(DEFAULT_CYCLE_TIME))
+ .min(Duration::compareTo)
+ .orElse(DEFAULT_CYCLE_TIME)
+ .toMillis();
return onSubscribeCreateSubscription(cycleTime).thenApply(response -> {
long subscriptionId = response.getSubscriptionId();
+ // The server may revise the publishing interval; everything
we time against it
+ // (the publish request cadence and its timeouts) has to
follow the revised value,
+ // not what we asked for.
+ long revisedCycleTime = revisedPublishingInterval(response,
cycleTime);
OpcuaSubscriptionHandle handle = new
OpcuaSubscriptionHandle(this,
- conversation, subscriptionRequest, subscriptionId,
cycleTime);
+ conversation, subscriptionRequest, subscriptionId,
cycleTime, revisedCycleTime);
if (subscriptionRequest.getConsumer() != null) {
handle.register(subscriptionRequest.getConsumer());
}
@@ -2408,7 +2418,7 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
subscriptions.put(handle.getSubscriptionId(), handle);
return handle;
})
- .thenCompose(handle ->
handle.onSubscribeCreateMonitoredItemsRequest())
+
.thenCompose(OpcuaSubscriptionHandle::onSubscribeCreateMonitoredItemsRequest)
.thenApply(handle -> {
Map<String, PlcResponseItem<PlcSubscriptionHandle>> values =
new HashMap<>();
for (String tagName : subscriptionRequest.getTagNames()) {
@@ -2424,6 +2434,23 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
});
}
+ /**
+ * The publishing interval the server actually granted, in milliseconds.
Servers are free to
+ * revise the interval we requested (typically upwards, to their smallest
supported rate).
+ * Falls back to the requested value if the server reports something
unusable.
+ */
+ private static long revisedPublishingInterval(CreateSubscriptionResponse
response, long requestedCycleTime) {
+ double revised = response.getRevisedPublishingInterval();
+ if (revised <= 0) {
+ return requestedCycleTime;
+ }
+ long revisedMillis = Math.round(revised);
+ if (revisedMillis != requestedCycleTime) {
+ LOGGER.debug("Server revised the publishing interval from {}ms to
{}ms", requestedCycleTime, revisedMillis);
+ }
+ return revisedMillis;
+ }
+
private CompletableFuture<CreateSubscriptionResponse>
onSubscribeCreateSubscription(long cycleTime) {
LOGGER.trace("Entering creating subscription request");
@@ -2477,12 +2504,4 @@ public class OpcuaConnection extends
ConnectionBase<OpcuaConfiguration> implemen
return (dateTime - EPOCH_OFFSET) / 10000;
}
- private GuidValue toGuidValue(String identifier) {
- LOGGER.error("Querying Guid nodes is not supported");
- byte[] data4 = new byte[]{0, 0};
- byte[] data5 = new byte[]{0, 0, 0, 0, 0, 0};
- return new GuidValue(0L, 0, 0, data4, data5);
- }
-
-
}
diff --git
a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java
b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java
index eb71c81df7..a9c82da9f6 100644
---
a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java
+++
b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java
@@ -20,15 +20,12 @@ package org.apache.plc4x.java.opcua.protocol;
import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
-import java.util.Map.Entry;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
-import org.apache.plc4x.java.api.messages.PlcMetadataKeys;
import org.apache.plc4x.java.api.messages.PlcSubscriptionEvent;
import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
-import org.apache.plc4x.java.api.metadata.Metadata;
import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem;
import org.apache.plc4x.java.api.model.PlcConsumerRegistration;
@@ -49,6 +46,7 @@ import org.apache.plc4x.java.spi.values.PlcStruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.CompletableFuture;
@@ -78,6 +76,17 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
public OpcuaSubscriptionHandle(OpcuaConnection plcSubscriber,
Conversation conversation, PlcSubscriptionRequest subscriptionRequest,
Long subscriptionId, long cycleTime) {
+ this(plcSubscriber, conversation, subscriptionRequest, subscriptionId,
cycleTime, cycleTime);
+ }
+
+ /**
+ * @param cycleTime the publishing interval that was requested
+ * @param revisedCycleTime the publishing interval the server granted; the
publish request
+ * cadence and its timeouts are derived from this
one
+ */
+ public OpcuaSubscriptionHandle(OpcuaConnection plcSubscriber,
+ Conversation conversation, PlcSubscriptionRequest subscriptionRequest,
Long subscriptionId,
+ long cycleTime, long revisedCycleTime) {
this.consumers = new HashSet<>();
this.tagConsumers = new HashMap<>();
this.subscriptionRequest = subscriptionRequest;
@@ -86,7 +95,7 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
this.subscriptionId = subscriptionId;
this.plcSubscriber = plcSubscriber;
this.cycleTime = cycleTime;
- this.revisedCycleTime = cycleTime;
+ this.revisedCycleTime = revisedCycleTime;
}
public CompletableFuture<OpcuaSubscriptionHandle>
onSubscribeCreateMonitoredItemsRequest() {
@@ -135,9 +144,14 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
}
long clientHandle = clientHandles.getAndIncrement();
+ // Each monitored item is sampled at the rate its own tag asked
for. Using the
+ // subscription's cycle time for every item would silently give
all tags the rate of
+ // whichever tag happened to come first - see the discussion in
GH-1896.
+ double samplingInterval = tagDefaultPlcSubscription.getDuration()
+ .map(Duration::toMillis).orElse(cycleTime);
MonitoringParameters parameters = new MonitoringParameters(
clientHandle,
- (double) cycleTime, // sampling interval
+ samplingInterval,
eventFilter, // filter, null means use default
1L, // queue size
true // discard oldest
@@ -164,7 +178,7 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
logger.info("Error while sending the Create Monitored Item
Subscription Message", error);
}
}).thenApply(responseMessage -> {
- MonitoredItemCreateResult[] array =
responseMessage.getResults().stream().toArray(MonitoredItemCreateResult[]::new);
+ MonitoredItemCreateResult[] array =
responseMessage.getResults().toArray(MonitoredItemCreateResult[]::new);
for (int index = 0, arrayLength = array.length; index <
arrayLength; index++) {
MonitoredItemCreateResult result = array[index];
if
(OpcuaStatusCode.enumForValue(result.getStatusCode().getStatusCode()) !=
OpcuaStatusCode.Good) {
@@ -186,59 +200,49 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
* The server will respond at most once every cycle.
*/
private void sendPublishRequest() {
- List<Long> outstandingRequests = new LinkedList<>();
-
- //If we are waiting on a response and haven't received one, just wait
until we do. A keep alive will be sent out eventually
- if (outstandingRequests.size() <= 1) {
- RequestHeader requestHeader =
conversation.createRequestHeader(this.revisedCycleTime * 10);
-
- //Make a copy of the outstanding requests, so it isn't modified
while we are putting the ack list together.
- List<SubscriptionAcknowledgement> acks = new
ArrayList<>(outstandingAcknowledgements);
- // do not send -1 when requesting publish, the -1 value indicates
NULL value
- // which might result in corruption of subscription for some
servers
- int ackLength = acks.size();
- outstandingAcknowledgements.removeAll(acks);
-
- PublishRequest publishRequest = new PublishRequest(requestHeader,
acks);
- logger.trace("Sent publish request with {} acks", ackLength);
- // Create Consumer for the response message, error and timeout to
be sent to the Secure Channel
- conversation.submit(publishRequest,
PublishResponse.class).thenAccept(responseMessage -> {
-
outstandingRequests.remove(responseMessage.getResponseHeader().getRequestHandle());
-
- for (long availableSequenceNumber :
responseMessage.getAvailableSequenceNumbers()) {
- outstandingAcknowledgements.add(new
SubscriptionAcknowledgement(this.subscriptionId, availableSequenceNumber));
- }
+ RequestHeader requestHeader =
conversation.createRequestHeader(this.revisedCycleTime * 10);
+
+ //Make a copy of the outstanding requests, so it isn't modified while
we are putting the ack list together.
+ List<SubscriptionAcknowledgement> acks = new
ArrayList<>(outstandingAcknowledgements);
+ // do not send -1 when requesting publish, the -1 value indicates NULL
value
+ // which might result in corruption of subscription for some servers
+ int ackLength = acks.size();
+ outstandingAcknowledgements.removeAll(acks);
- NotificationMessage message =
responseMessage.getNotificationMessage();
- if (message.getNotificationData() != null) {
- for (ExtensionObject notificationMessage :
message.getNotificationData()) {
- ExtensionObjectDefinition notification =
notificationMessage.getBody();
- if (notification instanceof
DataChangeNotification) {
- logger.trace("Found a Data Change
Notification");
- DataChangeNotification data =
(DataChangeNotification) notification;
- if (!data.getMonitoredItems().isEmpty()) {
- onMonitoredValue(data.getMonitoredItems());
- }
- } else if (notification instanceof
EventNotificationList) {
- logger.trace("Found a Event Notification");
- EventNotificationList data =
(EventNotificationList) notification;
- if (!data.getEvents().isEmpty()) {
- onEventNotification(data.getEvents());
- }
- } else {
- logger.warn("Unsupported Notification type
{}", notification.getClass().getName());
- }
+ PublishRequest publishRequest = new PublishRequest(requestHeader,
acks);
+ logger.trace("Sent publish request with {} acks", ackLength);
+ // Create Consumer for the response message, error and timeout to be
sent to the Secure Channel
+ conversation.submit(publishRequest,
PublishResponse.class).thenAccept(responseMessage -> {
+ for (long availableSequenceNumber :
responseMessage.getAvailableSequenceNumbers()) {
+ outstandingAcknowledgements.add(new
SubscriptionAcknowledgement(this.subscriptionId, availableSequenceNumber));
+ }
+
+ NotificationMessage message =
responseMessage.getNotificationMessage();
+ if (message.getNotificationData() != null) {
+ for (ExtensionObject notificationMessage :
message.getNotificationData()) {
+ ExtensionObjectDefinition notification =
notificationMessage.getBody();
+ if (notification instanceof DataChangeNotification data) {
+ logger.trace("Found a Data Change Notification");
+ if (!data.getMonitoredItems().isEmpty()) {
+ onMonitoredValue(data.getMonitoredItems());
+ }
+ } else if (notification instanceof EventNotificationList
data) {
+ logger.trace("Found a Event Notification");
+ if (!data.getEvents().isEmpty()) {
+ onEventNotification(data.getEvents());
}
+ } else {
+ logger.warn("Unsupported Notification type {}",
notification.getClass().getName());
}
- }).whenComplete((result, error) -> {
- if (error != null) {
- logger.warn("Publish request of subscription {} resulted
in error reported by server", subscriptionId, error);
- } else {
- logger.trace("Completed publish request for subscription
{}", subscriptionId);
}
- });
- outstandingRequests.add(requestHeader.getRequestHandle());
- }
+ }
+ }).whenComplete((result, error) -> {
+ if (error != null) {
+ logger.warn("Publish request of subscription {} resulted in
error reported by server", subscriptionId, error);
+ } else {
+ logger.trace("Completed publish request for subscription {}",
subscriptionId);
+ }
+ });
}
@@ -337,7 +341,7 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
}
public PlcConsumerRegistration registerTagConsumer(String tagName,
Consumer<PlcSubscriptionEvent> consumer) {
- logger.info("Registering a new OPCUA subscription consumer for tag
with name " + tagName);
+ logger.info("Registering a new OPCUA subscription consumer for tag
with name {}", tagName);
tagConsumers.put(tagName, consumer);
return new DefaultPlcConsumerRegistration(plcSubscriber, consumer,
this);
}
diff --git
a/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/OpcuaPlcDriverTest.java
b/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/OpcuaPlcDriverTest.java
index dc2fcdea7c..9f86217060 100644
---
a/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/OpcuaPlcDriverTest.java
+++
b/plc4j/drivers/opcua/src/test/java/org/apache/plc4x/java/opcua/OpcuaPlcDriverTest.java
@@ -676,6 +676,37 @@ public class OpcuaPlcDriverTest {
}
}
}
+ /**
+ * A subscription covering several tags is served by a single handle,
so
+ * getSubscriptionHandles() has to report exactly one - see GH-1896.
Reporting it once per
+ * tag makes the documented "register a consumer on every handle" loop
attach N consumers
+ * and deliver every event N times.
+ */
+ @Test
+ public void multiTagSubscriptionReportsOneHandle() throws Exception {
+ PlcConnectionManager connectionManager = new
DefaultPlcDriverManager().getConnectionManager();
+
+ try (PlcConnection connection =
connectionManager.getConnection(tcpConnectionAddress)) {
+ PlcSubscriptionRequest request =
connection.subscriptionRequestBuilder()
+ .addChangeOfStateTag("first",
OpcuaTag.of(INTEGER_IDENTIFIER_READ_WRITE))
+ .addChangeOfStateTag("second",
OpcuaTag.of(BOOL_IDENTIFIER_READ_WRITE))
+ .addChangeOfStateTag("third",
OpcuaTag.of(INT32_IDENTIFIER_READ_WRITE))
+ .build();
+
+ PlcSubscriptionResponse response = request.execute().get(60,
TimeUnit.SECONDS);
+
assertThat(response.getResponseCode("first")).isEqualTo(PlcResponseCode.OK);
+
assertThat(response.getResponseCode("second")).isEqualTo(PlcResponseCode.OK);
+
assertThat(response.getResponseCode("third")).isEqualTo(PlcResponseCode.OK);
+
+ assertThat(response.getSubscriptionHandles()).hasSize(1);
+
+ connection.unsubscriptionRequestBuilder()
+ .addHandles(response.getSubscriptionHandles())
+ .build()
+ .execute();
+ }
+ }
+
@Test
public void manySubscriptionsOnSingleConnection() throws Exception {
final int numberOfSubscriptions = 3;
diff --git
a/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/messages/DefaultPlcSubscriptionResponse.java
b/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/messages/DefaultPlcSubscriptionResponse.java
index ae8b2251be..b50c35ad1f 100644
---
a/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/messages/DefaultPlcSubscriptionResponse.java
+++
b/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/messages/DefaultPlcSubscriptionResponse.java
@@ -71,9 +71,16 @@ public class DefaultPlcSubscriptionResponse implements
PlcSubscriptionResponse {
@Override
public Collection<PlcSubscriptionHandle> getSubscriptionHandles() {
+ // A single handle can cover several tags - drivers commonly create
one handle for the whole
+ // request and map it to every tag name. Returning it once per tag
would make the usual
+ // "register a consumer on every handle" loop deliver each event N
times, see GH-1896.
+ // Duplicates are detected by identity: two distinct handles that
happen to compare equal
+ // are still two subscriptions.
+ Map<PlcSubscriptionHandle, Boolean> seen = new IdentityHashMap<>();
List<PlcSubscriptionHandle> handles = new ArrayList<>(values.size());
for (PlcResponseItem<PlcSubscriptionHandle> item : values.values()) {
- if (item != null && item.getValue() != null) {
+ if (item != null && item.getValue() != null
+ && seen.put(item.getValue(), Boolean.TRUE) == null) {
handles.add(item.getValue());
}
}
diff --git
a/plc4j/spi/drivers/src/test/java/org/apache/plc4x/java/spi/drivers/messages/DefaultPlcSubscriptionResponseTest.java
b/plc4j/spi/drivers/src/test/java/org/apache/plc4x/java/spi/drivers/messages/DefaultPlcSubscriptionResponseTest.java
new file mode 100644
index 0000000000..906a0ef6d5
--- /dev/null
+++
b/plc4j/spi/drivers/src/test/java/org/apache/plc4x/java/spi/drivers/messages/DefaultPlcSubscriptionResponseTest.java
@@ -0,0 +1,139 @@
+/*
+ * 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
+ *
+ * https://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.plc4x.java.spi.drivers.messages;
+
+import org.apache.plc4x.java.api.messages.PlcSubscriptionEvent;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
+import org.apache.plc4x.java.api.model.PlcConsumerRegistration;
+import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
+import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem;
+import org.apache.plc4x.java.spi.drivers.messages.items.PlcResponseItem;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class DefaultPlcSubscriptionResponseTest {
+
+ /**
+ * Drivers commonly create one handle for the whole request and map it to
every tag name.
+ * getSubscriptionHandles() must report that handle once, otherwise the
usual "register a
+ * consumer on every handle" loop delivers each event once per tag - see
GH-1896.
+ */
+ @Test
+ void oneHandleSharedByManyTagsIsReportedOnce() {
+ PlcSubscriptionHandle handle = new TestHandle();
+ Map<String, PlcResponseItem<PlcSubscriptionHandle>> values = new
LinkedHashMap<>();
+ for (int i = 0; i < 5; i++) {
+ values.put("tag" + i, new
DefaultPlcResponseItem<>(PlcResponseCode.OK, handle));
+ }
+
+ Collection<PlcSubscriptionHandle> handles =
response(values).getSubscriptionHandles();
+
+ assertEquals(1, handles.size());
+ assertSame(handle, handles.iterator().next());
+ }
+
+ /**
+ * Distinct handles must all survive, even when they compare equal - two
subscriptions are two
+ * subscriptions.
+ */
+ @Test
+ void distinctHandlesAreAllReported() {
+ PlcSubscriptionHandle first = new AlwaysEqualHandle();
+ PlcSubscriptionHandle second = new AlwaysEqualHandle();
+ Map<String, PlcResponseItem<PlcSubscriptionHandle>> values = new
LinkedHashMap<>();
+ values.put("a", new DefaultPlcResponseItem<>(PlcResponseCode.OK,
first));
+ values.put("b", new DefaultPlcResponseItem<>(PlcResponseCode.OK,
second));
+
+ Collection<PlcSubscriptionHandle> handles =
response(values).getSubscriptionHandles();
+
+ assertEquals(2, handles.size());
+ assertTrue(handles.stream().anyMatch(h -> h == first));
+ assertTrue(handles.stream().anyMatch(h -> h == second));
+ }
+
+ /**
+ * A mix: two tags share one handle, a third has its own.
+ */
+ @Test
+ void mixedSharedAndDistinctHandles() {
+ PlcSubscriptionHandle shared = new TestHandle();
+ PlcSubscriptionHandle own = new TestHandle();
+ Map<String, PlcResponseItem<PlcSubscriptionHandle>> values = new
LinkedHashMap<>();
+ values.put("a", new DefaultPlcResponseItem<>(PlcResponseCode.OK,
shared));
+ values.put("b", new DefaultPlcResponseItem<>(PlcResponseCode.OK,
shared));
+ values.put("c", new DefaultPlcResponseItem<>(PlcResponseCode.OK, own));
+
+ assertEquals(2, response(values).getSubscriptionHandles().size());
+ }
+
+ /**
+ * Tags that failed to subscribe carry a null handle and must simply be
skipped.
+ */
+ @Test
+ void tagsWithoutAHandleAreSkipped() {
+ PlcSubscriptionHandle handle = new TestHandle();
+ Map<String, PlcResponseItem<PlcSubscriptionHandle>> values = new
LinkedHashMap<>();
+ values.put("ok", new DefaultPlcResponseItem<>(PlcResponseCode.OK,
handle));
+ values.put("bad", new
DefaultPlcResponseItem<>(PlcResponseCode.INVALID_ADDRESS, null));
+
+ assertEquals(1, response(values).getSubscriptionHandles().size());
+ }
+
+ private DefaultPlcSubscriptionResponse response(Map<String,
PlcResponseItem<PlcSubscriptionHandle>> values) {
+ PlcSubscriptionRequest request =
Mockito.mock(PlcSubscriptionRequest.class);
+ Mockito.lenient().when(request.getTagNames()).thenReturn(new
LinkedHashSet<>(values.keySet()));
+ return new DefaultPlcSubscriptionResponse(request, values);
+ }
+
+ private static class TestHandle implements PlcSubscriptionHandle {
+ @Override
+ public PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent>
consumer) {
+ return null;
+ }
+ }
+
+ /** Compares equal to every other instance, to prove de-duplication is by
identity. */
+ private static class AlwaysEqualHandle implements PlcSubscriptionHandle {
+ @Override
+ public PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent>
consumer) {
+ return null;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof AlwaysEqualHandle;
+ }
+
+ @Override
+ public int hashCode() {
+ return 1;
+ }
+ }
+}
