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 2c5680ef03 feat: Added base classes for a subscription-eumulation
layer.
2c5680ef03 is described below
commit 2c5680ef03ee74da9b252bca4f0b7e33c9776381
Author: Christofer Dutz <[email protected]>
AuthorDate: Thu Jul 16 16:45:51 2026 +0200
feat: Added base classes for a subscription-eumulation layer.
---
.../model/DefaultPlcSubscriptionHandle.java | 66 +++
plc4j/utils/pom.xml | 1 +
plc4j/utils/subscription-emulation/README.md | 220 ++++++++
plc4j/utils/subscription-emulation/pom.xml | 147 +++++
.../PollingSubscriptionConnectionBase.java | 593 +++++++++++++++++++++
.../PollingSubscriptionConnectionBaseTest.java | 305 +++++++++++
.../src/test/resources/logback-test.xml | 16 +
7 files changed, 1348 insertions(+)
diff --git
a/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/model/DefaultPlcSubscriptionHandle.java
b/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/model/DefaultPlcSubscriptionHandle.java
new file mode 100644
index 0000000000..4c8fe2b6b1
--- /dev/null
+++
b/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/model/DefaultPlcSubscriptionHandle.java
@@ -0,0 +1,66 @@
+/*
+ * 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.plc4x.java.spi.drivers.model;
+
+import org.apache.plc4x.java.api.messages.PlcSubscriptionEvent;
+import org.apache.plc4x.java.api.model.PlcConsumerRegistration;
+import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
+import org.apache.plc4x.java.spi.drivers.functions.PlcSubscriber;
+
+import java.util.Collections;
+import java.util.function.Consumer;
+
+// Warning: do not override equals and hashCode as these should not include
the plcSubscriber.
+public class DefaultPlcSubscriptionHandle implements PlcSubscriptionHandle {
+
+ private final transient PlcSubscriber plcSubscriber;
+
+ public DefaultPlcSubscriptionHandle(PlcSubscriber plcSubscriber) {
+ this.plcSubscriber = plcSubscriber;
+ }
+
+ @Override
+ public PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent>
consumer) {
+ return plcSubscriber.registerConsumer(consumer,
Collections.singletonList(this));
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof DefaultPlcSubscriptionHandle)) {
+ return false;
+ }
+ // A handle is unique. Therefore, we use the default implementation
from Object
+ return (this == obj);
+ }
+
+ @Override
+ public int hashCode() {
+ // A handle is unique. Therefore, we use the default implementation
from Object
+ return System.identityHashCode(this);
+ }
+
+ @Override
+ public String toString() {
+ return "DefaultPlcSubscriptionHandle{" +
+ "plcSubscriber=" + plcSubscriber +
+ '}';
+ }
+
+}
diff --git a/plc4j/utils/pom.xml b/plc4j/utils/pom.xml
index 5d856b233d..db392fa000 100644
--- a/plc4j/utils/pom.xml
+++ b/plc4j/utils/pom.xml
@@ -37,6 +37,7 @@
<subprojects>
<subproject>audit-log</subproject>
+ <subproject>subscription-emulation</subproject>
<!-- Non-default input options -->
<subproject>pcap-replay</subproject>
<subproject>pcap-shared</subproject>
diff --git a/plc4j/utils/subscription-emulation/README.md
b/plc4j/utils/subscription-emulation/README.md
new file mode 100644
index 0000000000..e0ee5c5c83
--- /dev/null
+++ b/plc4j/utils/subscription-emulation/README.md
@@ -0,0 +1,220 @@
+<!--
+ 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.
+ -->
+
+# Subscription Emulation
+
+This module provides polling-based subscription emulation for drivers that
don't natively support subscriptions.
+
+## PollingSubscriptionConnectionBase
+
+The `PollingSubscriptionConnectionBase` class provides polling-based
subscription emulation for drivers that don't natively support subscriptions
(such as Modbus, BACnet, etc.).
+
+### Features
+
+- **CYCLIC subscriptions** - Polls tags at specified intervals
+- **CHANGE_OF_STATE subscriptions** - Polls tags and only fires events when
values change
+- **EVENT subscriptions** - Not supported (throws
`UnsupportedOperationException`)
+- **Automatic grouping** - Tags with the same subscription type and polling
interval are grouped into a single batch for efficiency
+- **Consumer registration** - Supports request-level, tag-specific, and
registered consumers
+- **Thread-safe** - Uses concurrent data structures for thread-safe operation
+
+### Basic Usage
+
+To add polling-based subscription support to your driver:
+
+1. Add this dependency to your driver's `pom.xml`:
+
+```xml
+<dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-utils-subscription-emulation</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+</dependency>
+```
+
+2. Extend `PollingSubscriptionConnectionBase` instead of `ConnectionBase` and
implement `PlcReader`:
+
+```java
+import
org.apache.plc4x.java.utils.subscriptionemulation.PollingSubscriptionConnectionBase;
+
+public class ModbusTcpConnection
+ extends PollingSubscriptionConnectionBase<ModbusTcpConfiguration>
+ implements PlcReader, PlcWriter {
+
+ public ModbusTcpConnection(ModbusTcpConfiguration configuration,
+ TransportInstance<?> transportInstance) {
+ super(configuration, transportInstance);
+ }
+
+ // Implement PlcReader interface
+ @Override
+ public CompletableFuture<PlcReadResponse> read(PlcReadRequest readRequest)
{
+ // Your read implementation
+ }
+
+ // Rest of your driver implementation...
+}
+```
+
+That's it! Your driver now supports subscriptions via polling.
+
+### How It Works
+
+When a subscription is created:
+
+1. The base class groups tags by subscription type and polling interval
+2. For each group, it creates an EventPump batch with a TimerTrigger
+3. The batch polls the tags using your driver's `read()` method
+4. For CYCLIC subscriptions, events are fired on every poll
+5. For CHANGE_OF_STATE subscriptions, events are only fired when values change
+
+### Customization
+
+You can override these methods to customize behavior:
+
+#### Custom Default Polling Interval
+
+```java
+@Override
+protected long getDefaultPollingInterval() {
+ return 500; // 500ms instead of default 1000ms
+}
+```
+
+#### Custom Value Comparison
+
+```java
+@Override
+protected boolean valuesEqual(PlcValue v1, PlcValue v2) {
+ // Custom comparison logic
+ // For example, treat small floating point differences as equal
+ if (v1 == v2) return true;
+ if (v1 == null || v2 == null) return false;
+
+ if (v1.isDouble() && v2.isDouble()) {
+ double diff = Math.abs(v1.getDouble() - v2.getDouble());
+ return diff < 0.001; // Treat differences < 0.001 as equal
+ }
+
+ return Objects.equals(v1.getObject(), v2.getObject());
+}
+```
+
+### Usage Examples
+
+#### CYCLIC Subscription
+
+```java
+PlcSubscriptionRequest request = DefaultPlcSubscriptionRequest.builder()
+ .addCyclicTagAddress("temperature", "MAIN.temperature",
Duration.ofMillis(100))
+ .addCyclicTagAddress("pressure", "MAIN.pressure", Duration.ofMillis(100))
+ .setConsumer(event -> {
+ System.out.println("Temperature: " + event.getInteger("temperature"));
+ System.out.println("Pressure: " + event.getInteger("pressure"));
+ })
+ .build();
+
+CompletableFuture<PlcSubscriptionResponse> future =
connection.subscribe(request);
+PlcSubscriptionResponse response = future.get();
+
+// Later, to unsubscribe:
+PlcUnsubscriptionRequest unsubRequest =
DefaultPlcUnsubscriptionRequest.builder()
+ .addHandles(response.getSubscriptionHandle("temperature"))
+ .addHandles(response.getSubscriptionHandle("pressure"))
+ .build();
+
+connection.unsubscribe(unsubRequest).get();
+```
+
+#### CHANGE_OF_STATE Subscription
+
+```java
+PlcSubscriptionRequest request = DefaultPlcSubscriptionRequest.builder()
+ .addChangeOfStateTagAddress("alarm", "MAIN.alarm", Duration.ofMillis(50))
+ .setConsumer(event -> {
+ System.out.println("Alarm state changed: " +
event.getBoolean("alarm"));
+ })
+ .build();
+
+connection.subscribe(request);
+```
+
+#### Tag-Specific Consumers
+
+```java
+PlcSubscriptionRequest request = DefaultPlcSubscriptionRequest.builder()
+ .addCyclicTagAddress("tag1", "MAIN.tag1", Duration.ofMillis(100))
+ .setTagConsumer("tag1", event -> {
+ System.out.println("Tag1 event: " + event.getInteger("tag1"));
+ })
+ .addCyclicTagAddress("tag2", "MAIN.tag2", Duration.ofMillis(100))
+ .setTagConsumer("tag2", event -> {
+ System.out.println("Tag2 event: " + event.getInteger("tag2"));
+ })
+ .build();
+
+connection.subscribe(request);
+```
+
+#### Consumer Registration
+
+```java
+// Subscribe
+PlcSubscriptionResponse response = connection.subscribe(request).get();
+PlcSubscriptionHandle handle1 = response.getSubscriptionHandle("tag1");
+PlcSubscriptionHandle handle2 = response.getSubscriptionHandle("tag2");
+
+// Register a consumer for specific handles
+PlcConsumerRegistration registration = connection.registerConsumer(
+ event -> {
+ System.out.println("Registered consumer received event");
+ },
+ Arrays.asList(handle1, handle2)
+);
+
+// Later, unregister
+connection.unregisterConsumer(registration);
+```
+
+### Performance Considerations
+
+- **Grouping**: Tags with the same subscription type and polling interval are
automatically grouped into a single batch, reducing overhead
+- **Concurrency**: All subscription management uses thread-safe concurrent
data structures
+- **Polling Intervals**: Choose appropriate polling intervals to balance
responsiveness and network/PLC load
+ - Too frequent polling can overload the PLC or network
+ - Too infrequent polling can miss value changes in CHANGE_OF_STATE
subscriptions
+
+### Requirements
+
+Your driver must:
+- Extend `PollingSubscriptionConnectionBase` instead of `ConnectionBase`
+- Implement the `PlcReader` interface
+- Add the `plc4j-utils-subscription-emulation` dependency
+
+### Limitations
+
+- **EVENT subscriptions are not supported** - Only CYCLIC and CHANGE_OF_STATE
are supported
+- **Polling overhead** - All subscriptions are emulated via polling, which may
not be as efficient as native protocol subscriptions
+- **Value change detection** - CHANGE_OF_STATE subscriptions compare values
using `Objects.equals()` by default, which may not be suitable for all data
types (e.g., floating point numbers with small differences)
+
+## Implementation Details
+
+For implementation details, see:
+- `PollingSubscriptionConnectionBase.java:73` - Main implementation
+- `EventPump` - The underlying polling engine (see
`java/tools/event-pump/README.md`)
diff --git a/plc4j/utils/subscription-emulation/pom.xml
b/plc4j/utils/subscription-emulation/pom.xml
new file mode 100644
index 0000000000..d45af62e96
--- /dev/null
+++ b/plc4j/utils/subscription-emulation/pom.xml
@@ -0,0 +1,147 @@
+<?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
+
+ 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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.1.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.1.0
http://maven.apache.org/xsd/maven-4.1.0.xsd";>
+
+ <parent>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-utils</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>plc4j-utils-subscription-emulation</artifactId>
+
+ <name>PLC4J: Utils: Subscription Emulation</name>
+
+ <properties>
+
<project.build.outputTimestamp>2026-07-16T15:00:59Z</project.build.outputTimestamp>
+ </properties>
+
+ <build>
+ <plugins>
+ <!-- Override JaCoCo coverage rules for this module -->
+ <!-- Subscription emulation has moderate coverage at 61% -->
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>check-coverage</id>
+ <phase>verify</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ <configuration>
+ <rules>
+ <rule implementation="org.jacoco.maven.RuleConfiguration">
+ <element>BUNDLE</element>
+ <limits>
+ <!-- Maintain 60% instruction coverage -->
+ <limit implementation="org.jacoco.report.check.Limit">
+ <counter>INSTRUCTION</counter>
+ <value>COVEREDRATIO</value>
+ <minimum>0.60</minimum>
+ </limit>
+ <!-- Allow up to 3 classes without coverage -->
+ <limit implementation="org.jacoco.report.check.Limit">
+ <counter>CLASS</counter>
+ <value>MISSEDCOUNT</value>
+ <maximum>3</maximum>
+ </limit>
+ </limits>
+ </rule>
+ </rules>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+ <!-- PLC4J API -->
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-api</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-spi-config</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-spi-drivers</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-transports-api</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-utils-audit-log-api</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-tools-event-pump</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ </dependency>
+
+ <!-- Logging -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+
+ <!-- Test Dependencies -->
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-spi-values</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-utils-test-utils</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git
a/plc4j/utils/subscription-emulation/src/main/java/org/apache/plc4x/java/utils/subscriptionemulation/PollingSubscriptionConnectionBase.java
b/plc4j/utils/subscription-emulation/src/main/java/org/apache/plc4x/java/utils/subscriptionemulation/PollingSubscriptionConnectionBase.java
new file mode 100644
index 0000000000..0a0f460ee6
--- /dev/null
+++
b/plc4j/utils/subscription-emulation/src/main/java/org/apache/plc4x/java/utils/subscriptionemulation/PollingSubscriptionConnectionBase.java
@@ -0,0 +1,593 @@
+/*
+ * 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.plc4x.java.utils.subscriptionemulation;
+
+import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.api.authentication.PlcAuthentication;
+import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
+import org.apache.plc4x.java.api.metadata.PlcConnectionMetadata;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionEvent;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse;
+import org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest;
+import org.apache.plc4x.java.api.messages.PlcUnsubscriptionResponse;
+import org.apache.plc4x.java.api.model.PlcTag;
+import org.apache.plc4x.java.api.model.PlcConsumerRegistration;
+import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
+import org.apache.plc4x.java.api.model.PlcSubscriptionTag;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
+import org.apache.plc4x.java.api.types.PlcSubscriptionType;
+import org.apache.plc4x.java.api.value.PlcValue;
+import org.apache.plc4x.java.spi.config.Configuration;
+import org.apache.plc4x.java.spi.drivers.ConnectionBase;
+import org.apache.plc4x.java.spi.drivers.functions.PlcSubscriber;
+import org.apache.plc4x.java.spi.drivers.messages.DefaultPlcSubscriptionEvent;
+import
org.apache.plc4x.java.spi.drivers.messages.DefaultPlcSubscriptionResponse;
+import
org.apache.plc4x.java.spi.drivers.messages.DefaultPlcUnsubscriptionResponse;
+import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem;
+import org.apache.plc4x.java.spi.drivers.messages.items.PlcResponseItem;
+import org.apache.plc4x.java.spi.drivers.model.DefaultPlcSubscriptionHandle;
+import org.apache.plc4x.java.spi.transports.api.TransportInstance;
+import org.apache.plc4x.java.tools.eventpump.EventPump;
+import org.apache.plc4x.java.tools.eventpump.TagBatch;
+import org.apache.plc4x.java.tools.eventpump.triggers.TimerTrigger;
+import org.apache.plc4x.java.utils.auditlog.api.AuditLog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+
+/**
+ * Extended base class for PLC connections that emulates subscriptions using
polling.
+ * <p>
+ * This class is designed for drivers that don't natively support
subscriptions (like Modbus).
+ * It uses the EventPump tool to poll tags at regular intervals and emulates:
+ * <ul>
+ * <li>CYCLIC subscriptions - polls at the specified interval</li>
+ * <li>CHANGE_OF_STATE subscriptions - polls and detects value changes</li>
+ * <li>EVENT subscriptions - not supported (throws
UnsupportedOperationException)</li>
+ * </ul>
+ * <p>
+ * To use this class, simply extend it instead of ConnectionBase and implement
PlcReader:
+ * <pre>
+ * public class MyConnection extends
PollingSubscriptionConnectionBase<MyConfig> implements PlcReader {
+ * // Your driver implementation
+ * }
+ * </pre>
+ *
+ * @param <C> the configuration type
+ */
+public abstract class PollingSubscriptionConnectionBase<C extends
Configuration> extends ConnectionBase<C> {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PollingSubscriptionConnectionBase.class);
+
+ private final EventPump eventPump = new EventPump();
+ private final AtomicLong subscriptionIdGenerator = new AtomicLong(1);
+ private final Map<Long, SubscriptionInfo> subscriptions = new
ConcurrentHashMap<>();
+ private final Map<PlcConsumerRegistration, ConsumerRegistrationInfo>
consumerRegistrations = new ConcurrentHashMap<>();
+ private final PlcConnectionManager singleConnectionManager;
+
+ public PollingSubscriptionConnectionBase(C configuration,
TransportInstance<?> transportInstance, AuditLog auditLog, String artifactId) {
+ super(configuration, transportInstance, auditLog);
+ // Create a connection manager that always returns this connection
+ this.singleConnectionManager = new SingleConnectionManager(this);
+ }
+
+ @Override
+ protected CompletableFuture<PlcSubscriptionResponse>
onSubscribe(PlcSubscriptionRequest subscriptionRequest) {
+ LinkedHashMap<String, PlcResponseItem<PlcSubscriptionHandle>>
responseItems = new LinkedHashMap<>();
+
+ try {
+ // Group tags by subscription type and polling interval
+ Map<SubscriptionKey, Map<String, PlcSubscriptionTag>> tagGroups =
groupTagsBySubscriptionType(subscriptionRequest);
+
+ // Create a batch for each group
+ for (Map.Entry<SubscriptionKey, Map<String, PlcSubscriptionTag>>
entry : tagGroups.entrySet()) {
+ SubscriptionKey key = entry.getKey();
+ Map<String, PlcSubscriptionTag> groupTags = entry.getValue();
+
+ Long subscriptionId =
subscriptionIdGenerator.getAndIncrement();
+ String batchId = "subscription-" + subscriptionId;
+
+ // Create tag batch with appropriate trigger
+ TimerTrigger trigger = createTrigger(key.pollingInterval);
+
+ // Convert tags into an address map
+ Map<String, String> tagAddresses = new HashMap<>();
+ groupTags.forEach((name, tag) -> tagAddresses.put(name,
tag.getAddressString()));
+
+ // Create subscription info for value change detection
+ SubscriptionInfo subscriptionInfo = new SubscriptionInfo(
+ subscriptionId,
+ batchId,
+ key.subscriptionType,
+ groupTags,
+ subscriptionRequest.getConsumer(),
+ new ConcurrentHashMap<>() // tag-specific consumers
+ );
+
+ // Add tag-specific consumers
+ groupTags.keySet().forEach(tagName -> {
+ Consumer<PlcSubscriptionEvent> tagConsumer =
subscriptionRequest.getTagConsumer(tagName);
+ if (tagConsumer != null) {
+ subscriptionInfo.tagConsumers.put(tagName,
tagConsumer);
+ }
+ });
+
+ TagBatch batch = TagBatch.builder()
+ .withBatchId(batchId)
+ .withConnectionManager(singleConnectionManager)
+ .withConnectionString("internal://subscription")
+ .addTagAddresses(tagAddresses)
+ .withTrigger(trigger)
+ .withListener((b, response) ->
handleSubscriptionEvent(subscriptionInfo, response))
+ .build();
+
+ eventPump.addBatch(batch);
+ eventPump.startBatch(batchId);
+
+ subscriptions.put(subscriptionId, subscriptionInfo);
+
+ // Create subscription handles for response
+ for (String tagName : groupTags.keySet()) {
+ PlcSubscriptionHandle handle = new
PollingSubscriptionHandle(this, subscriptionId, tagName);
+ responseItems.put(tagName, new
DefaultPlcResponseItem<>(PlcResponseCode.OK, handle));
+ }
+
+ LOGGER.debug("Created {} subscription batch '{}' with {} tags,
polling interval: {}ms",
+ key.subscriptionType, batchId, groupTags.size(),
key.pollingInterval);
+ }
+
+ return CompletableFuture.completedFuture(
+ new DefaultPlcSubscriptionResponse(subscriptionRequest,
responseItems));
+
+ } catch (Exception e) {
+ LOGGER.error("Error creating subscription", e);
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+
+ @Override
+ protected CompletableFuture<PlcUnsubscriptionResponse>
onUnsubscribe(PlcUnsubscriptionRequest unsubscriptionRequest) {
+ Set<Long> subscriptionsToRemove = new HashSet<>();
+
+ for (PlcSubscriptionHandle handle :
unsubscriptionRequest.getSubscriptionHandles()) {
+ if (handle instanceof PollingSubscriptionHandle pollingHandle) {
+ Long subscriptionId = pollingHandle.getSubscriptionId();
+ SubscriptionInfo info = subscriptions.get(subscriptionId);
+
+ if (info != null) {
+ subscriptionsToRemove.add(subscriptionId);
+ } else {
+ LOGGER.warn("Subscription not found for handle: {}",
handle);
+ }
+ } else {
+ LOGGER.warn("Invalid handle type: {}", handle.getClass());
+ }
+ }
+
+ // Stop and remove all identified subscriptions
+ for (Long subscriptionId : subscriptionsToRemove) {
+ SubscriptionInfo info = subscriptions.remove(subscriptionId);
+ if (info != null) {
+ eventPump.stopBatch(info.batchId);
+ eventPump.removeBatch(info.batchId);
+ LOGGER.debug("Unsubscribed from batch '{}'", info.batchId);
+ }
+ }
+
+ return CompletableFuture.completedFuture(
+ new DefaultPlcUnsubscriptionResponse(unsubscriptionRequest));
+ }
+
+ @Override
+ protected PlcConsumerRegistration
onRegisterConsumer(Consumer<PlcSubscriptionEvent> consumer,
Collection<PlcSubscriptionHandle> handles) {
+ List<PlcSubscriptionHandle> handleList = new ArrayList<>(handles);
+ int consumerId = (int) subscriptionIdGenerator.getAndIncrement();
+ PlcConsumerRegistration registration = new PlcConsumerRegistration() {
+ @Override
+ public void unregister() {
+ onUnregisterConsumer(this);
+ }
+
+ @Override
+ public List<PlcSubscriptionHandle> getSubscriptionHandles() {
+ return Collections.unmodifiableList(handleList);
+ }
+
+ @Override
+ public Integer getConsumerId() {
+ return consumerId;
+ }
+ };
+ ConsumerRegistrationInfo info = new ConsumerRegistrationInfo(consumer,
new HashSet<>(handleList));
+ consumerRegistrations.put(registration, info);
+
+ LOGGER.debug("Registered consumer for {} subscription handles",
handles.size());
+ return registration;
+ }
+
+ @Override
+ protected void onUnregisterConsumer(PlcConsumerRegistration registration) {
+ ConsumerRegistrationInfo removed =
consumerRegistrations.remove(registration);
+ if (removed != null) {
+ LOGGER.debug("Unregistered consumer");
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ // Stop all subscriptions
+ eventPump.close();
+ subscriptions.clear();
+ consumerRegistrations.clear();
+
+ super.close();
+ LOGGER.info("Closed polling subscription connection, all subscriptions
stopped");
+ }
+
+ /**
+ * Handles a subscription event from the event pump.
+ * For CHANGE_OF_STATE subscriptions, only fires if values have changed.
+ */
+ private void handleSubscriptionEvent(SubscriptionInfo info,
PlcReadResponse response) {
+ try {
+ boolean shouldFire = false;
+ LinkedHashMap<String, PlcResponseItem<PlcValue>> eventValues = new
LinkedHashMap<>();
+
+ for (String tagName : response.getTagNames()) {
+ PlcResponseCode code = response.getResponseCode(tagName);
+ if (code == PlcResponseCode.OK) {
+ PlcValue currentValue = response.getPlcValue(tagName);
+ PlcValue previousValue = info.previousValues.get(tagName);
+
+ if (info.subscriptionType == PlcSubscriptionType.CYCLIC) {
+ // For cyclic subscriptions, always fire
+ shouldFire = true;
+ eventValues.put(tagName, new
DefaultPlcResponseItem<>(code, currentValue));
+ info.previousValues.put(tagName, currentValue);
+
+ } else if (info.subscriptionType ==
PlcSubscriptionType.CHANGE_OF_STATE) {
+ // For change-of-state, only fire if value changed
+ if (previousValue == null ||
!valuesEqual(previousValue, currentValue)) {
+ shouldFire = true;
+ eventValues.put(tagName, new
DefaultPlcResponseItem<>(code, currentValue));
+ info.previousValues.put(tagName, currentValue);
+ LOGGER.trace("Value changed for tag '{}': {} ->
{}", tagName, previousValue, currentValue);
+ }
+ }
+ } else {
+ // Include error responses
+ eventValues.put(tagName, new
DefaultPlcResponseItem<>(code, null));
+ }
+ }
+
+ if (shouldFire && !eventValues.isEmpty()) {
+ PlcSubscriptionEvent event = new DefaultPlcSubscriptionEvent(
+ Instant.now(),
+ eventValues
+ );
+
+ // Fire to request-level consumer
+ if (info.requestConsumer != null) {
+ info.requestConsumer.accept(event);
+ }
+
+ // Fire to tag-specific consumers
+ for (String tagName : eventValues.keySet()) {
+ Consumer<PlcSubscriptionEvent> tagConsumer =
info.tagConsumers.get(tagName);
+ if (tagConsumer != null) {
+ tagConsumer.accept(event);
+ }
+ }
+
+ // Fire to registered consumers
+ for (ConsumerRegistrationInfo regInfo :
consumerRegistrations.values()) {
+ // Check if this registration includes any of our
subscription handles
+ for (PlcSubscriptionHandle handle : regInfo.handles) {
+ if (handle instanceof PollingSubscriptionHandle
pollingHandle) {
+ if
(pollingHandle.getSubscriptionId().equals(info.subscriptionId)) {
+ regInfo.consumer.accept(event);
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ } catch (Exception e) {
+ LOGGER.error("Error handling subscription event for batch '{}'",
info.batchId, e);
+ }
+ }
+
+ /**
+ * Groups subscription tags by type and polling interval.
+ */
+ private Map<SubscriptionKey, Map<String, PlcSubscriptionTag>>
groupTagsBySubscriptionType(PlcSubscriptionRequest request) {
+ Map<SubscriptionKey, Map<String, PlcSubscriptionTag>> groups = new
LinkedHashMap<>();
+
+ for (String tagName : request.getTagNames()) {
+ PlcSubscriptionTag tag = request.getTag(tagName);
+ PlcSubscriptionType type = tag.getPlcSubscriptionType();
+
+ // EVENT subscriptions are not supported
+ if (type == PlcSubscriptionType.EVENT) {
+ throw new UnsupportedOperationException(
+ "EVENT subscriptions are not supported with polling-based
subscriptions. " +
+ "Use CYCLIC or CHANGE_OF_STATE instead.");
+ }
+
+ long pollingInterval;
+ if (type == PlcSubscriptionType.CYCLIC &&
tag.getDuration().isPresent()) {
+ pollingInterval = tag.getDuration().get().toMillis();
+ } else {
+ // Default polling interval for CHANGE_OF_STATE or if not
specified
+ pollingInterval = getDefaultPollingInterval();
+ }
+
+ SubscriptionKey key = new SubscriptionKey(type, pollingInterval);
+ groups.computeIfAbsent(key, k -> new
LinkedHashMap<>()).put(tagName, tag);
+ }
+
+ return groups;
+ }
+
+ /**
+ * Creates a trigger based on the polling interval.
+ */
+ private TimerTrigger createTrigger(long pollingIntervalMs) {
+ return new TimerTrigger(pollingIntervalMs, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Compares two PlcValues for equality.
+ * Override this method if you need custom value comparison logic.
+ */
+ protected boolean valuesEqual(PlcValue v1, PlcValue v2) {
+ if (v1 == v2) return true;
+ if (v1 == null || v2 == null) return false;
+
+ // Compare as objects
+ return Objects.equals(v1.getObject(), v2.getObject());
+ }
+
+ /**
+ * Get the default polling interval for CHANGE_OF_STATE subscriptions.
+ * Override this method to provide a driver-specific default.
+ *
+ * @return Default polling interval in milliseconds (default: 1000ms)
+ */
+ protected long getDefaultPollingInterval() {
+ return 1000; // 1 second default
+ }
+
+ /**
+ * Key for grouping subscriptions by type and interval.
+ */
+ private static class SubscriptionKey {
+ final PlcSubscriptionType subscriptionType;
+ final long pollingInterval;
+
+ SubscriptionKey(PlcSubscriptionType subscriptionType, long
pollingInterval) {
+ this.subscriptionType = subscriptionType;
+ this.pollingInterval = pollingInterval;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof SubscriptionKey that)) return false;
+ return pollingInterval == that.pollingInterval && subscriptionType
== that.subscriptionType;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(subscriptionType, pollingInterval);
+ }
+ }
+
+ /**
+ * Information about an active subscription.
+ */
+ private static class SubscriptionInfo {
+ final Long subscriptionId;
+ final String batchId;
+ final PlcSubscriptionType subscriptionType;
+ final Map<String, PlcSubscriptionTag> tags;
+ final Consumer<PlcSubscriptionEvent> requestConsumer;
+ final Map<String, Consumer<PlcSubscriptionEvent>> tagConsumers;
+ final Map<String, PlcValue> previousValues = new ConcurrentHashMap<>();
+
+ SubscriptionInfo(Long subscriptionId,
+ String batchId,
+ PlcSubscriptionType subscriptionType,
+ Map<String, PlcSubscriptionTag> tags,
+ Consumer<PlcSubscriptionEvent> requestConsumer,
+ Map<String, Consumer<PlcSubscriptionEvent>>
tagConsumers) {
+ this.subscriptionId = subscriptionId;
+ this.batchId = batchId;
+ this.subscriptionType = subscriptionType;
+ this.tags = tags;
+ this.requestConsumer = requestConsumer;
+ this.tagConsumers = tagConsumers;
+ }
+ }
+
+ /**
+ * Information about a consumer registration.
+ */
+ private static class ConsumerRegistrationInfo {
+ final Consumer<PlcSubscriptionEvent> consumer;
+ final Set<PlcSubscriptionHandle> handles;
+
+ ConsumerRegistrationInfo(Consumer<PlcSubscriptionEvent> consumer,
Set<PlcSubscriptionHandle> handles) {
+ this.consumer = consumer;
+ this.handles = handles;
+ }
+ }
+
+ /**
+ * Custom subscription handle for polling-based subscriptions that stores
the subscription ID and tag name.
+ */
+ private static class PollingSubscriptionHandle extends
DefaultPlcSubscriptionHandle {
+ private final Long subscriptionId;
+ private final String tagName;
+
+ PollingSubscriptionHandle(PlcSubscriber subscriber, Long
subscriptionId, String tagName) {
+ super(subscriber);
+ this.subscriptionId = subscriptionId;
+ this.tagName = tagName;
+ }
+
+ public Long getSubscriptionId() {
+ return subscriptionId;
+ }
+
+ public String getTagName() {
+ return tagName;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof PollingSubscriptionHandle other)) {
+ return false;
+ }
+ return Objects.equals(subscriptionId, other.subscriptionId) &&
+ Objects.equals(tagName, other.tagName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(subscriptionId, tagName);
+ }
+
+ @Override
+ public String toString() {
+ return "PollingSubscriptionHandle{" +
+ "subscriptionId=" + subscriptionId +
+ ", tagName='" + tagName + '\'' +
+ '}';
+ }
+ }
+
+ /**
+ * Simple connection manager that always returns the same connection
wrapped in a non-closing proxy.
+ * This is used internally to adapt TagBatch's new API to work with a
single connection.
+ * The wrapper prevents TagBatch from closing the connection when using
try-with-resources.
+ */
+ private static class SingleConnectionManager implements
PlcConnectionManager {
+ private final PlcConnection wrappedConnection;
+
+ SingleConnectionManager(PlcConnection connection) {
+ // Wrap the connection to prevent it from being closed by TagBatch
+ this.wrappedConnection = new
NonClosingConnectionWrapper(connection);
+ }
+
+ @Override
+ public PlcConnection getConnection(String connectionString) throws
PlcConnectionException {
+ return wrappedConnection;
+ }
+
+ @Override
+ public PlcConnection getConnection(String connectionString,
PlcAuthentication authentication) throws PlcConnectionException {
+ return wrappedConnection;
+ }
+ }
+
+ /**
+ * Wrapper that delegates all calls to the underlying connection except
close().
+ * This prevents TagBatch from closing the connection when used in
try-with-resources.
+ */
+ private static class NonClosingConnectionWrapper implements PlcConnection {
+ private final PlcConnection delegate;
+
+ NonClosingConnectionWrapper(PlcConnection delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public void connect() {
+ // Already connected
+ }
+
+ @Override
+ public boolean isConnected() {
+ return delegate.isConnected();
+ }
+
+ @Override
+ public void close() {
+ // Intentionally do nothing - we don't want TagBatch to close the
underlying connection
+ }
+
+ @Override
+ public Optional<PlcTag> parseTagAddress(String tagAddress) {
+ return delegate.parseTagAddress(tagAddress);
+ }
+
+ @Override
+ public Optional<PlcValue> parseTagValue(PlcTag tag, Object... values) {
+ return delegate.parseTagValue(tag, values);
+ }
+
+ @Override
+ public CompletableFuture<? extends
org.apache.plc4x.java.api.messages.PlcPingResponse> ping() {
+ return delegate.ping();
+ }
+
+ @Override
+ public org.apache.plc4x.java.api.messages.PlcReadRequest.Builder
readRequestBuilder() {
+ return delegate.readRequestBuilder();
+ }
+
+ @Override
+ public org.apache.plc4x.java.api.messages.PlcWriteRequest.Builder
writeRequestBuilder() {
+ return delegate.writeRequestBuilder();
+ }
+
+ @Override
+ public
org.apache.plc4x.java.api.messages.PlcSubscriptionRequest.Builder
subscriptionRequestBuilder() {
+ return delegate.subscriptionRequestBuilder();
+ }
+
+ @Override
+ public
org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest.Builder
unsubscriptionRequestBuilder() {
+ return delegate.unsubscriptionRequestBuilder();
+ }
+
+ @Override
+ public org.apache.plc4x.java.api.messages.PlcBrowseRequest.Builder
browseRequestBuilder() {
+ return delegate.browseRequestBuilder();
+ }
+
+ @Override
+ public PlcConnectionMetadata getMetadata() {
+ return delegate.getMetadata();
+ }
+ }
+}
diff --git
a/plc4j/utils/subscription-emulation/src/test/java/org/apache/plc4x/java/utils/subscriptionemulation/PollingSubscriptionConnectionBaseTest.java
b/plc4j/utils/subscription-emulation/src/test/java/org/apache/plc4x/java/utils/subscriptionemulation/PollingSubscriptionConnectionBaseTest.java
new file mode 100644
index 0000000000..78bfd12e71
--- /dev/null
+++
b/plc4j/utils/subscription-emulation/src/test/java/org/apache/plc4x/java/utils/subscriptionemulation/PollingSubscriptionConnectionBaseTest.java
@@ -0,0 +1,305 @@
+/*
+ * 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.plc4x.java.utils.subscriptionemulation;
+
+import org.apache.plc4x.java.api.messages.*;
+import org.apache.plc4x.java.api.model.PlcTag;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
+import org.apache.plc4x.java.api.value.PlcValue;
+import org.apache.plc4x.java.spi.config.Configuration;
+import org.apache.plc4x.java.spi.drivers.messages.DefaultPlcReadResponse;
+import
org.apache.plc4x.java.spi.drivers.messages.DefaultPlcSubscriptionRequest;
+import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem;
+import org.apache.plc4x.java.spi.drivers.tags.PlcTagHandler;
+import org.apache.plc4x.java.spi.transports.api.TransportInstance;
+import org.apache.plc4x.java.spi.values.PlcValueHandler;
+import org.apache.plc4x.java.utils.auditlog.api.AuditLog;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * Tests for PollingSubscriptionConnectionBase.
+ */
+class PollingSubscriptionConnectionBaseTest {
+
+ private TestConnection connection;
+ private PlcTagHandler tagHandler;
+ private AtomicInteger mockValueCounter;
+
+ @BeforeEach
+ void setUp() {
+ mockValueCounter = new AtomicInteger(100);
+ tagHandler = mock(PlcTagHandler.class);
+
+ // Setup tag handler to return mock tags
+ PlcTag mockTag = mock(PlcTag.class);
+ when(mockTag.getAddressString()).thenReturn("MAIN.tag1");
+ when(tagHandler.parseTag(anyString())).thenReturn(mockTag);
+
+ connection = new TestConnection(mockValueCounter, tagHandler);
+ }
+
+ @Test
+ void testCyclicSubscriptionFiresRegularly() throws Exception {
+ // Arrange
+ CountDownLatch latch = new CountDownLatch(3);
+ List<Integer> receivedValues = Collections.synchronizedList(new
ArrayList<>());
+
+ PlcSubscriptionRequest request = new
DefaultPlcSubscriptionRequest.Builder(connection, tagHandler)
+ .addCyclicTagAddress("tag1", "MAIN.tag1", Duration.ofMillis(100))
+ .setConsumer(event -> {
+ PlcValue value = event.getPlcValue("tag1");
+ receivedValues.add(value.getInteger());
+ latch.countDown();
+ })
+ .build();
+
+ // Act
+ CompletableFuture<PlcSubscriptionResponse> subscribeFuture =
connection.subscribe(request);
+ PlcSubscriptionResponse response = subscribeFuture.get(1,
TimeUnit.SECONDS);
+
+ // Assert subscription was successful
+ assertEquals(PlcResponseCode.OK, response.getResponseCode("tag1"));
+ assertNotNull(response.getSubscriptionHandle("tag1"));
+
+ // Wait for multiple events (generous timeout for CI/heavy classpath
environments)
+ boolean completed = latch.await(2, TimeUnit.SECONDS);
+ assertTrue(completed, "Should have received at least 3 events");
+ assertTrue(receivedValues.size() >= 3, "Should have received at least
3 values");
+
+ // Cleanup
+ connection.close();
+ }
+
+ @Test
+ void testChangeOfStateSubscriptionOnlyFiresOnChange() throws Exception {
+ // Arrange
+ CountDownLatch latch = new CountDownLatch(2);
+ List<Integer> receivedValues = Collections.synchronizedList(new
ArrayList<>());
+
+ PlcSubscriptionRequest request = new
DefaultPlcSubscriptionRequest.Builder(connection, tagHandler)
+ .addChangeOfStateTagAddress("tag1", "MAIN.tag1")
+ .setConsumer(event -> {
+ PlcValue value = event.getPlcValue("tag1");
+ receivedValues.add(value.getInteger());
+ latch.countDown();
+ })
+ .build();
+
+ // Act
+ CompletableFuture<PlcSubscriptionResponse> subscribeFuture =
connection.subscribe(request);
+ PlcSubscriptionResponse response = subscribeFuture.get(1,
TimeUnit.SECONDS);
+
+ // Assert subscription was successful
+ assertEquals(PlcResponseCode.OK, response.getResponseCode("tag1"));
+
+ // Simulate value changes
+ Thread.sleep(200); // Wait for the first poll (value=100)
+ mockValueCounter.set(200); // Change value
+ Thread.sleep(300); // Wait for polls
+
+ boolean completed = latch.await(2, TimeUnit.SECONDS);
+ assertTrue(completed, "Should have received at least 2 change events");
+
+ // Should have received at least 2 different values
+ assertTrue(receivedValues.size() >= 2, "Should have at least 2
values");
+
+ // Cleanup
+ connection.close();
+ }
+
+ @Test
+ void testEventSubscriptionThrowsException() {
+ // Arrange
+ PlcSubscriptionRequest request = new
DefaultPlcSubscriptionRequest.Builder(connection, tagHandler)
+ .addEventTagAddress("tag1", "MAIN.tag1")
+ .build();
+
+ // Act & Assert
+ assertThrows(Exception.class, () -> {
+ CompletableFuture<PlcSubscriptionResponse> subscribeFuture =
connection.subscribe(request);
+ subscribeFuture.get(1, TimeUnit.SECONDS);
+ });
+
+ // Cleanup
+ assertDoesNotThrow(() -> connection.close());
+ }
+
+ @Test
+ void testConnectionWithoutReaderDoesNotFireEvents() throws Exception {
+ // Arrange - Create a connection that does NOT override onRead()
+ // Subscribe will succeed, but polling will fail because default
onRead() returns "not supported"
+ TestConnectionWithoutReader connection = new
TestConnectionWithoutReader(tagHandler);
+
+ AtomicInteger eventCount = new AtomicInteger(0);
+
+ PlcSubscriptionRequest request = new
DefaultPlcSubscriptionRequest.Builder(connection, tagHandler)
+ .addCyclicTagAddress("tag1", "MAIN.tag1", Duration.ofMillis(50))
+ .setConsumer(event -> eventCount.incrementAndGet())
+ .build();
+
+ // Act - subscribe should succeed
+ CompletableFuture<PlcSubscriptionResponse> subscribeFuture =
connection.subscribe(request);
+ PlcSubscriptionResponse response = subscribeFuture.get(1,
TimeUnit.SECONDS);
+ assertEquals(PlcResponseCode.OK, response.getResponseCode("tag1"));
+
+ // Wait a bit — events should NOT fire because onRead() returns "not
supported"
+ Thread.sleep(200);
+ assertEquals(0, eventCount.get(), "Should not receive events when
onRead is not overridden");
+
+ // Cleanup
+ assertDoesNotThrow(() -> connection.close());
+ }
+
+ @Test
+ void testCustomValueComparison() throws Exception {
+ // Create a connection with custom value comparison that ignores
changes
+ TestConnectionWithCustomComparison customConnection = new
TestConnectionWithCustomComparison(mockValueCounter, tagHandler);
+
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicInteger eventCount = new AtomicInteger(0);
+
+ PlcSubscriptionRequest request = new
DefaultPlcSubscriptionRequest.Builder(customConnection, tagHandler)
+ .addChangeOfStateTagAddress("tag1", "MAIN.tag1")
+ .setConsumer(event -> {
+ eventCount.incrementAndGet();
+ latch.countDown();
+ })
+ .build();
+
+ // Subscribe
+ CompletableFuture<PlcSubscriptionResponse> subscribeFuture =
customConnection.subscribe(request);
+ subscribeFuture.get(1, TimeUnit.SECONDS);
+
+ // Wait for first event
+ latch.await(2, TimeUnit.SECONDS);
+ assertEquals(1, eventCount.get(), "Should have received initial
event");
+
+ // Change value but custom comparator always returns true (equal)
+ mockValueCounter.set(999);
+ Thread.sleep(150);
+
+ // Should still only have 1 event because custom comparator says
values are equal
+ assertTrue(eventCount.get() <= 2, "Should not receive many events with
custom always-equal comparator");
+
+ // Cleanup
+ customConnection.close();
+ }
+
+ // Test stub implementation
+ static class TestConnection extends
PollingSubscriptionConnectionBase<TestConfiguration> {
+ private final AtomicInteger valueSupplier;
+ private final PlcTagHandler tagHandler;
+
+ public TestConnection(AtomicInteger valueSupplier, PlcTagHandler
tagHandler) {
+ super(new TestConfiguration(), mock(TransportInstance.class),
mock(AuditLog.class), "test");
+ this.valueSupplier = valueSupplier;
+ this.tagHandler = tagHandler;
+ }
+
+ @Override
+ protected CompletableFuture<PlcReadResponse> onRead(PlcReadRequest
readRequest) {
+ LinkedHashMap<String, DefaultPlcResponseItem<PlcValue>>
responseItems = new LinkedHashMap<>();
+
+ for (String tagName : readRequest.getTagNames()) {
+ PlcValue value = mock(PlcValue.class);
+ when(value.getInteger()).thenReturn(valueSupplier.get());
+ when(value.getObject()).thenReturn(valueSupplier.get());
+
+ responseItems.put(tagName, new
DefaultPlcResponseItem<>(PlcResponseCode.OK, value));
+ }
+
+ PlcReadResponse response = new DefaultPlcReadResponse(readRequest,
(Map) responseItems);
+ return CompletableFuture.completedFuture(response);
+ }
+
+ @Override
+ protected PlcTagHandler getTagHandler() {
+ return tagHandler;
+ }
+
+ @Override
+ protected PlcValueHandler getValueHandler() {
+ return null;
+ }
+
+ @Override
+ protected int getMaxConcurrentRequests() {
+ return 10;
+ }
+
+ @Override
+ protected long getDefaultPollingInterval() {
+ return 50; // Fast polling for tests
+ }
+ }
+
+ // Test connection that does NOT implement PlcReader
+ static class TestConnectionWithoutReader extends
PollingSubscriptionConnectionBase<TestConfiguration> {
+ private final PlcTagHandler tagHandler;
+
+ public TestConnectionWithoutReader(PlcTagHandler tagHandler) {
+ super(new TestConfiguration(), mock(TransportInstance.class),
mock(AuditLog.class), "test");
+ this.tagHandler = tagHandler;
+ }
+
+ @Override
+ protected PlcTagHandler getTagHandler() {
+ return tagHandler;
+ }
+
+ @Override
+ protected PlcValueHandler getValueHandler() {
+ return null;
+ }
+
+ @Override
+ protected int getMaxConcurrentRequests() {
+ return 10;
+ }
+ }
+
+ // Test connection with custom value comparison
+ static class TestConnectionWithCustomComparison extends TestConnection {
+ public TestConnectionWithCustomComparison(AtomicInteger valueSupplier,
PlcTagHandler tagHandler) {
+ super(valueSupplier, tagHandler);
+ }
+
+ @Override
+ protected boolean valuesEqual(PlcValue v1, PlcValue v2) {
+ // Always return true (values are always "equal")
+ return true;
+ }
+ }
+
+ // Test configuration
+ static class TestConfiguration implements Configuration {
+ }
+}
diff --git
a/plc4j/utils/subscription-emulation/src/test/resources/logback-test.xml
b/plc4j/utils/subscription-emulation/src/test/resources/logback-test.xml
new file mode 100644
index 0000000000..ae7fbb5130
--- /dev/null
+++ b/plc4j/utils/subscription-emulation/src/test/resources/logback-test.xml
@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+ <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} -
%msg%n</pattern>
+ </encoder>
+ </appender>
+
+ <!-- Set default log level to WARN for cleaner test output -->
+ <root level="WARN">
+ <appender-ref ref="STDOUT"/>
+ </root>
+
+ <!-- You can enable DEBUG for specific loggers if needed during debugging
-->
+ <!-- <logger name="org.apache.plc4x.java.utils.subscriptionemulation"
level="DEBUG"/> -->
+</configuration>
