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 1e1edeffcb fix(plc4j/opcua): [Bug]: plc4j-driver-opcua - No
subscription events are generated for addCyclicField(...) tags
1e1edeffcb is described below
commit 1e1edeffcbfecaf80873a4ec9b67f83c44fd0b2f
Author: Christofer Dutz <[email protected]>
AuthorDate: Fri Aug 14 15:05:22 2026 +0200
fix(plc4j/opcua): [Bug]: plc4j-driver-opcua - No subscription events are
generated for addCyclicField(...) tags
Implemented an cyclic-change emulation layer for OPC-UA as OPC-UA generally
doesn't support cyclic subscriptions
Fixes: #1102:
---
.../opcua/protocol/OpcuaSubscriptionHandle.java | 80 +++++++++++++++++++++-
.../plc4x/java/opcua/OpcuaPlcDriverTest.java | 39 +++++++++++
2 files changed, 118 insertions(+), 1 deletion(-)
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 a9c82da9f6..64fca8cbb2 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,6 +20,7 @@ package org.apache.plc4x.java.opcua.protocol;
import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@@ -41,6 +42,7 @@ import
org.apache.plc4x.java.spi.drivers.messages.items.PlcResponseItem;
import
org.apache.plc4x.java.spi.drivers.messages.DefaultPlcConsumerRegistration;
import org.apache.plc4x.java.spi.drivers.messages.DefaultPlcSubscriptionTag;
import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
+import org.apache.plc4x.java.api.model.PlcSubscriptionTag;
import org.apache.plc4x.java.spi.values.PlcNull;
import org.apache.plc4x.java.spi.values.PlcStruct;
import org.slf4j.Logger;
@@ -74,6 +76,10 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
private final List<SubscriptionAcknowledgement>
outstandingAcknowledgements = new CopyOnWriteArrayList<>();
private ScheduledFuture<?> publishTask;
+ /** Most recent value per tag, used to re-report CYCLIC tags on their own
schedule. */
+ private final Map<String, PlcResponseItem<PlcValue>> lastValues = new
ConcurrentHashMap<>();
+ private final List<ScheduledFuture<?>> cyclicTasks = new
CopyOnWriteArrayList<>();
+
public OpcuaSubscriptionHandle(OpcuaConnection plcSubscriber,
Conversation conversation, PlcSubscriptionRequest subscriptionRequest,
Long subscriptionId, long cycleTime) {
this(plcSubscriber, conversation, subscriptionRequest, subscriptionId,
cycleTime, cycleTime);
@@ -190,6 +196,7 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
logger.trace("Scheduling publish event for subscription {}",
subscriptionId);
publishTask =
EXECUTOR.scheduleAtFixedRate(this::sendPublishRequest, revisedCycleTime / 2,
revisedCycleTime, TimeUnit.MILLISECONDS);
+ startCyclicEmitters();
return this;
});
}
@@ -256,15 +263,26 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
// Create Consumer for the response message, error and timeout to be
sent to the Secure Channel
conversation.submit(deleteSubscriptionRequest,
DeleteSubscriptionsResponse.class)
- .thenAccept(responseMessage -> publishTask.cancel(true))
.whenComplete((result, error) -> {
if (error != null) {
logger.error("Deletion of subscription resulted in error",
error);
}
+ // Stop our own scheduled work regardless of how the server
answered - a failed
+ // delete must not leave the publish loop and the cyclic
emitters running.
+ cancelScheduledTasks();
plcSubscriber.removeSubscription(subscriptionId);
});
}
+ private void cancelScheduledTasks() {
+ if (publishTask != null) {
+ publishTask.cancel(true);
+ publishTask = null;
+ }
+ cyclicTasks.forEach(task -> task.cancel(true));
+ cyclicTasks.clear();
+ }
+
/**
* Receive the returned values from the OPCUA server and format it so that
it can be received by the PLC4X client.
*
@@ -289,10 +307,70 @@ public class OpcuaSubscriptionHandle implements
PlcSubscriptionHandle {
}
Map<String, PlcResponseItem<PlcValue>> mappedResponse =
plcSubscriber.readResponse(tagMap, dataValues);
+ // Remember the values so cyclic tags can be reported again on their
own schedule even
+ // though the server only notifies us when something actually changes
- see GH-1102.
+ lastValues.putAll(mappedResponse);
PlcSubscriptionEvent event = new
DefaultPlcSubscriptionEvent(Instant.ofEpochMilli(receiveTs), mappedResponse);
consumers.forEach(plcSubscriptionEventConsumer ->
plcSubscriptionEventConsumer.accept(event));
}
+ /**
+ * Starts the emitters for CYCLIC tags.
+ * <p>
+ * OPC UA has no notion of a polling interval: a monitored item reports
its initial value and
+ * then only reports again when the value changes. A CYCLIC subscription
however promises an
+ * event every interval, so for those tags we re-report the most recently
received value on
+ * the requested schedule (see GH-1102). Tags sharing an interval are
reported together.
+ */
+ private void startCyclicEmitters() {
+ Map<Long, List<String>> tagsByInterval = new LinkedHashMap<>();
+ for (String tagName : tagNames) {
+ PlcSubscriptionTag tag = subscriptionRequest.getTag(tagName);
+ if (tag.getPlcSubscriptionType() != PlcSubscriptionType.CYCLIC) {
+ continue;
+ }
+ long interval =
tag.getDuration().map(Duration::toMillis).orElse(cycleTime);
+ tagsByInterval.computeIfAbsent(interval, k -> new
ArrayList<>()).add(tagName);
+ }
+
+ for (Map.Entry<Long, List<String>> entry : tagsByInterval.entrySet()) {
+ long interval = entry.getKey();
+ List<String> names = entry.getValue();
+ logger.debug("Reporting cyclic tags {} every {}ms", names,
interval);
+ cyclicTasks.add(EXECUTOR.scheduleAtFixedRate(
+ () -> emitCyclicValues(names), interval, interval,
TimeUnit.MILLISECONDS));
+ }
+ }
+
+ /** Reports the last known value of the given tags to the registered
consumers. */
+ private void emitCyclicValues(List<String> names) {
+ try {
+ Map<String, PlcResponseItem<PlcValue>> values = new
LinkedHashMap<>();
+ for (String tagName : names) {
+ PlcResponseItem<PlcValue> value = lastValues.get(tagName);
+ if (value != null) {
+ values.put(tagName, value);
+ }
+ }
+ if (values.isEmpty()) {
+ // Nothing received from the server yet - nothing to report.
+ return;
+ }
+ PlcSubscriptionEvent event = new
DefaultPlcSubscriptionEvent(Instant.now(), values);
+ for (String tagName : values.keySet()) {
+ Consumer<PlcSubscriptionEvent> tagConsumer =
tagConsumers.get(tagName);
+ if (tagConsumer != null) {
+ tagConsumer.accept(new
DefaultPlcSubscriptionEvent(Instant.now(),
+ Map.of(tagName, values.get(tagName))));
+ }
+ }
+ consumers.forEach(consumer -> consumer.accept(event));
+ } catch (Exception e) {
+ // A failing consumer must not kill the scheduled task for good.
+ logger.error("Error while reporting cyclic values for {}", names,
e);
+ }
+ }
+
private void onEventNotification(List<EventFieldList> events) {
long receiveTs = System.currentTimeMillis();
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 9f86217060..335d4f516c 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
@@ -27,6 +27,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.time.Duration;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -676,6 +677,44 @@ public class OpcuaPlcDriverTest {
}
}
}
+ /**
+ * A CYCLIC subscription has to report the value on every interval,
even when it never
+ * changes - see GH-1102. OPC UA itself only notifies on change, so a
static value used to
+ * produce exactly one event (the monitored item's initial value) and
nothing after that.
+ */
+ @Test
+ public void cyclicSubscriptionReportsRepeatedly() throws Exception {
+ PlcConnectionManager connectionManager = new
DefaultPlcDriverManager().getConnectionManager();
+
+ try (PlcConnection connection =
connectionManager.getConnection(tcpConnectionAddress)) {
+ ConcurrentLinkedDeque<PlcSubscriptionEvent> events = new
ConcurrentLinkedDeque<>();
+
+ PlcSubscriptionRequest request =
connection.subscriptionRequestBuilder()
+ // A value nobody is writing to - without the fix this
reports once and stops.
+ .addCyclicTagAddress("static",
STRING_IDENTIFIER_READ_WRITE, Duration.ofMillis(500))
+ .build();
+
+ PlcSubscriptionResponse response = request.execute().get(60,
TimeUnit.SECONDS);
+
assertThat(response.getResponseCode("static")).isEqualTo(PlcResponseCode.OK);
+ response.getSubscriptionHandles().forEach(handle ->
handle.register(events::add));
+
+ // Five cycles of head room for three expected reports.
+ for (int i = 0; i < 50 && events.size() < 3; i++) {
+ Thread.sleep(100);
+ }
+
+ assertThat(events.size())
+ .withFailMessage("expected repeated cyclic reports, got %d
event(s)", events.size())
+ .isGreaterThanOrEqualTo(3);
+ assertThat(events.getLast().getPlcValue("static")).isNotNull();
+
+ connection.unsubscriptionRequestBuilder()
+ .addHandles(response.getSubscriptionHandles())
+ .build()
+ .execute();
+ }
+ }
+
/**
* 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
