You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by cd...@apache.org on 2026/08/14 13:05:31 UTC
(plc4x) branch develop updated: fix(plc4j/opcua): [Bug]: plc4j-driver-opcua - No subscription events are generated for addCyclicField(...) tags
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 <ch...@c-ware.de>
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