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/07/16 13:45:00 UTC
(plc4x) 02/02: feat: Implemented a new component for fetching data from remote PLCs (replacement for the scraper)
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
commit 83023b00006470ad006af82540f0d2c2c478d19a
Author: Christofer Dutz <ch...@c-ware.de>
AuthorDate: Thu Jul 16 15:44:47 2026 +0200
feat: Implemented a new component for fetching data from remote PLCs (replacement for the scraper)
---
plc4j/tools/event-pump/README.md | 1289 ++++++++++++++++++++
plc4j/tools/{scraper-ng => event-pump}/pom.xml | 87 +-
.../plc4x/java/tools/eventpump/EventPump.java | 312 +++++
.../plc4x/java/tools/eventpump/TagBatch.java | 964 +++++++++++++++
.../eventpump/TransformedPlcReadResponse.java | 498 ++++++++
.../tools/eventpump/config/BatchConfiguration.java | 167 +++
.../eventpump/config/ConnectionConfiguration.java | 112 ++
.../eventpump/config/EventPumpConfiguration.java | 212 ++++
.../tools/eventpump/config/EventPumpFactory.java | 279 +++++
.../tools/eventpump/config/TagConfiguration.java | 149 +++
.../tools/eventpump/config/TagMapDeserializer.java | 70 ++
.../eventpump/config/TriggerConfiguration.java | 163 +++
.../transform/SimpleExpressionEvaluator.java | 467 +++++++
.../eventpump/transform/TransformException.java} | 25 +-
.../eventpump/transform/ValueTransformer.java | 69 ++
.../transform/ValueTransformerRegistry.java | 176 +++
.../eventpump/triggers/SubscriptionTrigger.java | 116 ++
.../tools/eventpump/triggers/TimerTrigger.java | 204 ++++
.../java/tools/eventpump/triggers/Trigger.java | 79 ++
.../plc4x/java/tools/eventpump/EventPumpTest.java | 464 +++++++
.../java/tools/eventpump/StubPlcConnection.java | 108 ++
.../java/tools/eventpump/TagBatchListenerTest.java | 101 ++
.../plc4x/java/tools/eventpump/TagBatchTest.java | 840 +++++++++++++
.../eventpump/TransformedPlcReadResponseTest.java | 882 ++++++++++++++
.../config/EventPumpConfigurationTest.java | 193 +++
.../eventpump/config/EventPumpFactoryTest.java | 452 +++++++
.../eventpump/config/TagConfigurationTest.java | 230 ++++
.../eventpump/config/TagMapDeserializerTest.java | 216 ++++
.../transform/SimpleExpressionEvaluatorTest.java | 248 ++++
.../transform/ValueTransformerRegistryTest.java | 221 ++++
.../eventpump/transform/ValueTransformerTest.java | 50 +
.../triggers/SubscriptionTriggerTest.java | 193 +++
.../tools/eventpump/triggers/TimerTriggerTest.java | 234 ++++
plc4j/tools/pom.xml | 1 +
34 files changed, 9815 insertions(+), 56 deletions(-)
diff --git a/plc4j/tools/event-pump/README.md b/plc4j/tools/event-pump/README.md
new file mode 100644
index 0000000000..6008bbc018
--- /dev/null
+++ b/plc4j/tools/event-pump/README.md
@@ -0,0 +1,1289 @@
+<!--
+ 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.
+ -->
+
+# Event Pump
+
+An event-driven system for fetching PLC tag data and delivering it to consumers.
+
+## Overview
+
+The Event Pump is a flexible component that allows you to fetch sets of tags from PLCs based on various triggers. One instance can manage multiple batches of tags from the same PLC or from multiple PLCs simultaneously.
+
+### Key Features
+
+- **Multiple Batch Support**: Fetch different sets of tags with different triggers
+- **Multi-PLC Capability**: Connect to and fetch from multiple PLCs simultaneously
+- **Flexible Triggers**:
+ - **Time-based**: Fetch at regular intervals
+ - **Subscription-based**: Fetch when a tag value changes
+ - **Custom**: Implement your own trigger logic
+- **Configuration Support**: Define batches in XML, JSON, or YAML
+- **Thread-Safe**: Concurrent batch management
+- **AutoCloseable**: Proper resource cleanup
+
+## Quick Start
+
+### Add Dependency
+
+```xml
+<dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-tools-event-pump</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+</dependency>
+```
+
+### Basic Usage (Programmatic)
+
+```java
+import org.apache.plc4x.java.tools.eventpump.*;
+import org.apache.plc4x.java.tools.eventpump.triggers.TimerTrigger;
+import org.apache.plc4x.java.tools.eventpump.triggers.Trigger;
+import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.spi.PlcDriverManager;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+// Create connection
+PlcConnectionManager connectionManager = PlcDriverManager.getDefault();
+PlcConnection connection = connectionManager.getConnection("ads:tcp://192.168.1.1");
+
+// Create a batch using the Builder pattern
+TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnection(connection)
+ .addTag("temperature", "MAIN.temperature")
+ .addTag("pressure", "MAIN.pressure")
+ .addTag("status", "MAIN.status")
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((batchRef, response) -> {
+ System.out.println("Temperature: " + response.getInteger("temperature"));
+ System.out.println("Pressure: " + response.getInteger("pressure"));
+ System.out.println("Status: " + response.getBoolean("status"));
+ })
+ .build();
+
+// Add to pump and start
+EventPump pump = new EventPump();
+pump.addBatch(batch);
+pump.startAll();
+
+// ... application runs ...
+
+// Cleanup
+pump.close();
+connection.close();
+```
+
+### Configuration-Based Usage
+
+Create a configuration file (YAML example):
+
+```yaml
+# config.yaml
+connections:
+ - id: plc1
+ url: "ads:tcp://192.168.1.1"
+ username: admin
+ password: secret
+
+batches:
+ - id: sensor-batch
+ connectionId: plc1
+ tags:
+ temperature: "MAIN.temperature"
+ pressure: "MAIN.pressure"
+ humidity: "MAIN.humidity"
+ trigger:
+ type: timer
+ intervalSeconds: 5
+ initialDelaySeconds: 0
+
+ - id: alarm-batch
+ connectionId: plc1
+ tags:
+ alarmStatus: "MAIN.alarmStatus"
+ alarmCode: "MAIN.alarmCode"
+ trigger:
+ type: subscription
+ tagName: alarmTrigger
+ tagAddress: "MAIN.alarmTrigger"
+```
+
+Load and use the configuration:
+
+```java
+import org.apache.plc4x.java.tools.eventpump.config.*;
+import java.io.File;
+
+// Load configuration
+PlcConnectionManager connectionManager = PlcDriverManager.getDefault();
+
+EventPump pump = EventPumpFactory.fromYaml(
+ new File("config.yaml"),
+ connectionManager,
+ (batch, response) -> {
+ System.out.println("Batch '" + batch.getBatchId() + "' fetched:");
+ response.getTagNames().forEach(tagName -> {
+ System.out.println(" " + tagName + ": " + response.getObject(tagName));
+ });
+ }
+);
+
+// Start all batches
+pump.startAll();
+
+// ... application runs ...
+
+// Cleanup
+pump.close();
+```
+
+## Core Concepts
+
+### EventPump
+
+The main component that manages multiple tag batches. Think of it as a container for batches.
+
+**Key Methods:**
+- `addBatch(TagBatch)`: Add a batch to the pump
+- `removeBatch(String)`: Remove a batch by ID
+- `getBatch(String)`: Get a specific batch by ID
+- `startBatch(String)`: Start a specific batch
+- `stopBatch(String)`: Stop a specific batch
+- `getAllBatches()`: Get all batches as a map
+- `getBatchIds()`: Get all batch IDs as a set
+- `getBatchCount()`: Get the number of batches
+- `getStartedBatchCount()`: Get the number of started batches
+- `startAll()`: Start all batches
+- `stopAll()`: Stop all batches
+- `close()`: Stop and cleanup all batches
+
+### TagBatch
+
+Represents a batch of tags to be fetched from a PLC. Each batch has:
+- A unique ID
+- A PLC connection
+- A map of tag names to tag addresses
+- A trigger that determines when to fetch
+- A listener that receives results
+
+**Creating a TagBatch:**
+
+Use the Builder pattern to create TagBatch instances:
+
+```java
+TagBatch batch = TagBatch.builder()
+ .withBatchId("my-batch")
+ .withConnection(connection)
+ .addTag("tag1", "MAIN.tag1")
+ .addTag("tag2", "MAIN.tag2")
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((b, response) -> {
+ // Handle results
+ })
+ .build();
+```
+
+**Key Methods:**
+- `start()`: Start fetching (starts the trigger)
+- `stop()`: Stop fetching (stops the trigger)
+- `fetchTags()`: Manually trigger a fetch (regardless of trigger state)
+- `setListener(TagBatchListener)`: Set/update the result handler (can also be set via builder)
+
+### Triggers
+
+Triggers determine when a batch should be fetched.
+
+#### TimerTrigger
+
+Fires at regular time intervals.
+
+```java
+// Fetch every 10 seconds
+Trigger trigger = new TimerTrigger(10, TimeUnit.SECONDS);
+
+// Fetch every 5 seconds, with a 2 second initial delay
+Trigger trigger = new TimerTrigger(5, 2, TimeUnit.SECONDS);
+```
+
+#### SubscriptionTrigger
+
+Fires when a subscribed tag value changes.
+
+```java
+// Trigger on any change
+Trigger trigger = new SubscriptionTrigger(
+ connection,
+ "triggerTag",
+ "MAIN.trigger"
+);
+
+// Trigger only when a value changes to true
+Trigger trigger = new SubscriptionTrigger(
+ connection,
+ "triggerTag",
+ "MAIN.trigger",
+ event -> event.getBoolean("triggerTag") == Boolean.TRUE
+);
+```
+
+#### Custom Triggers
+
+Implement the `Trigger` interface to create custom trigger logic:
+
+```java
+public class MyCustomTrigger implements Trigger {
+ // Implement trigger methods
+ @Override
+ public void start(TriggerListener listener) {
+ // Start your trigger logic
+ // Call listener.onTrigger(this) when it should fire
+ }
+
+ @Override
+ public void stop() {
+ // Stop your trigger logic
+ }
+
+ // ... other methods
+}
+```
+
+### Listeners
+
+The `TagBatchListener` interface handles fetch results. You can set a listener via the builder or using `setListener()`:
+
+```java
+// Option 1: Set listener via builder (recommended)
+TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnection(connection)
+ .addTag("temp", "MAIN.temperature")
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener(new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch batch, PlcReadResponse response) {
+ // Process successful fetch
+ for (String tagName : response.getTagNames()) {
+ Object value = response.getObject(tagName);
+ System.out.println(tagName + " = " + value);
+ }
+ }
+
+ @Override
+ public void onError(TagBatch batch, Throwable error) {
+ // Handle errors
+ System.err.println("Error in batch " + batch.getBatchId() + ": " + error.getMessage());
+ }
+ })
+ .build();
+
+// Option 2: Use a lambda (for simple cases)
+TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch2")
+ .withConnection(connection)
+ .addTag("pressure", "MAIN.pressure")
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((b, response) -> {
+ // Process results
+ System.out.println("Pressure: " + response.getObject("pressure"));
+ })
+ .build();
+
+// Option 3: Set/update listener after creation
+batch.setListener((b, response) -> {
+ // Process results
+});
+```
+
+## Configuration Formats
+
+The Event Pump supports configuration in three formats: YAML, JSON, and XML.
+
+### YAML Configuration
+
+```yaml
+connections:
+ - id: plc1
+ url: "ads:tcp://192.168.1.1"
+ - id: plc2
+ url: "modbus:tcp://192.168.1.2"
+
+batches:
+ - id: batch1
+ connectionId: plc1
+ tags:
+ temp1: "MAIN.temperature1"
+ temp2: "MAIN.temperature2"
+ trigger:
+ type: timer
+ intervalSeconds: 10
+
+ - id: batch2
+ connectionId: plc2
+ tags:
+ pressure: "40001"
+ trigger:
+ type: subscription
+ tagName: trigger
+ tagAddress: "40100"
+```
+
+### JSON Configuration
+
+```json
+{
+ "connections": [
+ {
+ "id": "plc1",
+ "url": "ads:tcp://192.168.1.1"
+ }
+ ],
+ "batches": [
+ {
+ "id": "batch1",
+ "connectionId": "plc1",
+ "tags": {
+ "temp1": "MAIN.temperature1",
+ "temp2": "MAIN.temperature2"
+ },
+ "trigger": {
+ "type": "timer",
+ "intervalSeconds": 10
+ }
+ }
+ ]
+}
+```
+
+### XML Configuration
+
+```xml
+<EventPumpConfiguration>
+ <connections>
+ <item>
+ <id>plc1</id>
+ <url>ads:tcp://192.168.1.1</url>
+ </item>
+ </connections>
+ <batches>
+ <item>
+ <id>batch1</id>
+ <connectionId>plc1</connectionId>
+ <tags>
+ <temp1>MAIN.temperature1</temp1>
+ <temp2>MAIN.temperature2</temp2>
+ </tags>
+ <trigger>
+ <type>timer</type>
+ <intervalSeconds>10</intervalSeconds>
+ </trigger>
+ </item>
+ </batches>
+</EventPumpConfiguration>
+```
+
+### Trigger Configuration
+
+#### Timer Trigger
+
+```yaml
+trigger:
+ type: timer
+ intervalSeconds: 5 # Required: interval in seconds
+ initialDelaySeconds: 2 # Optional: initial delay in seconds
+```
+
+#### Subscription Trigger
+
+```yaml
+trigger:
+ type: subscription
+ tagName: triggerTag # Required: name for the subscription
+ tagAddress: "MAIN.trigger" # Required: tag address to subscribe to
+ condition: "value == true" # Optional: condition expression (not yet implemented)
+```
+
+## Value Transformations
+
+The Event Pump supports optional value transformations, allowing you to transform raw PLC values using expressions before they reach your listener. This is useful for unit conversions, calculations, and data normalization.
+
+### Overview
+
+Value transformations are applied after values are read from the PLC, but before they're delivered to your listener. Each tag can have an optional transformation expression that manipulates the raw value.
+
+**Key Features:**
+- **Optional**: Transformations are completely optional - use them only where needed
+- **Per-Tag**: Each tag can have its own transformation or none at all
+- **Pluggable**: Default simple expression evaluator, but you can register custom transformers (e.g., SpEL for Spring applications)
+- **Context-Aware**: Transformations can access the current value and all other tag values in the batch
+
+### Basic Usage
+
+#### Extended Tag Format
+
+To use transformations, use the extended tag format in your configuration:
+
+```yaml
+batches:
+ - id: sensor-batch
+ connectionId: plc1
+ tags:
+ # Simple format (no transformation)
+ humidity: "MAIN.humidity"
+
+ # Extended format with transformation
+ temperature:
+ address: "MAIN.temperatureCelsius"
+ transform: "value * 1.8 + 32" # Convert Celsius to Fahrenheit
+
+ pressure:
+ address: "MAIN.pressureBar"
+ transform: "value * 14.5038" # Convert bar to PSI
+```
+
+#### Programmatic Usage
+
+```java
+TagBatch batch = TagBatch.builder()
+ .withBatchId("sensor-batch")
+ .withConnection(connection)
+ .addTag("humidity", "MAIN.humidity") // No transformation
+ .addTag("temperature", "MAIN.temperatureCelsius")
+ .addTransform("temperature", "value * 1.8 + 32") // Transform to Fahrenheit
+ .addTag("pressure", "MAIN.pressureBar")
+ .addTransform("pressure", "value * 14.5038") // Transform to PSI
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((b, response) -> {
+ // Values are already transformed
+ System.out.println("Temperature: " + response.getFloat("temperature") + "°F");
+ System.out.println("Pressure: " + response.getFloat("pressure") + " PSI");
+ })
+ .build();
+```
+
+### Expression Syntax
+
+The default `SimpleExpressionEvaluator` supports the following operations:
+
+#### Arithmetic Operators
+- `+` : Addition
+- `-` : Subtraction
+- `*` : Multiplication
+- `/` : Division
+- `%` : Modulo
+
+#### Comparison Operators
+- `==` : Equal to
+- `!=` : Not equal to
+- `<` : Less than
+- `>` : Greater than
+- `<=` : Less than or equal to
+- `>=` : Greater than or equal to
+
+#### Logical Operators
+- `&&` : Logical AND
+- `||` : Logical OR
+- `!` : Logical NOT
+
+#### Other Features
+- **Parentheses** for grouping: `(value + 10) * 2`
+- **Numeric literals**: Integer and floating-point numbers
+- **Boolean literals**: `true`, `false`
+- **Variable reference**: `value` (the current tag value)
+
+#### Operator Precedence (highest to lowest)
+1. Parentheses `( )`
+2. Unary operators `!`, `-`
+3. Multiplicative `*`, `/`, `%`
+4. Additive `+`, `-`
+5. Comparison `<`, `>`, `<=`, `>=`
+6. Equality `==`, `!=`
+7. Logical AND `&&`
+8. Logical OR `||`
+
+### Common Examples
+
+#### Temperature Conversion
+
+```yaml
+tags:
+ # Celsius to Fahrenheit
+ tempF:
+ address: "MAIN.tempCelsius"
+ transform: "value * 1.8 + 32"
+
+ # Fahrenheit to Celsius
+ tempC:
+ address: "MAIN.tempFahrenheit"
+ transform: "(value - 32) / 1.8"
+```
+
+#### Pressure Conversion
+
+```yaml
+tags:
+ # Bar to PSI
+ pressurePSI:
+ address: "MAIN.pressureBar"
+ transform: "value * 14.5038"
+
+ # kPa to Bar
+ pressureBar:
+ address: "MAIN.pressureKpa"
+ transform: "value / 100"
+```
+
+#### Scaling and Offset
+
+```yaml
+tags:
+ # Scale raw sensor value
+ scaledValue:
+ address: "MAIN.rawSensor"
+ transform: "value * 0.001" # Convert from millivolts to volts
+
+ # Apply offset and scale
+ calibratedValue:
+ address: "MAIN.rawValue"
+ transform: "(value - 100) * 2.5"
+```
+
+#### Boolean Transformations
+
+```yaml
+tags:
+ # Status to boolean
+ isRunning:
+ address: "MAIN.status"
+ transform: "value == 1"
+
+ # Alarm detection
+ alarmActive:
+ address: "MAIN.alarmCode"
+ transform: "value != 0"
+
+ # Within range check
+ withinRange:
+ address: "MAIN.pressure"
+ transform: "value >= 90 && value <= 110"
+```
+
+### Pluggable Transformers
+
+The transformation system is pluggable, allowing you to register custom expression evaluators. A Spring Expression Language (SpEL) transformer is included and **automatically registered** when Spring is detected on the classpath.
+
+#### Available Transformers
+
+By default, the following transformers are available:
+- `simple`: The default SimpleExpressionEvaluator (always registered)
+
+**Note**: The SpEL transformer is provided as an **optional** dependency. It will only be available if you add Spring Expression to your project dependencies.
+
+#### Spring Expression Language (SpEL)
+
+If Spring Expression is available on your classpath, the SpEL transformer is automatically registered and provides advanced expression capabilities:
+
+**Automatic Registration**: Add this dependency to enable SpEL:
+```xml
+<dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-expression</artifactId>
+ <version>6.2.15</version>
+</dependency>
+```
+
+**SpEL Syntax**: Variables must be accessed using `#variableName` syntax:
+
+```yaml
+tags:
+ tempF:
+ address: "MAIN.tempCelsius"
+ transform: "#value * 1.8 + 32" # Note the # prefix
+
+ average:
+ address: "MAIN.temp1"
+ transform: "(#temp1 + #temp2) / 2" # Multiple variables with #
+```
+
+**SpEL Features**:
+- Ternary operator: `#value > 10 ? #value : 0`
+- Method calls: `#value.toUpperCase()`, `T(java.lang.Math).sqrt(#value)`
+- String operations: `#value + ' suffix'`
+- Elvis operator: `#value ?: 100`
+- Complex logic: `#value > 50 && #value <= 100 ? 'normal' : 'abnormal'`
+
+#### Using SpEL as Default
+
+To use SpEL for all transformations by default:
+
+```java
+import org.apache.plc4x.java.tools.eventpump.transform.*;
+
+// Create a registry with default transformers (SpEL auto-registered if Spring is on the classpath)
+ValueTransformerRegistry registry = ValueTransformerRegistry.createDefault();
+
+// Use this registry for all batches in your EventPump
+TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString("ads:tcp://192.168.1.1")
+ .addTag("temp", "MAIN.temp")
+ .addTransform("temp", "#value * 1.8 + 32")
+ .withTransformerRegistry(registry) // Share registry across batches
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((b, r) -> handleResults(r))
+ .build();
+```
+
+#### Sharing Registry Across Multiple Batches
+
+For efficiency and consistency, you can share one registry instance across all batches:
+
+```java
+// Create one registry for the entire EventPump
+ValueTransformerRegistry registry = ValueTransformerRegistry.createDefault();
+
+// Use it for all batches
+TagBatch batch1 = TagBatch.builder()
+ .withBatchId("batch1")
+ .withTransformerRegistry(registry) // Share the same registry
+ // ... other configuration
+ .build();
+
+TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch2")
+ .withTransformerRegistry(registry) // Share the same registry
+ // ... other configuration
+ .build();
+```
+
+**Note**: When using `EventPumpFactory` to create batches from configuration files, a single registry is automatically created and shared across all batches in that EventPump.
+
+#### Checking Registered Transformers
+
+```java
+ValueTransformerRegistry registry = ValueTransformerRegistry.createDefault();
+
+// Get all registered transformer names
+Set<String> names = registry.getRegisteredNames();
+System.out.println("Available transformers: " + names);
+
+// Get the default transformer
+ValueTransformer defaultTransformer = registry.getDefault();
+System.out.println("Default transformer: " + defaultTransformer.getName());
+```
+
+#### Custom Transformers
+
+You can also implement your own transformers by implementing the `ValueTransformer` interface:
+
+```java
+public class MyCustomTransformer implements ValueTransformer {
+ @Override
+ public PlcValue transform(String expression, Map<String, PlcValue> context)
+ throws TransformException {
+ // Your custom transformation logic
+ return transformedValue;
+ }
+
+ @Override
+ public String getName() {
+ return "custom";
+ }
+}
+
+// Register your custom transformer
+ValueTransformerRegistry.getInstance().register(new MyCustomTransformer());
+```
+
+### Configuration Examples
+
+#### YAML with Transformations
+
+```yaml
+connections:
+ - id: plc1
+ url: "ads:tcp://192.168.1.1"
+
+batches:
+ - id: sensor-batch
+ connectionId: plc1
+ tags:
+ # Mix of simple and extended formats
+ humidity: "MAIN.humidity"
+
+ tempF:
+ address: "MAIN.tempCelsius"
+ transform: "value * 1.8 + 32"
+
+ pressurePSI:
+ address: "MAIN.pressureBar"
+ transform: "value * 14.5038"
+
+ # Extended format without transformation
+ status:
+ address: "MAIN.status"
+
+ trigger:
+ type: timer
+ intervalSeconds: 5
+```
+
+#### JSON with Transformations
+
+```json
+{
+ "batches": [
+ {
+ "id": "sensor-batch",
+ "connectionId": "plc1",
+ "tags": {
+ "humidity": "MAIN.humidity",
+ "tempF": {
+ "address": "MAIN.tempCelsius",
+ "transform": "value * 1.8 + 32"
+ },
+ "pressurePSI": {
+ "address": "MAIN.pressureBar",
+ "transform": "value * 14.5038"
+ }
+ },
+ "trigger": {
+ "type": "timer",
+ "intervalSeconds": 5
+ }
+ }
+ ]
+}
+```
+
+#### XML with Transformations
+
+```xml
+<batches>
+ <item>
+ <id>sensor-batch</id>
+ <connectionId>plc1</connectionId>
+ <tags>
+ <humidity>MAIN.humidity</humidity>
+ <tempF>
+ <address>MAIN.tempCelsius</address>
+ <transform>value * 1.8 + 32</transform>
+ </tempF>
+ <pressurePSI>
+ <address>MAIN.pressureBar</address>
+ <transform>value * 14.5038</transform>
+ </pressurePSI>
+ </tags>
+ <trigger>
+ <type>timer</type>
+ <intervalSeconds>5</intervalSeconds>
+ </trigger>
+ </item>
+</batches>
+```
+
+### Use Cases
+
+#### Industrial Monitoring
+
+```yaml
+batches:
+ - id: production-monitoring
+ connectionId: plc1
+ tags:
+ # Convert temperatures to preferred unit
+ ovenTempF:
+ address: "MAIN.ovenTempC"
+ transform: "value * 1.8 + 32"
+
+ # Calculate pressure deviation from setpoint
+ pressureDeviation:
+ address: "MAIN.actualPressure"
+ transform: "value - 100" # Assuming 100 is setpoint
+
+ # Normalize percentage
+ efficiency:
+ address: "MAIN.efficiencyRaw"
+ transform: "value / 100"
+```
+
+#### Energy Management
+
+```yaml
+batches:
+ - id: energy-monitoring
+ connectionId: plc1
+ tags:
+ # Convert watts to kilowatts
+ powerKW:
+ address: "MAIN.powerW"
+ transform: "value / 1000"
+
+ # Calculate power factor
+ powerFactor:
+ address: "MAIN.powerFactorPercent"
+ transform: "value / 100"
+
+ # Voltage scaling
+ voltageKV:
+ address: "MAIN.voltageV"
+ transform: "value / 1000"
+```
+
+### Error Handling
+
+If a transformation fails (e.g., division by zero, invalid expression), the batch's `onError()` method will be called:
+
+```java
+batch.setListener(new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch batch, PlcReadResponse response) {
+ // Process transformed values
+ }
+
+ @Override
+ public void onError(TagBatch batch, Throwable error) {
+ if (error instanceof TransformException) {
+ System.err.println("Transformation error: " + error.getMessage());
+ // Handle transformation error
+ } else {
+ // Handle other errors
+ }
+ }
+});
+```
+
+## Advanced Usage
+
+### Listing Batches
+
+You can get a list of all batch IDs to iterate or display them:
+
+```java
+EventPump pump = new EventPump();
+
+// Add some batches
+pump.addBatch(batch1);
+pump.addBatch(batch2);
+pump.addBatch(batch3);
+
+// Get all batch IDs
+java.util.Set<String> batchIds = pump.getBatchIds();
+
+// Print batch information
+System.out.println("Available batches:");
+for (String batchId : batchIds) {
+ TagBatch batch = pump.getBatch(batchId);
+ System.out.println(" - " + batchId +
+ " (started: " + batch.isStarted() +
+ ", tags: " + batch.getTags().size() + ")");
+}
+
+// Or check if a specific batch exists
+if (batchIds.contains("sensor-batch")) {
+ pump.startBatch("sensor-batch");
+}
+```
+
+### Multiple Batches, One PLC
+
+```java
+EventPump pump = new EventPump();
+
+// Batch 1: Fast polling of critical values
+TagBatch criticalBatch = TagBatch.builder()
+ .withBatchId("critical")
+ .withConnection(connection)
+ .addTag("emergency", "MAIN.emergencyStop")
+ .addTag("alarm", "MAIN.alarm")
+ .withTrigger(new TimerTrigger(1, TimeUnit.SECONDS)) // Every second
+ .withListener((b, r) -> handleCritical(r))
+ .build();
+
+// Batch 2: Slow polling of non-critical values
+TagBatch statusBatch = TagBatch.builder()
+ .withBatchId("status")
+ .withConnection(connection)
+ .addTag("uptime", "MAIN.uptime")
+ .addTag("cycleCount", "MAIN.cycleCount")
+ .withTrigger(new TimerTrigger(30, TimeUnit.SECONDS)) // Every 30 seconds
+ .withListener((b, r) -> handleStatus(r))
+ .build();
+
+pump.addBatch(criticalBatch);
+pump.addBatch(statusBatch);
+pump.startAll();
+```
+
+### Multiple PLCs
+
+```java
+EventPump pump = new EventPump();
+
+PlcConnection plc1 = connectionManager.getConnection("ads:tcp://192.168.1.1");
+PlcConnection plc2 = connectionManager.getConnection("modbus:tcp://192.168.1.2");
+PlcConnection plc3 = connectionManager.getConnection("s7://192.168.1.3");
+
+// Batch from PLC 1 (ADS)
+TagBatch batch1 = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnection(plc1)
+ .addTag("temp", "MAIN.temperature")
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((b, r) -> handlePlc1(r))
+ .build();
+
+// Batch from PLC 2 (Modbus)
+TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch2")
+ .withConnection(plc2)
+ .addTag("pressure", "40001")
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((b, r) -> handlePlc2(r))
+ .build();
+
+// Batch from PLC 3 (S7)
+TagBatch batch3 = TagBatch.builder()
+ .withBatchId("batch3")
+ .withConnection(plc3)
+ .addTag("flow", "%DB1.DBD0:REAL")
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((b, r) -> handlePlc3(r))
+ .build();
+
+pump.addBatch(batch1);
+pump.addBatch(batch2);
+pump.addBatch(batch3);
+pump.startAll();
+```
+
+### Event-Driven Fetching
+
+Use subscription triggers to fetch data only when needed:
+
+```java
+// Subscribe to a trigger tag
+SubscriptionTrigger trigger = new SubscriptionTrigger(
+ connection,
+ "productionStarted",
+ "MAIN.productionStarted"
+);
+
+// When productionStarted changes, fetch the batch
+TagBatch batch = TagBatch.builder()
+ .withBatchId("production-batch")
+ .withConnection(connection)
+ .addTag("productCount", "MAIN.productCount")
+ .addTag("batchNumber", "MAIN.batchNumber")
+ .addTag("startTime", "MAIN.startTime")
+ .withTrigger(trigger)
+ .withListener((b, response) -> {
+ // Process production data only when production starts
+ logProductionData(response);
+ })
+ .build();
+```
+
+### Manual Fetching
+
+You can manually trigger a fetch at any time:
+
+```java
+TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnection(connection)
+ .addTag("value1", "MAIN.value1")
+ .addTag("value2", "MAIN.value2")
+ .withTrigger(new TimerTrigger(10, TimeUnit.SECONDS))
+ .withListener((b, r) -> handleResults(r))
+ .build();
+
+pump.addBatch(batch);
+
+// Start the batch (trigger will fire automatically)
+pump.startBatch("batch1");
+
+// Manually fetch outside of trigger schedule
+batch.fetchTags().thenRun(() -> {
+ System.out.println("Manual fetch completed");
+});
+```
+
+## Error Handling
+
+### Batch-Level Error Handling
+
+```java
+batch.setListener(new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch batch, PlcReadResponse response) {
+ // Check individual tag responses
+ for (String tagName : response.getTagNames()) {
+ if (response.getResponseCode(tagName) == PlcResponseCode.OK) {
+ Object value = response.getObject(tagName);
+ // Process value
+ } else {
+ System.err.println("Error reading tag " + tagName + ": " +
+ response.getResponseCode(tagName));
+ }
+ }
+ }
+
+ @Override
+ public void onError(TagBatch batch, Throwable error) {
+ System.err.println("Batch error: " + error.getMessage());
+ error.printStackTrace();
+
+ // Optional: Implement retry logic
+ // Optional: Send alert
+ // Optional: Stop batch if errors persist
+ }
+});
+```
+
+### Connection Error Handling
+
+If a connection fails, batches using that connection will receive errors via `onError()`:
+
+```java
+batch.setListener(new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch batch, PlcReadResponse response) {
+ // Process results
+ }
+
+ @Override
+ public void onError(TagBatch batch, Throwable error) {
+ if (error instanceof PlcConnectionException) {
+ // Connection lost - attempt reconnection
+ handleConnectionLoss(batch);
+ } else {
+ // Other error - log and continue
+ logger.error("Batch error", error);
+ }
+ }
+});
+```
+
+## Best Practices
+
+### 1. Use Try-With-Resources
+
+```java
+// Good - automatic cleanup
+try (EventPump pump = new EventPump()) {
+ pump.addBatch(batch1);
+ pump.addBatch(batch2);
+ pump.startAll();
+
+ // Run for some time
+ Thread.sleep(60000);
+
+} // Automatic cleanup on exit
+```
+
+### 2. Set Appropriate Intervals
+
+Consider your use case when setting timer intervals:
+
+```java
+// For critical monitoring: short intervals
+new TimerTrigger(1, TimeUnit.SECONDS);
+
+// For status updates: medium intervals
+new TimerTrigger(30, TimeUnit.SECONDS);
+
+// For historical logging: long intervals
+new TimerTrigger(5, TimeUnit.MINUTES);
+```
+
+### 3. Batch Related Tags Together
+
+Group tags that are logically related and need the same trigger:
+
+```java
+// Good - temperature sensors together
+Map<String, String> tempTags = new HashMap<>();
+tempTags.put("temp1", "MAIN.temperature1");
+tempTags.put("temp2", "MAIN.temperature2");
+tempTags.put("temp3", "MAIN.temperature3");
+
+// Bad - mixing unrelated tags with different update needs
+Map<String, String> mixedTags = new HashMap<>();
+mixedTags.put("temp1", "MAIN.temperature1"); // Updates every second
+mixedTags.put("uptime", "MAIN.uptime"); // Updates every hour
+```
+
+### 4. Reuse Connections
+
+Don't create multiple connections to the same PLC:
+
+```java
+// Good - one connection, multiple batches
+PlcConnection connection = connectionManager.getConnection("ads:tcp://192.168.1.1");
+
+TagBatch batch1 = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnection(connection)
+ .addTag("temp", "MAIN.temp")
+ .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ .withListener((b, r) -> handleBatch1(r))
+ .build();
+
+TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch2")
+ .withConnection(connection) // Reuse same connection
+ .addTag("pressure", "MAIN.pressure")
+ .withTrigger(new TimerTrigger(10, TimeUnit.SECONDS))
+ .withListener((b, r) -> handleBatch2(r))
+ .build();
+
+// Bad - multiple connections to same PLC
+PlcConnection conn1 = connectionManager.getConnection("ads:tcp://192.168.1.1");
+PlcConnection conn2 = connectionManager.getConnection("ads:tcp://192.168.1.1");
+```
+
+### 5. Monitor Performance
+
+```java
+batch.setListener((batch, response) -> {
+ long fetchTime = System.currentTimeMillis();
+ // Process response
+
+ // Log slow fetches
+ long duration = System.currentTimeMillis() - fetchTime;
+ if (duration > 1000) {
+ logger.warn("Slow fetch in batch {}: {}ms", batch.getBatchId(), duration);
+ }
+});
+```
+
+## Limitations
+
+1. **Subscription Support**: Not all PLC protocols support subscriptions. Check protocol capabilities before using `SubscriptionTrigger`.
+
+2. **Expression Conditions**: Advanced expression-based conditions for subscription triggers are not yet implemented. Use predicate functions instead.
+
+3. **SimpleExpressionEvaluator Limitations**: The default expression evaluator does not support:
+ - Ternary operator (`condition ? true : false`)
+ - Function calls (e.g., `Math.sqrt(value)`)
+ - String operations
+ - Cross-tag references in expressions (access to other tag values)
+
+ For advanced expression needs, register a custom transformer like SpEL (see Value Transformations section).
+
+4. **No Built-in Retry Logic**: If a fetch fails, the batch continues on the next trigger. Implement custom retry logic in `onError()` if needed.
+
+5. **No Data Buffering**: Results are delivered immediately to listeners. If you need buffering, implement it in your listener.
+
+6. **No Automatic Reconnection**: If a connection fails, batches will error until the connection is restored. Implement reconnection logic in your application if needed.
+
+## Testing
+
+Run the tests:
+
+```bash
+mvn test
+```
+
+The module includes comprehensive tests for:
+- TimerTrigger functionality
+- EventPump management
+- Configuration serialization/deserialization
+- Value transformations (SimpleExpressionEvaluator)
+- Transformation registry and pluggable transformers
+- TransformedPlcReadResponse wrapper
+- Extended tag configuration format
+
+## Troubleshooting
+
+### Batch Not Firing
+
+**Symptom**: Batch never fetches data
+
+**Possible Causes**:
+1. Batch not started: Call `pump.startBatch(id)` or `pump.startAll()`
+2. No listener set: Call `batch.setListener()` before starting
+3. Trigger not configured correctly
+
+**Solution**:
+```java
+// Verify batch is started
+System.out.println("Started: " + batch.isStarted());
+
+// Verify trigger is running
+System.out.println("Trigger running: " + batch.getTrigger().isRunning());
+
+// Test manual fetch
+batch.fetchTags().get(); // Throws exception if connection/tags are wrong
+```
+
+### Subscription Trigger Not Working
+
+**Symptom**: SubscriptionTrigger never fires
+
+**Possible Causes**:
+1. Protocol doesn't support subscriptions
+2. Tag address is incorrect
+3. Tag value not changing
+
+**Solution**:
+```java
+// Test if protocol supports subscriptions
+try {
+ PlcSubscriptionRequest request = connection.subscriptionRequestBuilder()
+ .addChangeOfStateTagAddress("test", "MAIN.test")
+ .build();
+ // If this throws UnsupportedOperationException, subscriptions are not supported
+} catch (UnsupportedOperationException e) {
+ System.err.println("Protocol does not support subscriptions");
+}
+```
+
+### High CPU Usage
+
+**Symptom**: CPU usage is high
+
+**Possible Causes**:
+1. Too many batches
+2. Timer intervals are too short
+3. Too many tags per batch
+
+**Solution**:
+```java
+// Increase timer intervals
+new TimerTrigger(5, TimeUnit.SECONDS); // Instead of 100ms
+
+// Reduce number of batches
+// Combine related batches
+
+// Monitor batch count
+System.out.println("Active batches: " + pump.getStartedBatchCount());
+```
+
+## See Also
+
+- [Connection Cache](../connection-cache/README.md) - For managing PLC connection pooling
+- [Apache PLC4X Documentation](https://plc4x.apache.org/) - For PLC protocol details
diff --git a/plc4j/tools/scraper-ng/pom.xml b/plc4j/tools/event-pump/pom.xml
similarity index 51%
rename from plc4j/tools/scraper-ng/pom.xml
rename to plc4j/tools/event-pump/pom.xml
index 558f9534df..e4fa2af2d2 100644
--- a/plc4j/tools/scraper-ng/pom.xml
+++ b/plc4j/tools/event-pump/pom.xml
@@ -17,9 +17,8 @@
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">
-
- <modelVersion>4.1.0</modelVersion>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.plc4x</groupId>
@@ -27,70 +26,74 @@
<version>1.0.0-SNAPSHOT</version>
</parent>
- <artifactId>plc4j-scraper-ng</artifactId>
+ <artifactId>plc4j-tools-event-pump</artifactId>
- <name>PLC4J: Tools: Scraper (NG)</name>
- <description>Utility to efficiently collect a large number of items on multiple devices by different triggers.</description>
+ <name>PLC4J: Tools: Event-Pump</name>
+ <description>An event-based system for fetching PLC data and pumping it to a consumer.</description>
<properties>
- <project.build.outputTimestamp>2025-08-02T13:55:11Z</project.build.outputTimestamp>
+ <project.build.outputTimestamp>2026-07-16T15:00:59Z</project.build.outputTimestamp>
</properties>
<dependencies>
- <!-- Apache Commons -->
- <!--dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-math3</artifactId>
- </dependency>
+ <!-- PLC4J API -->
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-pool2</artifactId>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-api</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
</dependency>
+
+ <!-- SPI Values (for value transformation) -->
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-spi-values</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
</dependency>
+
+ <!-- Logging -->
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-collections4</artifactId>
- </dependency-->
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
- <!-- PLC4X dependencies -->
- <!--dependency>
- <groupId>org.apache.plc4x</groupId>
- <artifactId>plc4j-api</artifactId>
- <version>0.12.0-SNAPSHOT</version>
+ <!-- Configuration parsing (Jackson for JSON/YAML) -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.plc4x</groupId>
- <artifactId>plc4j-connection-pool</artifactId>
- <version>0.12.0-SNAPSHOT</version>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.plc4x</groupId>
- <artifactId>plc4j-connection-cache</artifactId>
- <version>0.12.0-SNAPSHOT</version>
- </dependency-->
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.dataformat</groupId>
+ <artifactId>jackson-dataformat-yaml</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.dataformat</groupId>
+ <artifactId>jackson-dataformat-xml</artifactId>
+ </dependency>
+ <!-- Testing -->
<dependency>
- <groupId>org.apache.plc4x</groupId>
- <artifactId>plc4j-spi</artifactId>
- <version>1.0.0-SNAPSHOT</version>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
<scope>test</scope>
</dependency>
-
- <!-- Test dependencies-->
<dependency>
- <groupId>org.awaitility</groupId>
- <artifactId>awaitility</artifactId>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
<scope>test</scope>
</dependency>
<dependency>
- <groupId>org.apache.plc4x</groupId>
- <artifactId>plc4j-driver-mock</artifactId>
- <version>1.0.0-SNAPSHOT</version>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/EventPump.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/EventPump.java
new file mode 100644
index 0000000000..bd52664448
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/EventPump.java
@@ -0,0 +1,312 @@
+/*
+ * 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.tools.eventpump;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * An event-driven system for fetching PLC tag data and delivering it to consumers.
+ * <p>
+ * The EventPump manages multiple {@link TagBatch}es, each of which:
+ * - Connects to one or more PLCs
+ * - Defines a set of tags to fetch
+ * - Has a trigger that determines when to fetch
+ * - Delivers results to a listener
+ * <p>
+ * One EventPump instance can manage batches from multiple PLCs, and multiple
+ * batches from the same PLC with different triggers.
+ * <p>
+ * Features:
+ * - Multiple batches per PLC
+ * - Multiple PLCs per pump
+ * - Different triggers per batch (timer, subscription, etc.)
+ * - Thread-safe batch management
+ * - Automatic lifecycle management
+ * <p>
+ * Example usage:
+ * <pre>
+ * EventPump pump = new EventPump();
+ *
+ * // Add a batch that fetches every 5 seconds
+ * TagBatch batch1 = new TagBatch("batch1", connection, tags, new TimerTrigger(5, TimeUnit.SECONDS));
+ * batch1.setListener((batch, response) -> {
+ * // Process results
+ * });
+ * pump.addBatch(batch1);
+ *
+ * // Add a batch triggered by a tag change
+ * TagBatch batch2 = new TagBatch("batch2", connection, tags2, new SubscriptionTrigger(connection, "trigger", "MAIN.trigger"));
+ * batch2.setListener((batch, response) -> {
+ * // Process results
+ * });
+ * pump.addBatch(batch2);
+ *
+ * // Start all batches
+ * pump.startAll();
+ *
+ * // Stop when done
+ * pump.close();
+ * </pre>
+ */
+public class EventPump implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EventPump.class);
+
+ // Thread-safe map of batch ID -> TagBatch
+ private final ConcurrentHashMap<String, TagBatch> batches = new ConcurrentHashMap<>();
+
+ private volatile boolean closed = false;
+
+ /**
+ * Create a new EventPump.
+ */
+ public EventPump() {
+ LOGGER.debug("Created EventPump");
+ }
+
+ /**
+ * Add a batch to the pump.
+ * The batch will not start automatically - call {@link #startBatch(String)} or {@link #startAll()}.
+ *
+ * @param batch The batch to add
+ * @throws IllegalArgumentException if a batch with the same ID already exists
+ * @throws IllegalStateException if the pump is closed
+ */
+ public void addBatch(TagBatch batch) {
+ if (batch == null) {
+ throw new IllegalArgumentException("Batch cannot be null");
+ }
+ if (closed) {
+ throw new IllegalStateException("EventPump is closed");
+ }
+
+ String batchId = batch.getBatchId();
+
+ TagBatch existing = batches.putIfAbsent(batchId, batch);
+ if (existing != null) {
+ throw new IllegalArgumentException("Batch with ID '" + batchId + "' already exists");
+ }
+
+ LOGGER.info("Added batch '{}' to EventPump", batchId);
+ }
+
+ /**
+ * Remove a batch from the pump.
+ * The batch will be stopped and closed.
+ *
+ * @param batchId The ID of the batch to remove
+ * @return The removed batch, or null if not found
+ */
+ public TagBatch removeBatch(String batchId) {
+ if (batchId == null) {
+ throw new IllegalArgumentException("Batch ID cannot be null");
+ }
+
+ TagBatch batch = batches.remove(batchId);
+ if (batch != null) {
+ try {
+ batch.close();
+ LOGGER.info("Removed batch '{}' from EventPump", batchId);
+ } catch (Exception e) {
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error closing removed batch '{}'", batchId, e);
+ } else {
+ LOGGER.error("Error closing removed batch '{}': {}", batchId, e.getMessage());
+ }
+ }
+ }
+
+ return batch;
+ }
+
+ /**
+ * Get a batch by ID.
+ *
+ * @param batchId The batch ID
+ * @return The batch, or null if not found
+ */
+ public TagBatch getBatch(String batchId) {
+ return batches.get(batchId);
+ }
+
+ /**
+ * Get all batches.
+ *
+ * @return An unmodifiable map of batch ID -> TagBatch
+ */
+ public Map<String, TagBatch> getAllBatches() {
+ return Collections.unmodifiableMap(batches);
+ }
+
+ /**
+ * Get all batch IDs.
+ *
+ * @return An unmodifiable set of batch IDs
+ */
+ public java.util.Set<String> getBatchIds() {
+ return Collections.unmodifiableSet(batches.keySet());
+ }
+
+ /**
+ * Start a specific batch.
+ *
+ * @param batchId The ID of the batch to start
+ * @throws IllegalArgumentException if the batch doesn't exist
+ * @throws IllegalStateException if the pump is closed
+ */
+ public void startBatch(String batchId) {
+ if (closed) {
+ throw new IllegalStateException("EventPump is closed");
+ }
+
+ TagBatch batch = batches.get(batchId);
+ if (batch == null) {
+ throw new IllegalArgumentException("Batch '" + batchId + "' not found");
+ }
+
+ batch.start();
+ LOGGER.info("Started batch '{}'", batchId);
+ }
+
+ /**
+ * Stop a specific batch.
+ *
+ * @param batchId The ID of the batch to stop
+ * @throws IllegalArgumentException if the batch doesn't exist
+ */
+ public void stopBatch(String batchId) {
+ TagBatch batch = batches.get(batchId);
+ if (batch == null) {
+ throw new IllegalArgumentException("Batch '" + batchId + "' not found");
+ }
+
+ batch.stop();
+ LOGGER.info("Stopped batch '{}'", batchId);
+ }
+
+ /**
+ * Start all batches.
+ *
+ * @throws IllegalStateException if the pump is closed
+ */
+ public void startAll() {
+ if (closed) {
+ throw new IllegalStateException("EventPump is closed");
+ }
+
+ batches.values().forEach(batch -> {
+ if (!batch.isStarted()) {
+ try {
+ batch.start();
+ } catch (Exception e) {
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error starting batch '{}'", batch.getBatchId(), e);
+ } else {
+ LOGGER.error("Error starting batch '{}': {}", batch.getBatchId(), e.getMessage());
+ }
+ }
+ }
+ });
+
+ LOGGER.info("Started all batches ({} total)", batches.size());
+ }
+
+ /**
+ * Stop all batches.
+ */
+ public void stopAll() {
+ batches.values().forEach(batch -> {
+ if (batch.isStarted()) {
+ try {
+ batch.stop();
+ } catch (Exception e) {
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error stopping batch '{}'", batch.getBatchId(), e);
+ } else {
+ LOGGER.error("Error stopping batch '{}': {}", batch.getBatchId(), e.getMessage());
+ }
+ }
+ }
+ });
+
+ LOGGER.info("Stopped all batches ({} total)", batches.size());
+ }
+
+ /**
+ * Get the number of batches in the pump.
+ *
+ * @return The number of batches
+ */
+ public int getBatchCount() {
+ return batches.size();
+ }
+
+ /**
+ * Get the number of started batches.
+ *
+ * @return The number of started batches
+ */
+ public int getStartedBatchCount() {
+ return (int) batches.values().stream()
+ .filter(TagBatch::isStarted)
+ .count();
+ }
+
+ /**
+ * Check if the pump is closed.
+ *
+ * @return true if closed, false otherwise
+ */
+ public boolean isClosed() {
+ return closed;
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+
+ LOGGER.info("Closing EventPump with {} batches", batches.size());
+
+ // Stop and close all batches
+ batches.values().forEach(batch -> {
+ try {
+ batch.close();
+ } catch (Exception e) {
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error closing batch '{}'", batch.getBatchId(), e);
+ } else {
+ LOGGER.error("Error closing batch '{}': {}", batch.getBatchId(), e.getMessage());
+ }
+ }
+ });
+
+ batches.clear();
+ closed = true;
+
+ LOGGER.info("EventPump closed");
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TagBatch.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TagBatch.java
new file mode 100644
index 0000000000..7f30e54dab
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TagBatch.java
@@ -0,0 +1,964 @@
+/*
+ * 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.tools.eventpump;
+
+import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.api.value.PlcValue;
+import org.apache.plc4x.java.tools.eventpump.transform.TransformException;
+import org.apache.plc4x.java.tools.eventpump.transform.ValueTransformer;
+import org.apache.plc4x.java.tools.eventpump.transform.ValueTransformerRegistry;
+import org.apache.plc4x.java.tools.eventpump.triggers.Trigger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Represents a batch of tags to be fetched from a PLC.
+ * <p>
+ * A batch consists of:
+ * - A unique batch ID
+ * - A PLC connection
+ * - A map of tag names to tag addresses
+ * - A trigger that determines when to fetch the tags
+ * - A listener that is notified when tags are fetched
+ * <p>
+ * When the trigger fires, the batch automatically reads all configured tags
+ * from the PLC and notifies the listener with the results.
+ * <p>
+ * Example using the Builder pattern:
+ * <pre>
+ * TagBatch batch = TagBatch.builder()
+ * .withBatchId("batch1")
+ * .withConnectionManager(connectionManager)
+ * .withConnectionString("ads://192.168.1.1:851")
+ * .addTag("temperature", "MAIN.temperature")
+ * .addTag("pressure", "MAIN.pressure")
+ * .withTrigger(new TimerTrigger(5, TimeUnit.SECONDS))
+ * .withListener((b, response) -> {
+ * System.out.println("Temperature: " + response.getInt("temperature"));
+ * System.out.println("Pressure: " + response.getInt("pressure"));
+ * })
+ * .build();
+ *
+ * batch.start();
+ * </pre>
+ */
+public class TagBatch implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TagBatch.class);
+
+ private final String batchId;
+ private final PlcConnectionManager connectionManager;
+ private final String connectionString;
+ private final Map<String, String> tags; // tagName -> tagAddress
+ private final Map<String, String> transforms; // tagName -> transformation expression
+ private final Trigger trigger;
+ private final ValueTransformer valueTransformer;
+
+ // Exponential backoff state for handling consecutive fetch failures.
+ // Prevents log flooding and resource waste during network outages.
+ private static final long DEFAULT_MAX_BACKOFF_MS = 60_000;
+ private static final long DEFAULT_INITIAL_BACKOFF_MS = 1_000;
+ private final long maxBackoffMs;
+ private final long initialBackoffMs;
+ private int consecutiveFailures;
+ private long nextAllowedFetchTimeMs;
+
+ // Guard against overlapping fetch cycles. If a previous fetch hasn't completed
+ // when the next trigger fires, the new fetch is skipped to prevent queue pile-up.
+ private final AtomicBoolean fetchInProgress = new AtomicBoolean(false);
+ private volatile long lastFetchDurationMs = -1;
+ private volatile long skippedFetchCount = 0;
+
+ private volatile TagBatchListener listener;
+ private volatile boolean started = false;
+ private volatile boolean closed = false;
+
+ // Cache for read request optimization — reused across poll cycles as
+ // long as the tag set is unchanged. Re-enabling this cache is the
+ // single biggest steady-state polling-CPU optimization for the S7+
+ // driver (and others that stamp resolution state on tag instances):
+ // a fresh PlcReadRequest every cycle would re-parse every address
+ // string and allocate fresh PlcTag instances, defeating any tag-side
+ // resolution caches. Invalidated whenever the underlying tag set is
+ // mutated via {@link #addTag(String, String)} / {@link #addTags(Map)} /
+ // {@link #removeTag(String)} / {@link #removeTags} / {@link #clearTags()}.
+ private volatile Map<String, String> cachedTagSnapshot = null;
+ private volatile PlcReadRequest cachedReadRequest = null;
+
+ /**
+ * Private constructor - use Builder to create instances.
+ *
+ * @param batchId A unique identifier for this batch
+ * @param connectionManager The connection manager to use for leasing connections
+ * @param connectionString The connection string to use for getting connections
+ * @param tags A map of tag names to tag addresses
+ * @param transforms A map of tag names to transformation expressions
+ * @param trigger The trigger that determines when to fetch tags
+ * @param listener The listener to be notified when tags are fetched
+ * @param valueTransformer The value transformer to use (or null for default from registry)
+ * @param transformerRegistry The transformer registry (or null to create default)
+ */
+ private TagBatch(String batchId, PlcConnectionManager connectionManager, String connectionString,
+ Map<String, String> tags, Map<String, String> transforms, Trigger trigger,
+ TagBatchListener listener, ValueTransformer valueTransformer,
+ ValueTransformerRegistry transformerRegistry,
+ long maxBackoffMs, long initialBackoffMs) {
+ if (batchId == null || batchId.trim().isEmpty()) {
+ throw new IllegalArgumentException("Batch ID cannot be null or empty");
+ }
+ if (connectionManager == null) {
+ throw new IllegalArgumentException("Connection manager cannot be null");
+ }
+ if (connectionString == null || connectionString.trim().isEmpty()) {
+ throw new IllegalArgumentException("Connection string cannot be null or empty");
+ }
+ if (tags == null || tags.isEmpty()) {
+ tags = new LinkedHashMap<>();
+ }
+ if (trigger == null) {
+ throw new IllegalArgumentException("Trigger cannot be null");
+ }
+
+ this.batchId = batchId;
+ this.connectionManager = connectionManager;
+ this.connectionString = connectionString;
+ this.tags = new LinkedHashMap<>(tags); // Preserve order
+ this.transforms = transforms != null ? new LinkedHashMap<>(transforms) : new LinkedHashMap<>();
+ this.trigger = trigger;
+ this.listener = listener;
+
+ // Determine the value transformer to use
+ if (valueTransformer != null) {
+ // Use explicitly provided transformer
+ this.valueTransformer = valueTransformer;
+ } else {
+ // Use default from registry (create default registry if not provided)
+ ValueTransformerRegistry registry = transformerRegistry != null ?
+ transformerRegistry : ValueTransformerRegistry.createDefault();
+ this.valueTransformer = registry.getDefault();
+ }
+
+ // Initialize backoff state
+ this.maxBackoffMs = maxBackoffMs;
+ this.initialBackoffMs = initialBackoffMs;
+ this.consecutiveFailures = 0;
+ this.nextAllowedFetchTimeMs = 0;
+
+ LOGGER.debug("Created TagBatch '{}' with {} tags ({} with transformations)",
+ batchId, tags.size(), this.transforms.size());
+ }
+
+ /**
+ * Set the listener to be notified when tags are fetched.
+ *
+ * @param listener The listener
+ */
+ public void setListener(TagBatchListener listener) {
+ this.listener = listener;
+ }
+
+ /**
+ * Create a new Builder for constructing TagBatch instances.
+ *
+ * @return A new Builder
+ */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /**
+ * Builder for creating TagBatch instances using a fluent API.
+ */
+ public static class Builder {
+ private String batchId;
+ private PlcConnectionManager connectionManager;
+ private String connectionString;
+ private final Map<String, String> tagAddresses = new LinkedHashMap<>();
+ private final Map<String, String> transforms = new LinkedHashMap<>();
+ private Trigger trigger;
+ private TagBatchListener listener;
+ private ValueTransformer valueTransformer;
+ private ValueTransformerRegistry transformerRegistry;
+ private long maxBackoffMs = DEFAULT_MAX_BACKOFF_MS;
+ private long initialBackoffMs = DEFAULT_INITIAL_BACKOFF_MS;
+
+ /**
+ * Set the batch ID.
+ *
+ * @param batchId The batch ID
+ * @return This builder
+ */
+ public Builder withBatchId(String batchId) {
+ this.batchId = batchId;
+ return this;
+ }
+
+ /**
+ * Set the connection manager.
+ *
+ * @param connectionManager The connection manager to use for leasing connections
+ * @return This builder
+ */
+ public Builder withConnectionManager(PlcConnectionManager connectionManager) {
+ this.connectionManager = connectionManager;
+ return this;
+ }
+
+ /**
+ * Set the connection string.
+ *
+ * @param connectionString The connection string to use when getting connections
+ * @return This builder
+ */
+ public Builder withConnectionString(String connectionString) {
+ this.connectionString = connectionString;
+ return this;
+ }
+
+ /**
+ * Add a single tag address to the batch.
+ *
+ * @param tagName The name of the tag
+ * @param tagAddress The address of the tag
+ * @return This builder
+ */
+ public Builder addTagAddress(String tagName, String tagAddress) {
+ this.tagAddresses.put(tagName, tagAddress);
+ return this;
+ }
+
+ /**
+ * Add multiple tag addresses to the batch.
+ *
+ * @param tags A map of tag names to tag addresses
+ * @return This builder
+ */
+ public Builder addTagAddresses(Map<String, String> tags) {
+ this.tagAddresses.putAll(tags);
+ return this;
+ }
+
+ /**
+ * Set the trigger for this batch.
+ *
+ * @param trigger The trigger
+ * @return This builder
+ */
+ public Builder withTrigger(Trigger trigger) {
+ this.trigger = trigger;
+ return this;
+ }
+
+ /**
+ * Set the listener for this batch.
+ *
+ * @param listener The listener
+ * @return This builder
+ */
+ public Builder withListener(TagBatchListener listener) {
+ this.listener = listener;
+ return this;
+ }
+
+ /**
+ * Add a transformation expression for a specific tag.
+ *
+ * @param tagName The tag name
+ * @param expression The transformation expression
+ * @return This builder
+ */
+ public Builder addTransform(String tagName, String expression) {
+ this.transforms.put(tagName, expression);
+ return this;
+ }
+
+ /**
+ * Add multiple transformation expressions.
+ *
+ * @param transforms A map of tag names to transformation expressions
+ * @return This builder
+ */
+ public Builder addTransforms(Map<String, String> transforms) {
+ this.transforms.putAll(transforms);
+ return this;
+ }
+
+ /**
+ * Set a custom value transformer.
+ * If not set, the default transformer from the registry will be used.
+ *
+ * @param valueTransformer The value transformer
+ * @return This builder
+ */
+ public Builder withValueTransformer(ValueTransformer valueTransformer) {
+ this.valueTransformer = valueTransformer;
+ return this;
+ }
+
+ /**
+ * Set the transformer registry to use.
+ * If not set, a default registry will be created with standard transformers.
+ *
+ * @param transformerRegistry The transformer registry
+ * @return This builder
+ */
+ public Builder withTransformerRegistry(ValueTransformerRegistry transformerRegistry) {
+ this.transformerRegistry = transformerRegistry;
+ return this;
+ }
+
+ /**
+ * Set the maximum backoff interval for exponential backoff during consecutive fetch failures.
+ * Default is 60 seconds.
+ *
+ * @param maxBackoffMs The maximum backoff interval in milliseconds
+ * @return This builder
+ */
+ public Builder withMaxBackoffMs(long maxBackoffMs) {
+ this.maxBackoffMs = maxBackoffMs;
+ return this;
+ }
+
+ /**
+ * Set the initial backoff interval for exponential backoff during consecutive fetch failures.
+ * Default is 1 second. The backoff doubles on each consecutive failure.
+ *
+ * @param initialBackoffMs The initial backoff interval in milliseconds
+ * @return This builder
+ */
+ public Builder withInitialBackoffMs(long initialBackoffMs) {
+ this.initialBackoffMs = initialBackoffMs;
+ return this;
+ }
+
+ /**
+ * Build the TagBatch instance.
+ * Validates that all required fields are set.
+ * Note: The listener is optional and can be set later using setListener().
+ *
+ * @return A new TagBatch instance
+ * @throws IllegalStateException if any required field is not set
+ */
+ public TagBatch build() {
+ if (batchId == null || batchId.trim().isEmpty()) {
+ throw new IllegalStateException("Batch ID must be set");
+ }
+ if (connectionManager == null) {
+ throw new IllegalStateException("Connection manager must be set");
+ }
+ if (connectionString == null || connectionString.trim().isEmpty()) {
+ throw new IllegalStateException("Connection string must be set");
+ }
+ if (trigger == null) {
+ throw new IllegalStateException("Trigger must be set");
+ }
+ // Note: listener is optional and can be null
+
+ return new TagBatch(batchId, connectionManager, connectionString, tagAddresses, transforms,
+ trigger, listener, valueTransformer, transformerRegistry, maxBackoffMs, initialBackoffMs);
+ }
+ }
+
+ /**
+ * Start the batch.
+ * This starts the trigger, which will cause tag fetching to begin.
+ *
+ * @throws IllegalStateException if the batch is already started or no listener is set
+ */
+ public void start() {
+ if (closed) {
+ throw new IllegalStateException("Batch is closed");
+ }
+ if (started) {
+ throw new IllegalStateException("Batch is already started");
+ }
+ if (listener == null) {
+ throw new IllegalStateException("Listener must be set before starting");
+ }
+
+ // Start the trigger with a callback to fetch tags
+ trigger.start(t -> fetchTags());
+ started = true;
+
+ LOGGER.info("Started TagBatch '{}'", batchId);
+ }
+
+ /**
+ * Stop the batch.
+ * This stops the trigger, preventing further tag fetching.
+ */
+ public void stop() {
+ if (!started) {
+ return;
+ }
+
+ trigger.stop();
+ started = false;
+
+ LOGGER.info("Stopped TagBatch '{}'", batchId);
+ }
+
+ /**
+ * Manually fetch tags (regardless of trigger state).
+ * This can be used to force an immediate read.
+ * <p>
+ * Each call to fetchTags() will lease a fresh connection from the connection manager,
+ * perform the read, and then return the connection (via try-with-resources).
+ * <p>
+ * To optimize performance, this method caches the PlcReadRequest and reuses it
+ * when the batch configuration (tags) hasn't changed between runs.
+ *
+ * @return A future that completes when the fetch is done
+ */
+ public CompletableFuture<Void> fetchTags() {
+ if (closed) {
+ LOGGER.warn("Batch '{}' is closed, cannot fetch tags", batchId);
+ return CompletableFuture.failedFuture(new IllegalStateException("Batch is closed"));
+ }
+ // If we don't have any tags, we can skip the read
+ if (tags.isEmpty()) {
+ LOGGER.debug("Batch '{}' has no tags, skipping fetch", batchId);
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Skip this trigger if we're within the backoff window due to consecutive failures
+ if (System.currentTimeMillis() < nextAllowedFetchTimeMs) {
+ LOGGER.trace("Batch '{}' skipping fetch due to backoff (next attempt at {})",
+ batchId, nextAllowedFetchTimeMs);
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Skip this trigger if the previous fetch hasn't completed yet.
+ // This prevents queue pile-up when the PLC response time exceeds the polling interval.
+ if (!fetchInProgress.compareAndSet(false, true)) {
+ skippedFetchCount++;
+ LOGGER.warn("Batch '{}' skipping fetch — previous fetch still in progress " +
+ "(last fetch took {}ms, skipped {} consecutive triggers)",
+ batchId, lastFetchDurationMs, skippedFetchCount);
+ if (listener != null) {
+ listener.onFetchSkipped(this, lastFetchDurationMs, skippedFetchCount);
+ }
+ return CompletableFuture.completedFuture(null);
+ }
+ skippedFetchCount = 0;
+ long fetchStartTimeMs = System.currentTimeMillis();
+
+ LOGGER.debug("Fetching tags for batch '{}' (started={})", batchId, started);
+
+ try {
+ // Create a snapshot of tags to avoid ConcurrentModificationException
+ Map<String, String> tagSnapshot;
+ synchronized (this) {
+ tagSnapshot = new LinkedHashMap<>(tags);
+ }
+
+ // Lease a connection from the manager for this read operation
+ LOGGER.debug("Batch '{}' - Leasing connection for fetch", batchId);
+ PlcConnection connection = connectionManager.getConnection(connectionString);
+ LOGGER.debug("Batch '{}' - Connection leased successfully", batchId);
+
+ try {
+ PlcReadRequest request;
+
+ // Check if we can reuse the cached request. Reusing the
+ // PlcReadRequest reuses the underlying PlcTag instances,
+ // so driver-side resolution caches (e.g. S7+'s
+ // S7CommPlusResolutionState stamped on S7Tag) survive
+ // across poll cycles. Without this cache, every cycle
+ // rebuilds tags from address strings, which defeats
+ // those resolution caches and re-parses every address.
+ Map<String, String> cachedSnap = cachedTagSnapshot;
+ PlcReadRequest cachedReq = cachedReadRequest;
+ if (cachedReq != null && tagSnapshot.equals(cachedSnap)) {
+ request = cachedReq;
+ LOGGER.trace("Reusing cached read request for batch '{}'", batchId);
+ } else {
+ PlcReadRequest.Builder builder = connection.readRequestBuilder();
+ for (Map.Entry<String, String> entry : tagSnapshot.entrySet()) {
+ builder.addTagAddress(entry.getKey(), entry.getValue());
+ }
+ request = builder.build();
+ cachedReadRequest = request;
+ cachedTagSnapshot = tagSnapshot;
+ LOGGER.trace("Built and cached new read request for batch '{}'", batchId);
+ }
+
+ // Execute read and ensure the connection is returned AFTER the future completes
+ LOGGER.debug("Batch '{}' - Executing read request", batchId);
+ CompletableFuture<? extends PlcReadResponse> readFuture = request.execute();
+
+ // Process the response and close the connection afterward
+ return readFuture
+ .whenComplete((response, throwable) -> {
+ try {
+ if (throwable != null) {
+ // Handle error case — enter or increase backoff
+ enterBackoff();
+ LOGGER.error("Error fetching tags for batch '{}': {}", batchId, throwable.getMessage());
+ if (listener != null) {
+ listener.onError(this, throwable);
+ }
+ } else {
+ // Handle success case
+ try {
+ LOGGER.debug("Batch '{}' - Read request completed successfully", batchId);
+ // Apply transformations if any
+ PlcReadResponse finalResponse = applyTransformations(response);
+
+ // Notify listener
+ if (listener != null) {
+ listener.onTagsFetched(this, finalResponse);
+ }
+ LOGGER.trace("Successfully fetched tags for batch '{}'", batchId);
+ // Reset backoff only after listener processed successfully
+ resetBackoff();
+ } catch (Exception e) {
+ // Listener failed to process the response — enter backoff
+ // to prevent rapid-fire retries
+ enterBackoff();
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error in tag batch listener for batch '{}'", batchId, e);
+ } else {
+ LOGGER.error("Error in tag batch listener for batch '{}': {}", batchId, e.getMessage());
+ }
+ if (listener != null) {
+ listener.onError(this, e);
+ }
+ }
+ }
+ } finally {
+ // Track fetch duration and release the in-progress guard
+ lastFetchDurationMs = System.currentTimeMillis() - fetchStartTimeMs;
+ fetchInProgress.set(false);
+ LOGGER.debug("Batch '{}' - Fetch cycle completed in {}ms", batchId, lastFetchDurationMs);
+
+ // ALWAYS close connection after processing (success or error)
+ LOGGER.debug("Batch '{}' - Returning connection to pool", batchId);
+ try {
+ connection.close();
+ LOGGER.debug("Batch '{}' - Connection returned successfully", batchId);
+ } catch (Exception e) {
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error returning connection to pool for batch '{}'", batchId, e);
+ } else {
+ LOGGER.error("Error returning connection to pool for batch '{}': {}", batchId, e.getMessage());
+ }
+ }
+ }
+ })
+ .thenApply(response -> null); // Convert to CompletableFuture<Void>
+ } catch (Exception e) {
+ // If we fail to build the request, close the connection immediately
+ lastFetchDurationMs = System.currentTimeMillis() - fetchStartTimeMs;
+ fetchInProgress.set(false);
+ try {
+ connection.close();
+ } catch (Exception closeEx) {
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error closing connection after request build failure for batch '{}'", batchId, closeEx);
+ } else {
+ LOGGER.error("Error closing connection after request build failure for batch '{}': {}", batchId, closeEx.getMessage());
+ }
+ }
+ throw e;
+ }
+
+ } catch (Exception e) {
+ // Connection or request build failure — enter or increase backoff
+ lastFetchDurationMs = System.currentTimeMillis() - fetchStartTimeMs;
+ fetchInProgress.set(false);
+ enterBackoff();
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error building read request for batch '{}'", batchId, e);
+ } else {
+ LOGGER.error("Error building read request for batch '{}': {}", batchId, e.getMessage());
+ }
+ if (listener != null) {
+ listener.onError(this, e);
+ }
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+
+ /**
+ * Check if the batch is started.
+ *
+ * @return true if started, false otherwise
+ */
+ public boolean isStarted() {
+ return started;
+ }
+
+ /**
+ * Get the batch ID.
+ *
+ * @return The batch ID
+ */
+ public String getBatchId() {
+ return batchId;
+ }
+
+ /**
+ * Get the connection manager being used.
+ *
+ * @return The connection manager
+ */
+ public PlcConnectionManager getConnectionManager() {
+ return connectionManager;
+ }
+
+ /**
+ * Get the connection string being used.
+ *
+ * @return The connection string
+ */
+ public String getConnectionString() {
+ return connectionString;
+ }
+
+ /**
+ * Get the tags being fetched.
+ *
+ * @return An unmodifiable map of tag names to addresses
+ */
+ public Map<String, String> getTags() {
+ return Collections.unmodifiableMap(tags);
+ }
+
+ /**
+ * Get the names of all tags in the batch.
+ *
+ * @return An unmodifiable set of tag names
+ */
+ public synchronized Set<String> getTagNames() {
+ return Collections.unmodifiableSet(tags.keySet());
+ }
+
+ /**
+ * Invalidate the cached read request — the next fetch cycle will
+ * rebuild from the current tag set. Called whenever the underlying
+ * tag map is mutated.
+ */
+ private void invalidateCache() {
+ cachedReadRequest = null;
+ cachedTagSnapshot = null;
+ }
+
+ /**
+ * Add a tag to the batch.
+ * Note: If the batch is currently started, the tag will be included in the next fetch cycle.
+ *
+ * @param tagName The name of the tag
+ * @param tagAddress The address of the tag
+ * @throws IllegalArgumentException if tagName or tagAddress is null or empty
+ */
+ public synchronized void addTag(String tagName, String tagAddress) {
+ if (tagName == null || tagName.trim().isEmpty()) {
+ throw new IllegalArgumentException("Tag name cannot be null or empty");
+ }
+ if (tagAddress == null || tagAddress.trim().isEmpty()) {
+ throw new IllegalArgumentException("Tag address cannot be null or empty");
+ }
+ if (closed) {
+ throw new IllegalStateException("Batch is closed");
+ }
+
+ tags.put(tagName, tagAddress);
+ invalidateCache();
+ LOGGER.debug("Added tag '{}' to batch '{}'", tagName, batchId);
+ }
+
+ /**
+ * Add multiple tags to the batch.
+ * Note: If the batch is currently started, the tags will be included in the next fetch cycle.
+ *
+ * @param tagsToAdd A map of tag names to tag addresses
+ * @throws IllegalArgumentException if tagsToAdd is null or empty
+ */
+ public synchronized void addTags(Map<String, String> tagsToAdd) {
+ if (tagsToAdd == null || tagsToAdd.isEmpty()) {
+ throw new IllegalArgumentException("Tags map cannot be null or empty");
+ }
+ if (closed) {
+ throw new IllegalStateException("Batch is closed");
+ }
+
+ tags.putAll(tagsToAdd);
+ invalidateCache();
+ LOGGER.debug("Added {} tags to batch '{}'", tagsToAdd.size(), batchId);
+ }
+
+ /**
+ * Remove a tag from the batch.
+ * Note: If the batch is currently started, the tag will be excluded from the next fetch cycle.
+ *
+ * @param tagName The name of the tag to remove
+ * @return true if the tag was removed, false if it didn't exist
+ */
+ public synchronized boolean removeTag(String tagName) {
+ if (closed) {
+ throw new IllegalStateException("Batch is closed");
+ }
+
+ boolean removed = tags.remove(tagName) != null;
+ if (removed) {
+ invalidateCache();
+ LOGGER.debug("Removed tag '{}' from batch '{}'", tagName, batchId);
+ }
+ return removed;
+ }
+
+ /**
+ * Remove multiple tags from the batch.
+ * Note: If the batch is currently started, the tags will be excluded from the next fetch cycle.
+ *
+ * @param tagNames Collection of tag names to remove
+ * @return The number of tags that were actually removed
+ */
+ public synchronized int removeTags(Collection<String> tagNames) {
+ if (closed) {
+ throw new IllegalStateException("Batch is closed");
+ }
+ if (tagNames == null) {
+ return 0;
+ }
+
+ int count = 0;
+ for (String tagName : tagNames) {
+ if (tags.remove(tagName) != null) {
+ count++;
+ }
+ }
+
+ if (count > 0) {
+ invalidateCache();
+ LOGGER.debug("Removed {} tags from batch '{}'", count, batchId);
+ }
+ return count;
+ }
+
+ /**
+ * Remove all tags from the batch.
+ * Warning: After calling this method, the batch will have no tags and fetchTags() will do nothing
+ * until tags are added again.
+ */
+ public synchronized void clearTags() {
+ if (closed) {
+ throw new IllegalStateException("Batch is closed");
+ }
+
+ int count = tags.size();
+ tags.clear();
+ invalidateCache();
+ LOGGER.debug("Cleared all {} tags from batch '{}'", count, batchId);
+ }
+
+ /**
+ * Check if the batch contains a specific tag.
+ *
+ * @param tagName The tag name to check
+ * @return true if the tag exists in the batch, false otherwise
+ */
+ public synchronized boolean hasTag(String tagName) {
+ return tags.containsKey(tagName);
+ }
+
+ /**
+ * Get the number of tags in the batch.
+ *
+ * @return The number of tags
+ */
+ public synchronized int getTagCount() {
+ return tags.size();
+ }
+
+ /**
+ * Get the trigger.
+ *
+ * @return The trigger
+ */
+ public Trigger getTrigger() {
+ return trigger;
+ }
+
+ /**
+ * Apply transformations to the read response.
+ *
+ * @param response The original response
+ * @return The transformed response (or original if no transformations)
+ */
+ private PlcReadResponse applyTransformations(PlcReadResponse response) {
+ if (transforms.isEmpty()) {
+ return response;
+ }
+
+ Map<String, PlcValue> transformedValues = new HashMap<>();
+ boolean hasTransformations = false;
+
+ for (String tagName : response.getTagNames()) {
+ String expression = transforms.get(tagName);
+ if (expression != null && !expression.trim().isEmpty()) {
+ try {
+ // Get the original value
+ PlcValue originalValue = response.getPlcValue(tagName);
+
+ // Build context with the original value available as "value"
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", originalValue);
+
+ // Also add all tag values by name for cross-tag expressions
+ for (String otherTagName : response.getTagNames()) {
+ context.put(otherTagName, response.getPlcValue(otherTagName));
+ }
+
+ // Apply transformation
+ PlcValue transformedValue = valueTransformer.transform(expression, context);
+ transformedValues.put(tagName, transformedValue);
+ hasTransformations = true;
+
+ LOGGER.trace("Transformed tag '{}' from {} to {} using expression: {}",
+ tagName, originalValue, transformedValue, expression);
+
+ } catch (TransformException e) {
+ LOGGER.error("Failed to transform tag '{}' with expression '{}': {}",
+ tagName, expression, e.getMessage());
+ // Keep original value on transformation error
+ }
+ }
+ }
+
+ if (hasTransformations) {
+ return new TransformedPlcReadResponse(response, transformedValues);
+ } else {
+ return response;
+ }
+ }
+
+ /**
+ * Enter or increase backoff after a failed fetch attempt.
+ * Uses exponential backoff: initialBackoffMs * 2^(failures-1), capped at maxBackoffMs.
+ */
+ private void enterBackoff() {
+ consecutiveFailures++;
+ long backoffMs = Math.min(initialBackoffMs * (1L << (consecutiveFailures - 1)), maxBackoffMs);
+ nextAllowedFetchTimeMs = System.currentTimeMillis() + backoffMs;
+ LOGGER.info("Batch '{}' entering backoff: next attempt in {}ms (failure {})",
+ batchId, backoffMs, consecutiveFailures);
+ }
+
+ /**
+ * Reset backoff after a successful fetch. Resumes normal polling immediately.
+ */
+ private void resetBackoff() {
+ if (consecutiveFailures > 0) {
+ LOGGER.info("Batch '{}' backoff reset after {} failures", batchId, consecutiveFailures);
+ consecutiveFailures = 0;
+ nextAllowedFetchTimeMs = 0;
+ }
+ }
+
+ /**
+ * Get the current number of consecutive failures (for testing).
+ *
+ * @return The number of consecutive failures
+ */
+ int getConsecutiveFailures() {
+ return consecutiveFailures;
+ }
+
+ /**
+ * Get the next allowed fetch time in millis (for testing).
+ *
+ * @return The next allowed fetch time
+ */
+ long getNextAllowedFetchTimeMs() {
+ return nextAllowedFetchTimeMs;
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+
+ stop();
+ trigger.close();
+ closed = true;
+
+ LOGGER.debug("Closed TagBatch '{}'", batchId);
+ }
+
+ /**
+ * Get the duration of the last fetch cycle in milliseconds.
+ *
+ * @return The duration in milliseconds, or -1 if no fetch has completed yet
+ */
+ public long getLastFetchDurationMs() {
+ return lastFetchDurationMs;
+ }
+
+ /**
+ * Listener interface for tag batch events.
+ */
+ public interface TagBatchListener {
+ /**
+ * Called when tags are successfully fetched.
+ *
+ * @param batch The batch that was fetched
+ * @param response The read response containing tag values
+ */
+ void onTagsFetched(TagBatch batch, PlcReadResponse response);
+
+ /**
+ * Called when an error occurs during tag fetching.
+ *
+ * @param batch The batch that encountered the error
+ * @param error The error that occurred
+ */
+ default void onError(TagBatch batch, Throwable error) {
+ // Default implementation: log the error (stacktrace only at trace level)
+ Logger logger = LoggerFactory.getLogger(TagBatchListener.class);
+ if (logger.isTraceEnabled()) {
+ logger.error("Error in tag batch '{}': {}", batch.getBatchId(), error.getMessage(), error);
+ } else {
+ logger.error("Error in tag batch '{}': {}", batch.getBatchId(), error.getMessage());
+ }
+ }
+
+ /**
+ * Called when a fetch trigger is skipped because the previous fetch is still in progress.
+ * This indicates that the polling interval is shorter than the PLC response time.
+ *
+ * @param batch The batch that skipped the fetch
+ * @param lastFetchDurationMs The duration of the last completed fetch in milliseconds
+ * @param consecutiveSkips The number of consecutive triggers that have been skipped
+ */
+ default void onFetchSkipped(TagBatch batch, long lastFetchDurationMs, long consecutiveSkips) {
+ // Default implementation: already logged at WARN level in TagBatch.fetchTags()
+ }
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TransformedPlcReadResponse.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TransformedPlcReadResponse.java
new file mode 100644
index 0000000000..4e47e3d95b
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/TransformedPlcReadResponse.java
@@ -0,0 +1,498 @@
+/*
+ * 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.tools.eventpump;
+
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.api.metadata.Metadata;
+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 java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Wrapper for PlcReadResponse that applies value transformations.
+ * <p>
+ * This class delegates most calls to the underlying response, but returns
+ * transformed values for tags that have transformation expressions configured.
+ */
+class TransformedPlcReadResponse implements PlcReadResponse {
+
+ private final PlcReadResponse delegate;
+ private final Map<String, PlcValue> transformedValues;
+
+ /**
+ * Create a transformed response.
+ *
+ * @param delegate The original response
+ * @param transformedValues Map of tag names to transformed values
+ */
+ public TransformedPlcReadResponse(PlcReadResponse delegate, Map<String, PlcValue> transformedValues) {
+ this.delegate = delegate;
+ this.transformedValues = new HashMap<>(transformedValues);
+ }
+
+ private PlcValue getValueOrDelegate(String tagName) {
+ return transformedValues.containsKey(tagName) ? transformedValues.get(tagName) : delegate.getPlcValue(tagName);
+ }
+
+ @Override
+ public PlcReadRequest getRequest() {
+ return delegate.getRequest();
+ }
+
+ @Override
+ public Collection<String> getTagNames() {
+ return delegate.getTagNames();
+ }
+
+ @Override
+ public PlcTag getTag(String name) {
+ return delegate.getTag(name);
+ }
+
+ @Override
+ public PlcResponseCode getResponseCode(String tagName) {
+ return delegate.getResponseCode(tagName);
+ }
+
+ @Override
+ public Metadata getTagMetadata(String name) {
+ return delegate.getTagMetadata(name);
+ }
+
+ @Override
+ public PlcValue getPlcValue(String tagName) {
+ return getValueOrDelegate(tagName);
+ }
+
+ @Override
+ public PlcValue getAsPlcValue() {
+ return delegate.getAsPlcValue();
+ }
+
+ @Override
+ public int getNumberOfValues(String name) {
+ return delegate.getNumberOfValues(name);
+ }
+
+ @Override
+ public Object getObject(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getObject();
+ }
+ return delegate.getObject(name);
+ }
+
+ @Override
+ public Object getObject(String name, int index) {
+ return delegate.getObject(name, index);
+ }
+
+ @Override
+ public Collection<Object> getAllObjects(String name) {
+ return delegate.getAllObjects(name);
+ }
+
+ // Boolean methods
+ @Override
+ public boolean isValidBoolean(String name) {
+ return delegate.isValidBoolean(name);
+ }
+
+ @Override
+ public boolean isValidBoolean(String name, int index) {
+ return delegate.isValidBoolean(name, index);
+ }
+
+ @Override
+ public Boolean getBoolean(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getBoolean();
+ }
+ return delegate.getBoolean(name);
+ }
+
+ @Override
+ public Boolean getBoolean(String name, int index) {
+ return delegate.getBoolean(name, index);
+ }
+
+ @Override
+ public Collection<Boolean> getAllBooleans(String name) {
+ return delegate.getAllBooleans(name);
+ }
+
+ // Byte methods
+ @Override
+ public boolean isValidByte(String name) {
+ return delegate.isValidByte(name);
+ }
+
+ @Override
+ public boolean isValidByte(String name, int index) {
+ return delegate.isValidByte(name, index);
+ }
+
+ @Override
+ public Byte getByte(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getByte();
+ }
+ return delegate.getByte(name);
+ }
+
+ @Override
+ public Byte getByte(String name, int index) {
+ return delegate.getByte(name, index);
+ }
+
+ @Override
+ public Collection<Byte> getAllBytes(String name) {
+ return delegate.getAllBytes(name);
+ }
+
+ // Short methods
+ @Override
+ public boolean isValidShort(String name) {
+ return delegate.isValidShort(name);
+ }
+
+ @Override
+ public boolean isValidShort(String name, int index) {
+ return delegate.isValidShort(name, index);
+ }
+
+ @Override
+ public Short getShort(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getShort();
+ }
+ return delegate.getShort(name);
+ }
+
+ @Override
+ public Short getShort(String name, int index) {
+ return delegate.getShort(name, index);
+ }
+
+ @Override
+ public Collection<Short> getAllShorts(String name) {
+ return delegate.getAllShorts(name);
+ }
+
+ // Integer methods
+ @Override
+ public boolean isValidInteger(String name) {
+ return delegate.isValidInteger(name);
+ }
+
+ @Override
+ public boolean isValidInteger(String name, int index) {
+ return delegate.isValidInteger(name, index);
+ }
+
+ @Override
+ public Integer getInteger(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getInteger();
+ }
+ return delegate.getInteger(name);
+ }
+
+ @Override
+ public Integer getInteger(String name, int index) {
+ return delegate.getInteger(name, index);
+ }
+
+ @Override
+ public Collection<Integer> getAllIntegers(String name) {
+ return delegate.getAllIntegers(name);
+ }
+
+ // BigInteger methods
+ @Override
+ public boolean isValidBigInteger(String name) {
+ return delegate.isValidBigInteger(name);
+ }
+
+ @Override
+ public boolean isValidBigInteger(String name, int index) {
+ return delegate.isValidBigInteger(name, index);
+ }
+
+ @Override
+ public BigInteger getBigInteger(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getBigInteger();
+ }
+ return delegate.getBigInteger(name);
+ }
+
+ @Override
+ public BigInteger getBigInteger(String name, int index) {
+ return delegate.getBigInteger(name, index);
+ }
+
+ @Override
+ public Collection<BigInteger> getAllBigIntegers(String name) {
+ return delegate.getAllBigIntegers(name);
+ }
+
+ // Long methods
+ @Override
+ public boolean isValidLong(String name) {
+ return delegate.isValidLong(name);
+ }
+
+ @Override
+ public boolean isValidLong(String name, int index) {
+ return delegate.isValidLong(name, index);
+ }
+
+ @Override
+ public Long getLong(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getLong();
+ }
+ return delegate.getLong(name);
+ }
+
+ @Override
+ public Long getLong(String name, int index) {
+ return delegate.getLong(name, index);
+ }
+
+ @Override
+ public Collection<Long> getAllLongs(String name) {
+ return delegate.getAllLongs(name);
+ }
+
+ // Float methods
+ @Override
+ public boolean isValidFloat(String name) {
+ return delegate.isValidFloat(name);
+ }
+
+ @Override
+ public boolean isValidFloat(String name, int index) {
+ return delegate.isValidFloat(name, index);
+ }
+
+ @Override
+ public Float getFloat(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getFloat();
+ }
+ return delegate.getFloat(name);
+ }
+
+ @Override
+ public Float getFloat(String name, int index) {
+ return delegate.getFloat(name, index);
+ }
+
+ @Override
+ public Collection<Float> getAllFloats(String name) {
+ return delegate.getAllFloats(name);
+ }
+
+ // Double methods
+ @Override
+ public boolean isValidDouble(String name) {
+ return delegate.isValidDouble(name);
+ }
+
+ @Override
+ public boolean isValidDouble(String name, int index) {
+ return delegate.isValidDouble(name, index);
+ }
+
+ @Override
+ public Double getDouble(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getDouble();
+ }
+ return delegate.getDouble(name);
+ }
+
+ @Override
+ public Double getDouble(String name, int index) {
+ return delegate.getDouble(name, index);
+ }
+
+ @Override
+ public Collection<Double> getAllDoubles(String name) {
+ return delegate.getAllDoubles(name);
+ }
+
+ // BigDecimal methods
+ @Override
+ public boolean isValidBigDecimal(String name) {
+ return delegate.isValidBigDecimal(name);
+ }
+
+ @Override
+ public boolean isValidBigDecimal(String name, int index) {
+ return delegate.isValidBigDecimal(name, index);
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getBigDecimal();
+ }
+ return delegate.getBigDecimal(name);
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(String name, int index) {
+ return delegate.getBigDecimal(name, index);
+ }
+
+ @Override
+ public Collection<BigDecimal> getAllBigDecimals(String name) {
+ return delegate.getAllBigDecimals(name);
+ }
+
+ // String methods
+ @Override
+ public boolean isValidString(String name) {
+ return delegate.isValidString(name);
+ }
+
+ @Override
+ public boolean isValidString(String name, int index) {
+ return delegate.isValidString(name, index);
+ }
+
+ @Override
+ public String getString(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getString();
+ }
+ return delegate.getString(name);
+ }
+
+ @Override
+ public String getString(String name, int index) {
+ return delegate.getString(name, index);
+ }
+
+ @Override
+ public Collection<String> getAllStrings(String name) {
+ return delegate.getAllStrings(name);
+ }
+
+ // Time methods
+ @Override
+ public boolean isValidTime(String name) {
+ return delegate.isValidTime(name);
+ }
+
+ @Override
+ public boolean isValidTime(String name, int index) {
+ return delegate.isValidTime(name, index);
+ }
+
+ @Override
+ public LocalTime getTime(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getTime();
+ }
+ return delegate.getTime(name);
+ }
+
+ @Override
+ public LocalTime getTime(String name, int index) {
+ return delegate.getTime(name, index);
+ }
+
+ @Override
+ public Collection<LocalTime> getAllTimes(String name) {
+ return delegate.getAllTimes(name);
+ }
+
+ // Date methods
+ @Override
+ public boolean isValidDate(String name) {
+ return delegate.isValidDate(name);
+ }
+
+ @Override
+ public boolean isValidDate(String name, int index) {
+ return delegate.isValidDate(name, index);
+ }
+
+ @Override
+ public LocalDate getDate(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getDate();
+ }
+ return delegate.getDate(name);
+ }
+
+ @Override
+ public LocalDate getDate(String name, int index) {
+ return delegate.getDate(name, index);
+ }
+
+ @Override
+ public Collection<LocalDate> getAllDates(String name) {
+ return delegate.getAllDates(name);
+ }
+
+ // DateTime methods
+ @Override
+ public boolean isValidDateTime(String name) {
+ return delegate.isValidDateTime(name);
+ }
+
+ @Override
+ public boolean isValidDateTime(String name, int index) {
+ return delegate.isValidDateTime(name, index);
+ }
+
+ @Override
+ public LocalDateTime getDateTime(String name) {
+ if (transformedValues.containsKey(name)) {
+ return transformedValues.get(name).getDateTime();
+ }
+ return delegate.getDateTime(name);
+ }
+
+ @Override
+ public LocalDateTime getDateTime(String name, int index) {
+ return delegate.getDateTime(name, index);
+ }
+
+ @Override
+ public Collection<LocalDateTime> getAllDateTimes(String name) {
+ return delegate.getAllDateTimes(name);
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/BatchConfiguration.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/BatchConfiguration.java
new file mode 100644
index 0000000000..fcb74e0b8d
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/BatchConfiguration.java
@@ -0,0 +1,167 @@
+/*
+ * 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.tools.eventpump.config;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Configuration for a tag batch.
+ * <p>
+ * Supports both simple and extended tag formats for backward compatibility:
+ * <p>
+ * Simple format (YAML):
+ * <pre>
+ * batches:
+ * - id: batch1
+ * connectionId: plc1
+ * tags:
+ * temperature: "MAIN.temperature"
+ * pressure: "MAIN.pressure"
+ * </pre>
+ * <p>
+ * Extended format with transformations (YAML):
+ * <pre>
+ * batches:
+ * - id: batch1
+ * connectionId: plc1
+ * tags:
+ * temperature:
+ * address: "MAIN.temperature"
+ * transform: "value * 1.8 + 32"
+ * pressure:
+ * address: "MAIN.pressure"
+ * </pre>
+ */
+public class BatchConfiguration {
+
+ @JsonProperty("id")
+ private String id;
+
+ @JsonProperty("connectionId")
+ private String connectionId;
+
+ @JsonProperty("tags")
+ @JsonDeserialize(using = TagMapDeserializer.class)
+ private Map<String, TagConfiguration> tags = new LinkedHashMap<>();
+
+ @JsonProperty("trigger")
+ private TriggerConfiguration trigger;
+
+ /**
+ * Get the batch ID.
+ *
+ * @return The batch ID
+ */
+ public String getId() {
+ return id;
+ }
+
+ /**
+ * Set the batch ID.
+ *
+ * @param id The batch ID
+ */
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ /**
+ * Get the connection ID.
+ *
+ * @return The connection ID
+ */
+ public String getConnectionId() {
+ return connectionId;
+ }
+
+ /**
+ * Set the connection ID.
+ *
+ * @param connectionId The connection ID
+ */
+ public void setConnectionId(String connectionId) {
+ this.connectionId = connectionId;
+ }
+
+ /**
+ * Get the tag configurations (name -> TagConfiguration).
+ *
+ * @return The tags map
+ */
+ public Map<String, TagConfiguration> getTags() {
+ return tags;
+ }
+
+ /**
+ * Set the tag configurations.
+ *
+ * @param tags The tags map
+ */
+ public void setTags(Map<String, TagConfiguration> tags) {
+ this.tags = tags;
+ }
+
+ /**
+ * Get a simple map of tag names to addresses (for backward compatibility).
+ *
+ * @return A map of tag name -> address
+ */
+ public Map<String, String> getSimpleTagMap() {
+ Map<String, String> simpleMap = new LinkedHashMap<>();
+ for (Map.Entry<String, TagConfiguration> entry : tags.entrySet()) {
+ simpleMap.put(entry.getKey(), entry.getValue().getAddress());
+ }
+ return simpleMap;
+ }
+
+ /**
+ * Set tags from a simple map (for backward compatibility and testing).
+ *
+ * @param simpleTags A map of tag name -> address
+ */
+ public void setSimpleTags(Map<String, String> simpleTags) {
+ this.tags.clear();
+ for (Map.Entry<String, String> entry : simpleTags.entrySet()) {
+ this.tags.put(entry.getKey(), new TagConfiguration(entry.getValue()));
+ }
+ }
+
+ /**
+ * Get the trigger configuration.
+ *
+ * @return The trigger configuration
+ */
+ public TriggerConfiguration getTrigger() {
+ return trigger;
+ }
+
+ /**
+ * Set the trigger configuration.
+ *
+ * @param trigger The trigger configuration
+ */
+ public void setTrigger(TriggerConfiguration trigger) {
+ this.trigger = trigger;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/ConnectionConfiguration.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/ConnectionConfiguration.java
new file mode 100644
index 0000000000..c399f78f04
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/ConnectionConfiguration.java
@@ -0,0 +1,112 @@
+/*
+ * 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.tools.eventpump.config;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Configuration for a PLC connection.
+ */
+public class ConnectionConfiguration {
+
+ @JsonProperty("id")
+ private String id;
+
+ @JsonProperty("url")
+ private String url;
+
+ @JsonProperty("username")
+ private String username;
+
+ @JsonProperty("password")
+ private String password;
+
+ /**
+ * Get the connection ID.
+ *
+ * @return The connection ID
+ */
+ public String getId() {
+ return id;
+ }
+
+ /**
+ * Set the connection ID.
+ *
+ * @param id The connection ID
+ */
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ /**
+ * Get the connection URL.
+ *
+ * @return The connection URL
+ */
+ public String getUrl() {
+ return url;
+ }
+
+ /**
+ * Set the connection URL.
+ *
+ * @param url The connection URL
+ */
+ public void setUrl(String url) {
+ this.url = url;
+ }
+
+ /**
+ * Get the username for authentication.
+ *
+ * @return The username, or null if not using authentication
+ */
+ public String getUsername() {
+ return username;
+ }
+
+ /**
+ * Set the username for authentication.
+ *
+ * @param username The username
+ */
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ /**
+ * Get the password for authentication.
+ *
+ * @return The password, or null if not using authentication
+ */
+ public String getPassword() {
+ return password;
+ }
+
+ /**
+ * Set the password for authentication.
+ *
+ * @param password The password
+ */
+ public void setPassword(String password) {
+ this.password = password;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpConfiguration.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpConfiguration.java
new file mode 100644
index 0000000000..ac69aae462
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpConfiguration.java
@@ -0,0 +1,212 @@
+/*
+ * 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.tools.eventpump.config;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.xml.XmlMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Configuration model for EventPump.
+ * <p>
+ * This class can be serialized/deserialized from XML, JSON, or YAML files.
+ * <p>
+ * Example YAML configuration:
+ * <pre>
+ * connections:
+ * - id: plc1
+ * url: "ads:tcp://192.168.1.1"
+ *
+ * batches:
+ * - id: batch1
+ * connectionId: plc1
+ * tags:
+ * temperature: "MAIN.temperature"
+ * pressure: "MAIN.pressure"
+ * trigger:
+ * type: timer
+ * intervalSeconds: 5
+ *
+ * - id: batch2
+ * connectionId: plc1
+ * tags:
+ * status: "MAIN.status"
+ * trigger:
+ * type: subscription
+ * tagName: triggerTag
+ * tagAddress: "MAIN.trigger"
+ * </pre>
+ */
+public class EventPumpConfiguration {
+
+ @JsonProperty("connections")
+ private List<ConnectionConfiguration> connections = new ArrayList<>();
+
+ @JsonProperty("batches")
+ private List<BatchConfiguration> batches = new ArrayList<>();
+
+ /**
+ * Get the connection configurations.
+ *
+ * @return The list of connection configurations
+ */
+ public List<ConnectionConfiguration> getConnections() {
+ return connections;
+ }
+
+ /**
+ * Set the connection configurations.
+ *
+ * @param connections The connection configurations
+ */
+ public void setConnections(List<ConnectionConfiguration> connections) {
+ this.connections = connections;
+ }
+
+ /**
+ * Get the batch configurations.
+ *
+ * @return The list of batch configurations
+ */
+ public List<BatchConfiguration> getBatches() {
+ return batches;
+ }
+
+ /**
+ * Set the batch configurations.
+ *
+ * @param batches The batch configurations
+ */
+ public void setBatches(List<BatchConfiguration> batches) {
+ this.batches = batches;
+ }
+
+ /**
+ * Load configuration from a JSON file.
+ *
+ * @param file The file to load from
+ * @return The configuration
+ * @throws IOException if loading fails
+ */
+ public static EventPumpConfiguration fromJson(File file) throws IOException {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.readValue(file, EventPumpConfiguration.class);
+ }
+
+ /**
+ * Load configuration from a JSON input stream.
+ *
+ * @param inputStream The input stream to load from
+ * @return The configuration
+ * @throws IOException if loading fails
+ */
+ public static EventPumpConfiguration fromJson(InputStream inputStream) throws IOException {
+ ObjectMapper mapper = new ObjectMapper();
+ return mapper.readValue(inputStream, EventPumpConfiguration.class);
+ }
+
+ /**
+ * Load configuration from a YAML file.
+ *
+ * @param file The file to load from
+ * @return The configuration
+ * @throws IOException if loading fails
+ */
+ public static EventPumpConfiguration fromYaml(File file) throws IOException {
+ ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ return mapper.readValue(file, EventPumpConfiguration.class);
+ }
+
+ /**
+ * Load configuration from a YAML input stream.
+ *
+ * @param inputStream The input stream to load from
+ * @return The configuration
+ * @throws IOException if loading fails
+ */
+ public static EventPumpConfiguration fromYaml(InputStream inputStream) throws IOException {
+ ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ return mapper.readValue(inputStream, EventPumpConfiguration.class);
+ }
+
+ /**
+ * Load configuration from an XML file.
+ *
+ * @param file The file to load from
+ * @return The configuration
+ * @throws IOException if loading fails
+ */
+ public static EventPumpConfiguration fromXml(File file) throws IOException {
+ XmlMapper mapper = new XmlMapper();
+ return mapper.readValue(file, EventPumpConfiguration.class);
+ }
+
+ /**
+ * Load configuration from an XML input stream.
+ *
+ * @param inputStream The input stream to load from
+ * @return The configuration
+ * @throws IOException if loading fails
+ */
+ public static EventPumpConfiguration fromXml(InputStream inputStream) throws IOException {
+ XmlMapper mapper = new XmlMapper();
+ return mapper.readValue(inputStream, EventPumpConfiguration.class);
+ }
+
+ /**
+ * Save configuration to a JSON file.
+ *
+ * @param file The file to save to
+ * @throws IOException if saving fails
+ */
+ public void toJson(File file) throws IOException {
+ ObjectMapper mapper = new ObjectMapper();
+ mapper.writerWithDefaultPrettyPrinter().writeValue(file, this);
+ }
+
+ /**
+ * Save configuration to a YAML file.
+ *
+ * @param file The file to save to
+ * @throws IOException if saving fails
+ */
+ public void toYaml(File file) throws IOException {
+ ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ mapper.writeValue(file, this);
+ }
+
+ /**
+ * Save configuration to an XML file.
+ *
+ * @param file The file to save to
+ * @throws IOException if saving fails
+ */
+ public void toXml(File file) throws IOException {
+ XmlMapper mapper = new XmlMapper();
+ mapper.writerWithDefaultPrettyPrinter().writeValue(file, this);
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactory.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactory.java
new file mode 100644
index 0000000000..f05fa6656c
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactory.java
@@ -0,0 +1,279 @@
+/*
+ * 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.tools.eventpump.config;
+
+import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.tools.eventpump.EventPump;
+import org.apache.plc4x.java.tools.eventpump.TagBatch;
+import org.apache.plc4x.java.tools.eventpump.transform.ValueTransformerRegistry;
+import org.apache.plc4x.java.tools.eventpump.triggers.SubscriptionTrigger;
+import org.apache.plc4x.java.tools.eventpump.triggers.TimerTrigger;
+import org.apache.plc4x.java.tools.eventpump.triggers.Trigger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Factory for creating EventPump instances from configuration.
+ * <p>
+ * This class provides methods to build an EventPump from:
+ * - Configuration objects
+ * - Configuration files (JSON, YAML, XML)
+ * <p>
+ * Example usage:
+ * <pre>
+ * // From a configuration file
+ * PlcConnectionManager connectionManager = PlcDriverManager.getDefault();
+ * EventPump pump = EventPumpFactory.fromYaml(new File("config.yaml"), connectionManager,
+ * (batch, response) -> {
+ * // Handle all batches
+ * });
+ * pump.startAll();
+ * </pre>
+ */
+public class EventPumpFactory {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(EventPumpFactory.class);
+
+ /**
+ * Create an EventPump from configuration.
+ *
+ * @param config The configuration
+ * @param connectionManager The connection manager to use for creating connections
+ * @param defaultListener The default listener for all batches (can be null)
+ * @return The configured EventPump
+ * @throws Exception if creation fails
+ */
+ public static EventPump create(EventPumpConfiguration config,
+ PlcConnectionManager connectionManager,
+ TagBatch.TagBatchListener defaultListener) throws Exception {
+ return create(config, connectionManager, defaultListener, null);
+ }
+
+ /**
+ * Create an EventPump from configuration with a custom transformer registry.
+ *
+ * @param config The configuration
+ * @param connectionManager The connection manager to use for creating connections
+ * @param defaultListener The default listener for all batches (can be null)
+ * @param transformerRegistry The transformer registry to use (or null to create default)
+ * @return The configured EventPump
+ * @throws Exception if creation fails
+ */
+ public static EventPump create(EventPumpConfiguration config,
+ PlcConnectionManager connectionManager,
+ TagBatch.TagBatchListener defaultListener,
+ ValueTransformerRegistry transformerRegistry) throws Exception {
+
+ EventPump pump = new EventPump();
+
+ // Create or use a provided transformer registry (shared across all batches)
+ ValueTransformerRegistry registry = transformerRegistry != null ?
+ transformerRegistry : ValueTransformerRegistry.createDefault();
+
+ LOGGER.debug("Using transformer registry with transformers: {}", registry.getRegisteredNames());
+
+ // Store connection URLs (authentication info included in URL or stored separately if needed)
+ Map<String, String> connectionUrls = new HashMap<>();
+ for (ConnectionConfiguration connConfig : config.getConnections()) {
+ // Store the connection URL for later use
+ // Note: For authenticated connections, credentials would need to be handled separately
+ // or embedded in the URL format if the protocol supports it
+ connectionUrls.put(connConfig.getId(), connConfig.getUrl());
+ LOGGER.info("Registered connection '{}' to '{}'", connConfig.getId(), connConfig.getUrl());
+ }
+
+ // Create batches
+ for (BatchConfiguration batchConfig : config.getBatches()) {
+ String connectionUrl = connectionUrls.get(batchConfig.getConnectionId());
+ if (connectionUrl == null) {
+ throw new IllegalArgumentException("Connection '" + batchConfig.getConnectionId() +
+ "' not found for batch '" + batchConfig.getId() + "'");
+ }
+
+ // Create trigger (no longer needs connection parameter for timer triggers)
+ Trigger trigger = createTrigger(batchConfig.getTrigger());
+
+ // Create batch using builder with ConnectionManager
+ TagBatch.Builder batchBuilder = TagBatch.builder()
+ .withBatchId(batchConfig.getId())
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionUrl)
+ .addTagAddresses(batchConfig.getSimpleTagMap())
+ .withTrigger(trigger)
+ .withTransformerRegistry(registry); // Share registry across all batches
+
+ // Add transformations
+ for (Map.Entry<String, TagConfiguration> entry : batchConfig.getTags().entrySet()) {
+ if (entry.getValue().hasTransform()) {
+ batchBuilder.addTransform(entry.getKey(), entry.getValue().getTransform());
+ }
+ }
+
+ // Set listener if provided
+ if (defaultListener != null) {
+ batchBuilder.withListener(defaultListener);
+ }
+
+ TagBatch batch = batchBuilder.build();
+
+ pump.addBatch(batch);
+ LOGGER.info("Created batch '{}' with {} tags", batchConfig.getId(), batchConfig.getTags().size());
+ }
+
+ return pump;
+ }
+
+ /**
+ * Create a trigger from configuration.
+ *
+ * @param config The trigger configuration
+ * @return The trigger
+ */
+ private static Trigger createTrigger(TriggerConfiguration config) {
+ String type = config.getType();
+
+ if ("timer".equalsIgnoreCase(type)) {
+ // Timer trigger
+ long interval = config.getIntervalSeconds();
+ long initialDelay = config.getInitialDelaySeconds() != null ? config.getInitialDelaySeconds() : 0;
+
+ return new TimerTrigger(interval, initialDelay, TimeUnit.SECONDS);
+
+ } else if ("subscription".equalsIgnoreCase(type)) {
+ // Subscription trigger (placeholder)
+ String tagName = config.getTagName();
+ String tagAddress = config.getTagAddress();
+
+ if (tagName == null || tagAddress == null) {
+ throw new IllegalArgumentException("Subscription trigger requires tagName and tagAddress");
+ }
+
+ // Note: SubscriptionTrigger is currently a placeholder
+ // When implemented, it will need to manage its own connection or use a different approach
+ return new SubscriptionTrigger(tagName, tagAddress);
+
+ } else {
+ throw new IllegalArgumentException("Unknown trigger type: " + type);
+ }
+ }
+
+ /**
+ * Create an EventPump from a YAML file.
+ *
+ * @param file The YAML file
+ * @param connectionManager The connection manager
+ * @param defaultListener The default listener
+ * @return The configured EventPump
+ * @throws Exception if creation fails
+ */
+ public static EventPump fromYaml(java.io.File file,
+ PlcConnectionManager connectionManager,
+ TagBatch.TagBatchListener defaultListener) throws Exception {
+ return fromYaml(file, connectionManager, defaultListener, null);
+ }
+
+ /**
+ * Create an EventPump from a YAML file with a custom transformer registry.
+ *
+ * @param file The YAML file
+ * @param connectionManager The connection manager
+ * @param defaultListener The default listener
+ * @param transformerRegistry The transformer registry (or null to create default)
+ * @return The configured EventPump
+ * @throws Exception if creation fails
+ */
+ public static EventPump fromYaml(java.io.File file,
+ PlcConnectionManager connectionManager,
+ TagBatch.TagBatchListener defaultListener,
+ ValueTransformerRegistry transformerRegistry) throws Exception {
+ EventPumpConfiguration config = EventPumpConfiguration.fromYaml(file);
+ return create(config, connectionManager, defaultListener, transformerRegistry);
+ }
+
+ /**
+ * Create an EventPump from a JSON file.
+ *
+ * @param file The JSON file
+ * @param connectionManager The connection manager
+ * @param defaultListener The default listener
+ * @return The configured EventPump
+ * @throws Exception if creation fails
+ */
+ public static EventPump fromJson(java.io.File file,
+ PlcConnectionManager connectionManager,
+ TagBatch.TagBatchListener defaultListener) throws Exception {
+ return fromJson(file, connectionManager, defaultListener, null);
+ }
+
+ /**
+ * Create an EventPump from a JSON file with a custom transformer registry.
+ *
+ * @param file The JSON file
+ * @param connectionManager The connection manager
+ * @param defaultListener The default listener
+ * @param transformerRegistry The transformer registry (or null to create default)
+ * @return The configured EventPump
+ * @throws Exception if creation fails
+ */
+ public static EventPump fromJson(java.io.File file,
+ PlcConnectionManager connectionManager,
+ TagBatch.TagBatchListener defaultListener,
+ ValueTransformerRegistry transformerRegistry) throws Exception {
+ EventPumpConfiguration config = EventPumpConfiguration.fromJson(file);
+ return create(config, connectionManager, defaultListener, transformerRegistry);
+ }
+
+ /**
+ * Create an EventPump from an XML file.
+ *
+ * @param file The XML file
+ * @param connectionManager The connection manager
+ * @param defaultListener The default listener
+ * @return The configured EventPump
+ * @throws Exception if creation fails
+ */
+ public static EventPump fromXml(java.io.File file,
+ PlcConnectionManager connectionManager,
+ TagBatch.TagBatchListener defaultListener) throws Exception {
+ return fromXml(file, connectionManager, defaultListener, null);
+ }
+
+ /**
+ * Create an EventPump from an XML file with a custom transformer registry.
+ *
+ * @param file The XML file
+ * @param connectionManager The connection manager
+ * @param defaultListener The default listener
+ * @param transformerRegistry The transformer registry (or null to create default)
+ * @return The configured EventPump
+ * @throws Exception if creation fails
+ */
+ public static EventPump fromXml(java.io.File file,
+ PlcConnectionManager connectionManager,
+ TagBatch.TagBatchListener defaultListener,
+ ValueTransformerRegistry transformerRegistry) throws Exception {
+ EventPumpConfiguration config = EventPumpConfiguration.fromXml(file);
+ return create(config, connectionManager, defaultListener, transformerRegistry);
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TagConfiguration.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TagConfiguration.java
new file mode 100644
index 0000000000..3f76c9a9ef
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TagConfiguration.java
@@ -0,0 +1,149 @@
+/*
+ * 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.tools.eventpump.config;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Configuration for a single tag.
+ * <p>
+ * This class supports both simple string format (just the address) and
+ * extended format (address + optional transformation expression).
+ * <p>
+ * Simple format (YAML):
+ * <pre>
+ * tags:
+ * temperature: "MAIN.temperature"
+ * </pre>
+ * <p>
+ * Extended format (YAML):
+ * <pre>
+ * tags:
+ * temperature:
+ * address: "MAIN.temperature"
+ * transform: "value * 1.8 + 32"
+ * </pre>
+ */
+public class TagConfiguration {
+
+ @JsonProperty("address")
+ private String address;
+
+ @JsonProperty("transform")
+ private String transform;
+
+ /**
+ * Default constructor for Jackson.
+ */
+ public TagConfiguration() {
+ }
+
+ /**
+ * Create a tag configuration with just an address.
+ *
+ * @param address The tag address
+ */
+ public TagConfiguration(String address) {
+ this.address = address;
+ }
+
+ /**
+ * Create a tag configuration with address and transform.
+ *
+ * @param address The tag address
+ * @param transform The transformation expression (can be null)
+ */
+ public TagConfiguration(String address, String transform) {
+ this.address = address;
+ this.transform = transform;
+ }
+
+ /**
+ * Get the tag address.
+ *
+ * @return The tag address
+ */
+ public String getAddress() {
+ return address;
+ }
+
+ /**
+ * Set the tag address.
+ *
+ * @param address The tag address
+ */
+ public void setAddress(String address) {
+ this.address = address;
+ }
+
+ /**
+ * Get the transformation expression.
+ *
+ * @return The transformation expression, or null if no transformation
+ */
+ public String getTransform() {
+ return transform;
+ }
+
+ /**
+ * Set the transformation expression.
+ *
+ * @param transform The transformation expression
+ */
+ public void setTransform(String transform) {
+ this.transform = transform;
+ }
+
+ /**
+ * Check if this tag has a transformation.
+ *
+ * @return true if a transformation is configured
+ */
+ public boolean hasTransform() {
+ return transform != null && !transform.trim().isEmpty();
+ }
+
+ @Override
+ public String toString() {
+ if (hasTransform()) {
+ return "TagConfiguration{address='" + address + "', transform='" + transform + "'}";
+ } else {
+ return "TagConfiguration{address='" + address + "'}";
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ TagConfiguration that = (TagConfiguration) o;
+
+ if (address != null ? !address.equals(that.address) : that.address != null) return false;
+ return transform != null ? transform.equals(that.transform) : that.transform == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = address != null ? address.hashCode() : 0;
+ result = 31 * result + (transform != null ? transform.hashCode() : 0);
+ return result;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TagMapDeserializer.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TagMapDeserializer.java
new file mode 100644
index 0000000000..40674625d5
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TagMapDeserializer.java
@@ -0,0 +1,70 @@
+/*
+ * 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.tools.eventpump.config;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Custom deserializer for tag maps that supports both simple and extended formats.
+ * <p>
+ * Handles:
+ * - Simple format: "tagName": "address"
+ * - Extended format: "tagName": { "address": "...", "transform": "..." }
+ */
+public class TagMapDeserializer extends JsonDeserializer<Map<String, TagConfiguration>> {
+
+ @Override
+ public Map<String, TagConfiguration> deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+ Map<String, TagConfiguration> tags = new LinkedHashMap<>();
+ ObjectMapper mapper = (ObjectMapper) p.getCodec();
+ JsonNode node = mapper.readTree(p);
+
+ Iterator<Map.Entry<String, JsonNode>> fields = node.fields();
+ while (fields.hasNext()) {
+ Map.Entry<String, JsonNode> entry = fields.next();
+ String tagName = entry.getKey();
+ JsonNode tagNode = entry.getValue();
+
+ TagConfiguration tagConfig;
+ if (tagNode.isTextual()) {
+ // Simple format: just a string address
+ tagConfig = new TagConfiguration(tagNode.asText());
+ } else if (tagNode.isObject()) {
+ // Extended format: object with address and optional transform
+ tagConfig = mapper.treeToValue(tagNode, TagConfiguration.class);
+ } else {
+ throw new IOException("Invalid tag configuration for '" + tagName + "': expected string or object");
+ }
+
+ tags.put(tagName, tagConfig);
+ }
+
+ return tags;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TriggerConfiguration.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TriggerConfiguration.java
new file mode 100644
index 0000000000..096a125281
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/config/TriggerConfiguration.java
@@ -0,0 +1,163 @@
+/*
+ * 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.tools.eventpump.config;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Configuration for a trigger.
+ * <p>
+ * The type field determines the trigger type:
+ * - "timer": Time-based trigger (requires intervalSeconds)
+ * - "subscription": Subscription-based trigger (requires tagName and tagAddress)
+ */
+public class TriggerConfiguration {
+
+ @JsonProperty("type")
+ private String type;
+
+ // Timer trigger properties
+ @JsonProperty("intervalSeconds")
+ private Long intervalSeconds;
+
+ @JsonProperty("initialDelaySeconds")
+ private Long initialDelaySeconds;
+
+ // Subscription trigger properties
+ @JsonProperty("tagName")
+ private String tagName;
+
+ @JsonProperty("tagAddress")
+ private String tagAddress;
+
+ @JsonProperty("condition")
+ private String condition; // Optional: Expression for filtering events
+
+ /**
+ * Get the trigger type.
+ *
+ * @return The trigger type ("timer" or "subscription")
+ */
+ public String getType() {
+ return type;
+ }
+
+ /**
+ * Set the trigger type.
+ *
+ * @param type The trigger type
+ */
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ /**
+ * Get the interval in seconds (for timer triggers).
+ *
+ * @return The interval in seconds
+ */
+ public Long getIntervalSeconds() {
+ return intervalSeconds;
+ }
+
+ /**
+ * Set the interval in seconds (for timer triggers).
+ *
+ * @param intervalSeconds The interval in seconds
+ */
+ public void setIntervalSeconds(Long intervalSeconds) {
+ this.intervalSeconds = intervalSeconds;
+ }
+
+ /**
+ * Get the initial delay in seconds (for timer triggers).
+ *
+ * @return The initial delay in seconds, or null if not set
+ */
+ public Long getInitialDelaySeconds() {
+ return initialDelaySeconds;
+ }
+
+ /**
+ * Set the initial delay in seconds (for timer triggers).
+ *
+ * @param initialDelaySeconds The initial delay in seconds
+ */
+ public void setInitialDelaySeconds(Long initialDelaySeconds) {
+ this.initialDelaySeconds = initialDelaySeconds;
+ }
+
+ /**
+ * Get the tag name (for subscription triggers).
+ *
+ * @return The tag name
+ */
+ public String getTagName() {
+ return tagName;
+ }
+
+ /**
+ * Set the tag name (for subscription triggers).
+ *
+ * @param tagName The tag name
+ */
+ public void setTagName(String tagName) {
+ this.tagName = tagName;
+ }
+
+ /**
+ * Get the tag address (for subscription triggers).
+ *
+ * @return The tag address
+ */
+ public String getTagAddress() {
+ return tagAddress;
+ }
+
+ /**
+ * Set the tag address (for subscription triggers).
+ *
+ * @param tagAddress The tag address
+ */
+ public void setTagAddress(String tagAddress) {
+ this.tagAddress = tagAddress;
+ }
+
+ /**
+ * Get the condition expression (for subscription triggers).
+ * <p>
+ * This is an optional expression that can be used to filter subscription events.
+ * The syntax depends on the expression evaluator used.
+ *
+ * @return The condition expression, or null if not set
+ */
+ public String getCondition() {
+ return condition;
+ }
+
+ /**
+ * Set the condition expression (for subscription triggers).
+ *
+ * @param condition The condition expression
+ */
+ public void setCondition(String condition) {
+ this.condition = condition;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/SimpleExpressionEvaluator.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/SimpleExpressionEvaluator.java
new file mode 100644
index 0000000000..2fa8c8cfbf
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/SimpleExpressionEvaluator.java
@@ -0,0 +1,467 @@
+/*
+ * 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.tools.eventpump.transform;
+
+import org.apache.plc4x.java.api.value.PlcValue;
+import org.apache.plc4x.java.spi.values.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+/**
+ * Simple expression evaluator supporting basic arithmetic and comparisons.
+ * <p>
+ * Supported operations:
+ * - Arithmetic: +, -, *, /, % (modulo)
+ * - Comparisons: ==, !=, <, >, <=, >=
+ * - Logical: &&, ||, ! (not)
+ * - Parentheses: ( )
+ * - Variables: Any identifier (resolved from context)
+ * - Literals: Numeric (integer and floating point)
+ * <p>
+ * Examples:
+ * - "value * 1.8 + 32" - Convert Celsius to Fahrenheit
+ * - "temperature > 100" - Temperature check
+ * - "(pressure - 1013.25) / 1013.25 * 100" - Pressure deviation percentage
+ * - "active && temperature > 50" - Combined condition
+ */
+public class SimpleExpressionEvaluator implements ValueTransformer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SimpleExpressionEvaluator.class);
+
+ @Override
+ public String getName() {
+ return "simple";
+ }
+
+ @Override
+ public PlcValue transform(String expression, Map<String, PlcValue> context) throws TransformException {
+ if (expression == null || expression.trim().isEmpty()) {
+ throw new TransformException("Expression cannot be null or empty");
+ }
+
+ try {
+ Parser parser = new Parser(expression, context);
+ return parser.parse();
+ } catch (Exception e) {
+ throw new TransformException("Failed to evaluate expression: " + expression, e);
+ }
+ }
+
+ /**
+ * Simple recursive descent parser for expressions.
+ */
+ private static class Parser {
+ private final String expression;
+ private final Map<String, PlcValue> context;
+ private int pos = 0;
+
+ Parser(String expression, Map<String, PlcValue> context) {
+ this.expression = expression.trim();
+ this.context = context;
+ }
+
+ PlcValue parse() throws TransformException {
+ PlcValue result = parseLogicalOr();
+ if (pos < expression.length()) {
+ throw new TransformException("Unexpected character at position " + pos + ": " + expression.charAt(pos));
+ }
+ return result;
+ }
+
+ // Logical OR (lowest precedence)
+ private PlcValue parseLogicalOr() throws TransformException {
+ PlcValue left = parseLogicalAnd();
+
+ while (pos < expression.length() && peek("||")) {
+ consume("||");
+ PlcValue right = parseLogicalAnd();
+ left = new PlcBOOL(toBoolean(left) || toBoolean(right));
+ }
+
+ return left;
+ }
+
+ // Logical AND
+ private PlcValue parseLogicalAnd() throws TransformException {
+ PlcValue left = parseComparison();
+
+ while (pos < expression.length() && peek("&&")) {
+ consume("&&");
+ PlcValue right = parseComparison();
+ left = new PlcBOOL(toBoolean(left) && toBoolean(right));
+ }
+
+ return left;
+ }
+
+ // Comparison operators
+ private PlcValue parseComparison() throws TransformException {
+ PlcValue left = parseAddSub();
+
+ while (pos < expression.length()) {
+ if (peek("==")) {
+ consume("==");
+ PlcValue right = parseAddSub();
+ left = new PlcBOOL(compare(left, right) == 0);
+ } else if (peek("!=")) {
+ consume("!=");
+ PlcValue right = parseAddSub();
+ left = new PlcBOOL(compare(left, right) != 0);
+ } else if (peek("<=")) {
+ consume("<=");
+ PlcValue right = parseAddSub();
+ left = new PlcBOOL(compare(left, right) <= 0);
+ } else if (peek(">=")) {
+ consume(">=");
+ PlcValue right = parseAddSub();
+ left = new PlcBOOL(compare(left, right) >= 0);
+ } else if (peek("<")) {
+ consume("<");
+ PlcValue right = parseAddSub();
+ left = new PlcBOOL(compare(left, right) < 0);
+ } else if (peek(">")) {
+ consume(">");
+ PlcValue right = parseAddSub();
+ left = new PlcBOOL(compare(left, right) > 0);
+ } else {
+ break;
+ }
+ }
+
+ return left;
+ }
+
+ // Addition and subtraction
+ private PlcValue parseAddSub() throws TransformException {
+ PlcValue left = parseMulDiv();
+
+ while (pos < expression.length()) {
+ skipWhitespace();
+ if (pos >= expression.length()) break;
+
+ char op = expression.charAt(pos);
+ if (op == '+') {
+ pos++;
+ PlcValue right = parseMulDiv();
+ left = add(left, right);
+ } else if (op == '-' && (pos == 0 || !Character.isDigit(expression.charAt(pos - 1)))) {
+ // Only treat as subtraction if not part of a number
+ if (pos > 0 && Character.isWhitespace(expression.charAt(pos - 1))) {
+ pos++;
+ PlcValue right = parseMulDiv();
+ left = subtract(left, right);
+ } else {
+ break;
+ }
+ } else {
+ break;
+ }
+ }
+
+ return left;
+ }
+
+ // Multiplication, division, and modulo
+ private PlcValue parseMulDiv() throws TransformException {
+ PlcValue left = parseUnary();
+
+ while (pos < expression.length()) {
+ skipWhitespace();
+ if (pos >= expression.length()) break;
+
+ char op = expression.charAt(pos);
+ if (op == '*') {
+ pos++;
+ PlcValue right = parseUnary();
+ left = multiply(left, right);
+ } else if (op == '/') {
+ pos++;
+ PlcValue right = parseUnary();
+ left = divide(left, right);
+ } else if (op == '%') {
+ pos++;
+ PlcValue right = parseUnary();
+ left = modulo(left, right);
+ } else {
+ break;
+ }
+ }
+
+ return left;
+ }
+
+ // Unary operators (-, !)
+ private PlcValue parseUnary() throws TransformException {
+ skipWhitespace();
+
+ if (pos < expression.length()) {
+ char ch = expression.charAt(pos);
+ if (ch == '-') {
+ pos++;
+ PlcValue value = parseUnary();
+ return negate(value);
+ } else if (ch == '!') {
+ pos++;
+ PlcValue value = parseUnary();
+ return new PlcBOOL(!toBoolean(value));
+ }
+ }
+
+ return parsePrimary();
+ }
+
+ // Primary expressions (numbers, variables, parentheses)
+ private PlcValue parsePrimary() throws TransformException {
+ skipWhitespace();
+
+ if (pos >= expression.length()) {
+ throw new TransformException("Unexpected end of expression");
+ }
+
+ char ch = expression.charAt(pos);
+
+ // Parentheses
+ if (ch == '(') {
+ pos++;
+ PlcValue value = parseLogicalOr();
+ skipWhitespace();
+ if (pos >= expression.length() || expression.charAt(pos) != ')') {
+ throw new TransformException("Expected ')' at position " + pos);
+ }
+ pos++;
+ return value;
+ }
+
+ // Number
+ if (Character.isDigit(ch) || ch == '.') {
+ return parseNumber();
+ }
+
+ // Variable
+ if (Character.isLetter(ch) || ch == '_') {
+ return parseVariable();
+ }
+
+ throw new TransformException("Unexpected character at position " + pos + ": " + ch);
+ }
+
+ private PlcValue parseNumber() throws TransformException {
+ int start = pos;
+ boolean hasDecimal = false;
+
+ while (pos < expression.length()) {
+ char ch = expression.charAt(pos);
+ if (Character.isDigit(ch)) {
+ pos++;
+ } else if (ch == '.' && !hasDecimal) {
+ hasDecimal = true;
+ pos++;
+ } else {
+ break;
+ }
+ }
+
+ String numStr = expression.substring(start, pos);
+ try {
+ if (hasDecimal) {
+ return new PlcREAL(Double.parseDouble(numStr));
+ } else {
+ long value = Long.parseLong(numStr);
+ if (value >= Integer.MIN_VALUE && value <= Integer.MAX_VALUE) {
+ return new PlcDINT((int) value);
+ } else {
+ return new PlcLINT(value);
+ }
+ }
+ } catch (NumberFormatException e) {
+ throw new TransformException("Invalid number: " + numStr);
+ }
+ }
+
+ private PlcValue parseVariable() throws TransformException {
+ int start = pos;
+
+ while (pos < expression.length()) {
+ char ch = expression.charAt(pos);
+ if (Character.isLetterOrDigit(ch) || ch == '_') {
+ pos++;
+ } else {
+ break;
+ }
+ }
+
+ String varName = expression.substring(start, pos);
+
+ // Check for boolean literals
+ if ("true".equalsIgnoreCase(varName)) {
+ return new PlcBOOL(true);
+ } else if ("false".equalsIgnoreCase(varName)) {
+ return new PlcBOOL(false);
+ }
+
+ PlcValue value = context.get(varName);
+ if (value == null) {
+ throw new TransformException("Unknown variable: " + varName);
+ }
+
+ return value;
+ }
+
+ private void skipWhitespace() {
+ while (pos < expression.length() && Character.isWhitespace(expression.charAt(pos))) {
+ pos++;
+ }
+ }
+
+ private boolean peek(String str) {
+ skipWhitespace();
+ if (pos + str.length() > expression.length()) {
+ return false;
+ }
+ return expression.startsWith(str, pos);
+ }
+
+ private void consume(String str) {
+ skipWhitespace();
+ if (!peek(str)) {
+ throw new RuntimeException("Expected '" + str + "' at position " + pos);
+ }
+ pos += str.length();
+ }
+
+ // Arithmetic operations
+ private PlcValue add(PlcValue a, PlcValue b) throws TransformException {
+ if (isFloatingPoint(a) || isFloatingPoint(b)) {
+ return new PlcREAL(toDouble(a) + toDouble(b));
+ } else {
+ return new PlcLINT(toLong(a) + toLong(b));
+ }
+ }
+
+ private PlcValue subtract(PlcValue a, PlcValue b) throws TransformException {
+ if (isFloatingPoint(a) || isFloatingPoint(b)) {
+ return new PlcREAL(toDouble(a) - toDouble(b));
+ } else {
+ return new PlcLINT(toLong(a) - toLong(b));
+ }
+ }
+
+ private PlcValue multiply(PlcValue a, PlcValue b) throws TransformException {
+ if (isFloatingPoint(a) || isFloatingPoint(b)) {
+ return new PlcREAL(toDouble(a) * toDouble(b));
+ } else {
+ return new PlcLINT(toLong(a) * toLong(b));
+ }
+ }
+
+ private PlcValue divide(PlcValue a, PlcValue b) throws TransformException {
+ double divisor = toDouble(b);
+ if (divisor == 0) {
+ throw new TransformException("Division by zero");
+ }
+ return new PlcREAL(toDouble(a) / divisor);
+ }
+
+ private PlcValue modulo(PlcValue a, PlcValue b) throws TransformException {
+ long divisor = toLong(b);
+ if (divisor == 0) {
+ throw new TransformException("Modulo by zero");
+ }
+ return new PlcLINT(toLong(a) % divisor);
+ }
+
+ private PlcValue negate(PlcValue value) throws TransformException {
+ if (isFloatingPoint(value)) {
+ return new PlcREAL(-toDouble(value));
+ } else {
+ return new PlcLINT(-toLong(value));
+ }
+ }
+
+ private int compare(PlcValue a, PlcValue b) throws TransformException {
+ if (isFloatingPoint(a) || isFloatingPoint(b)) {
+ return Double.compare(toDouble(a), toDouble(b));
+ } else {
+ return Long.compare(toLong(a), toLong(b));
+ }
+ }
+
+ // Type conversion helpers
+ private boolean isFloatingPoint(PlcValue value) {
+ return value instanceof PlcREAL || value instanceof PlcLREAL;
+ }
+
+ private double toDouble(PlcValue value) throws TransformException {
+ if (value.isDouble()) {
+ return value.getDouble();
+ } else if (value.isFloat()) {
+ return value.getFloat();
+ } else if (value.isLong()) {
+ return value.getLong();
+ } else if (value.isInteger()) {
+ return value.getInteger();
+ } else if (value.isShort()) {
+ return value.getShort();
+ } else if (value.isByte()) {
+ return value.getByte();
+ } else if (value.isBigInteger()) {
+ return value.getBigInteger().doubleValue();
+ } else if (value.isBigDecimal()) {
+ return value.getBigDecimal().doubleValue();
+ }
+ throw new TransformException("Cannot convert to double: " + value.getClass().getSimpleName());
+ }
+
+ private long toLong(PlcValue value) throws TransformException {
+ if (value.isLong()) {
+ return value.getLong();
+ } else if (value.isInteger()) {
+ return value.getInteger();
+ } else if (value.isShort()) {
+ return value.getShort();
+ } else if (value.isByte()) {
+ return value.getByte();
+ } else if (value.isBigInteger()) {
+ return value.getBigInteger().longValue();
+ } else if (value.isDouble()) {
+ return (long) value.getDouble();
+ } else if (value.isFloat()) {
+ return (long) value.getFloat();
+ }
+ throw new TransformException("Cannot convert to long: " + value.getClass().getSimpleName());
+ }
+
+ private boolean toBoolean(PlcValue value) throws TransformException {
+ if (value.isBoolean()) {
+ return value.getBoolean();
+ }
+ // Non-zero numbers are true
+ if (value.isLong() || value.isInteger() || value.isShort() || value.isByte()) {
+ return toLong(value) != 0;
+ }
+ if (value.isDouble() || value.isFloat()) {
+ return toDouble(value) != 0.0;
+ }
+ throw new TransformException("Cannot convert to boolean: " + value.getClass().getSimpleName());
+ }
+ }
+}
diff --git a/plc4j/transports/pcap-shared/src/main/java/org/apache/plc4x/java/transport/pcap/PcapTransportConfiguration.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/TransformException.java
similarity index 59%
rename from plc4j/transports/pcap-shared/src/main/java/org/apache/plc4x/java/transport/pcap/PcapTransportConfiguration.java
rename to plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/TransformException.java
index f34b360af2..6452b32a53 100644
--- a/plc4j/transports/pcap-shared/src/main/java/org/apache/plc4x/java/transport/pcap/PcapTransportConfiguration.java
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/TransformException.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -16,22 +16,19 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.plc4x.java.transport.pcap;
-import org.apache.plc4x.java.spi.configuration.PlcTransportConfiguration;
-import org.apache.plc4x.java.utils.pcap.netty.handlers.PacketHandler;
+package org.apache.plc4x.java.tools.eventpump.transform;
-public interface PcapTransportConfiguration extends PlcTransportConfiguration {
- int NO_DEFAULT_PORT = -1;
-
- boolean getSupportVlans();
+/**
+ * Exception thrown when a value transformation fails.
+ */
+public class TransformException extends Exception {
- default int getDefaultPort() {
- return NO_DEFAULT_PORT;
+ public TransformException(String message) {
+ super(message);
}
- int getProtocolId();
-
- PacketHandler getPcapPacketHandler();
-
+ public TransformException(String message, Throwable cause) {
+ super(message, cause);
+ }
}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformer.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformer.java
new file mode 100644
index 0000000000..2e25575169
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformer.java
@@ -0,0 +1,69 @@
+/*
+ * 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.tools.eventpump.transform;
+
+import org.apache.plc4x.java.api.value.PlcValue;
+
+import java.util.Map;
+
+/**
+ * Interface for transforming PLC values using expressions.
+ * <p>
+ * Implementations can provide different expression languages, from simple arithmetic
+ * to more sophisticated languages like SpEL.
+ * <p>
+ * The transformer is invoked with an expression string and a context containing
+ * available variables (typically the tag values from a PLC read).
+ * <p>
+ * Example implementations:
+ * - SimpleExpressionEvaluator: Basic arithmetic and comparisons
+ * - SpELValueTransformer: Spring Expression Language support
+ */
+public interface ValueTransformer {
+
+ /**
+ * Transform a value using the given expression.
+ *
+ * @param expression The expression to evaluate (e.g., "value * 1.8 + 32")
+ * @param context A map of variable names to their PlcValue objects
+ * @return The transformed value
+ * @throws TransformException if the expression cannot be evaluated
+ */
+ PlcValue transform(String expression, Map<String, PlcValue> context) throws TransformException;
+
+ /**
+ * Check if this transformer supports the given expression.
+ * This allows for fallback to simpler transformers if needed.
+ *
+ * @param expression The expression to check
+ * @return true if this transformer can handle the expression
+ */
+ default boolean supports(String expression) {
+ return true;
+ }
+
+ /**
+ * Get the name of this transformer implementation.
+ * Used for registration and debugging.
+ *
+ * @return The transformer name (e.g., "simple", "spel")
+ */
+ String getName();
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerRegistry.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerRegistry.java
new file mode 100644
index 0000000000..d2a5abf4da
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerRegistry.java
@@ -0,0 +1,176 @@
+/*
+ * 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.tools.eventpump.transform;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Registry for managing ValueTransformer implementations.
+ * <p>
+ * This allows for pluggable expression evaluators. By default, the SimpleExpressionEvaluator
+ * is registered, but applications can register their own transformers (e.g., SpEL-based).
+ * <p>
+ * Thread-safe instance-based implementation. Use {@link #createDefault()} to create a registry
+ * with standard transformers pre-registered, or create an empty instance with the constructor.
+ * <p>
+ * Example usage:
+ * <pre>
+ * // Create a default registry with standard transformers
+ * ValueTransformerRegistry registry = ValueTransformerRegistry.createDefault();
+ *
+ * // Use it in a TagBatch
+ * TagBatch batch = TagBatch.builder()
+ * .withTransformerRegistry(registry)
+ * ...
+ * .build();
+ *
+ * // Or create a custom registry
+ * ValueTransformerRegistry customRegistry = new ValueTransformerRegistry();
+ * customRegistry.register(new MyCustomTransformer());
+ * </pre>
+ */
+public class ValueTransformerRegistry {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ValueTransformerRegistry.class);
+
+ private final Map<String, ValueTransformer> transformers = new ConcurrentHashMap<>();
+ private volatile String defaultTransformerName = "simple";
+
+ /**
+ * Create an empty registry with no transformers registered.
+ * Use {@link #createDefault()} if you want standard transformers pre-registered.
+ */
+ public ValueTransformerRegistry() {
+ // Empty constructor for custom registries
+ }
+
+ /**
+ * Create a registry with standard transformers pre-registered.
+ * This includes:
+ * - SimpleExpressionEvaluator (as "simple", set as default)
+ * - SpelValueTransformer (as "spel", if Spring Expression is on the classpath)
+ *
+ * @return A new registry with standard transformers
+ */
+ public static ValueTransformerRegistry createDefault() {
+ ValueTransformerRegistry registry = new ValueTransformerRegistry();
+
+ // Register the default simple evaluator
+ registry.register(new SimpleExpressionEvaluator());
+
+ return registry;
+ }
+
+
+ /**
+ * Register a transformer implementation.
+ *
+ * @param transformer The transformer to register
+ * @throws IllegalArgumentException if transformer is null or name is already registered
+ */
+ public void register(ValueTransformer transformer) {
+ if (transformer == null) {
+ throw new IllegalArgumentException("Transformer cannot be null");
+ }
+
+ String name = transformer.getName();
+ if (name == null || name.trim().isEmpty()) {
+ throw new IllegalArgumentException("Transformer name cannot be null or empty");
+ }
+
+ ValueTransformer existing = transformers.putIfAbsent(name, transformer);
+ if (existing != null) {
+ LOGGER.warn("Transformer '{}' is already registered, keeping existing", name);
+ } else {
+ LOGGER.info("Registered value transformer: {}", name);
+ }
+ }
+
+ /**
+ * Get a transformer by name.
+ *
+ * @param name The transformer name
+ * @return The transformer, or null if not found
+ */
+ public ValueTransformer get(String name) {
+ return transformers.get(name);
+ }
+
+ /**
+ * Get the default transformer.
+ *
+ * @return The default transformer
+ * @throws IllegalStateException if the default transformer is not registered
+ */
+ public ValueTransformer getDefault() {
+ ValueTransformer transformer = transformers.get(defaultTransformerName);
+ if (transformer == null) {
+ throw new IllegalStateException("Default transformer '" + defaultTransformerName + "' is not registered");
+ }
+ return transformer;
+ }
+
+ /**
+ * Set the default transformer by name.
+ *
+ * @param name The name of the transformer to use as default
+ * @throws IllegalArgumentException if the transformer is not registered
+ */
+ public void setDefault(String name) {
+ if (!transformers.containsKey(name)) {
+ throw new IllegalArgumentException("Transformer '" + name + "' is not registered");
+ }
+ this.defaultTransformerName = name;
+ LOGGER.info("Set default value transformer to: {}", name);
+ }
+
+ /**
+ * Unregister a transformer.
+ *
+ * @param name The name of the transformer to unregister
+ * @return true if the transformer was removed, false if it didn't exist
+ * @throws IllegalStateException if trying to unregister the default transformer
+ */
+ public boolean unregister(String name) {
+ if (name.equals(defaultTransformerName)) {
+ throw new IllegalStateException("Cannot unregister the default transformer");
+ }
+
+ ValueTransformer removed = transformers.remove(name);
+ if (removed != null) {
+ LOGGER.info("Unregistered value transformer: {}", name);
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Get all registered transformer names.
+ *
+ * @return A set of transformer names
+ */
+ public java.util.Set<String> getRegisteredNames() {
+ return java.util.Collections.unmodifiableSet(transformers.keySet());
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/SubscriptionTrigger.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/SubscriptionTrigger.java
new file mode 100644
index 0000000000..4dbd267aa9
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/SubscriptionTrigger.java
@@ -0,0 +1,116 @@
+/*
+ * 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.tools.eventpump.triggers;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A trigger that fires when a subscribed PLC tag value changes.
+ * <p>
+ * <b>NOTE:</b> This trigger implementation is currently a placeholder.
+ * Full subscription support requires protocol-specific implementations.
+ * <p>
+ * For now, use {@link TimerTrigger} for polling-based data collection.
+ * Subscription support will be added in a future release.
+ * <p>
+ * Example usage (when implemented):
+ * <pre>
+ * // Trigger on any value change
+ * Trigger trigger = new SubscriptionTrigger(connection, "triggerTag", "MAIN.trigger");
+ * </pre>
+ */
+public class SubscriptionTrigger implements Trigger {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionTrigger.class);
+
+ private final String tagName;
+ private final String tagAddress;
+
+ private volatile boolean running = false;
+ private volatile boolean closed = false;
+
+ /**
+ * Create a subscription trigger that fires on any value change.
+ *
+ * @param tagName The name for the subscribed tag
+ * @param tagAddress The tag address to subscribe to
+ */
+ public SubscriptionTrigger(String tagName, String tagAddress) {
+ if (tagName == null || tagName.trim().isEmpty()) {
+ throw new IllegalArgumentException("Tag name cannot be null or empty");
+ }
+ if (tagAddress == null || tagAddress.trim().isEmpty()) {
+ throw new IllegalArgumentException("Tag address cannot be null or empty");
+ }
+
+ this.tagName = tagName;
+ this.tagAddress = tagAddress;
+
+ LOGGER.debug("Created SubscriptionTrigger placeholder for tag '{}' at address '{}'", tagName, tagAddress);
+ }
+
+ @Override
+ public void start(TriggerListener listener) {
+ throw new UnsupportedOperationException(
+ "SubscriptionTrigger is not yet fully implemented. " +
+ "Use TimerTrigger for polling-based data collection. " +
+ "Subscription support will be added in a future release."
+ );
+ }
+
+ @Override
+ public void stop() {
+ // No-op for placeholder implementation
+ }
+
+ @Override
+ public boolean isRunning() {
+ return false; // Placeholder never runs
+ }
+
+ @Override
+ public String getType() {
+ return "Subscription [PLACEHOLDER] (tag=" + tagName + ", address=" + tagAddress + ")";
+ }
+
+ @Override
+ public void close() {
+ // No-op for placeholder implementation
+ }
+
+ /**
+ * Get the tag name being subscribed to.
+ *
+ * @return The tag name
+ */
+ public String getTagName() {
+ return tagName;
+ }
+
+ /**
+ * Get the tag address being subscribed to.
+ *
+ * @return The tag address
+ */
+ public String getTagAddress() {
+ return tagAddress;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTrigger.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTrigger.java
new file mode 100644
index 0000000000..23d4d8888c
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTrigger.java
@@ -0,0 +1,204 @@
+/*
+ * 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.tools.eventpump.triggers;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A trigger that fires at fixed time intervals.
+ * <p>
+ * This trigger uses a {@link Timer} to schedule periodic execution.
+ * The interval can be configured with any time unit (milliseconds, seconds, minutes, etc.).
+ * <p>
+ * Example usage:
+ * <pre>
+ * // Trigger every 5 seconds
+ * Trigger trigger = new TimerTrigger(5, TimeUnit.SECONDS);
+ * trigger.start(t -> System.out.println("Trigger fired!"));
+ * </pre>
+ */
+public class TimerTrigger implements Trigger {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TimerTrigger.class);
+
+ private final long intervalMs;
+ private final long initialDelayMs;
+ private final Timer timer;
+ private final boolean ownTimer;
+
+ private volatile TriggerListener listener;
+ private volatile TimerTask currentTask;
+ private volatile boolean running = false;
+ private volatile boolean closed = false;
+
+ /**
+ * Create a timer trigger with the specified interval and no initial delay.
+ *
+ * @param interval The interval between trigger firings
+ * @param unit The time unit of the interval
+ */
+ public TimerTrigger(long interval, TimeUnit unit) {
+ this(interval, 0, unit);
+ }
+
+ /**
+ * Create a timer trigger with the specified interval and initial delay.
+ *
+ * @param interval The interval between trigger firings
+ * @param initialDelay The initial delay before the first firing
+ * @param unit The time unit of the interval and delay
+ */
+ public TimerTrigger(long interval, long initialDelay, TimeUnit unit) {
+ this(interval, initialDelay, unit, null);
+ }
+
+ /**
+ * Create a timer trigger with a custom timer.
+ * Useful for sharing a timer across multiple triggers.
+ *
+ * @param interval The interval between trigger firings
+ * @param initialDelay The initial delay before the first firing
+ * @param unit The time unit of the interval and delay
+ * @param sharedTimer A shared timer (if null, a new daemon timer will be created)
+ */
+ public TimerTrigger(long interval, long initialDelay, TimeUnit unit, Timer sharedTimer) {
+ if (interval <= 0) {
+ throw new IllegalArgumentException("Interval must be positive");
+ }
+ if (initialDelay < 0) {
+ throw new IllegalArgumentException("Initial delay cannot be negative");
+ }
+
+ this.intervalMs = unit.toMillis(interval);
+ this.initialDelayMs = unit.toMillis(initialDelay);
+
+ if (sharedTimer != null) {
+ this.timer = sharedTimer;
+ this.ownTimer = false;
+ } else {
+ this.timer = new Timer("TimerTrigger-" + System.identityHashCode(this), true);
+ this.ownTimer = true;
+ }
+
+ LOGGER.debug("Created TimerTrigger with interval={}ms, initialDelay={}ms", intervalMs, initialDelayMs);
+ }
+
+ @Override
+ public void start(TriggerListener listener) {
+ if (listener == null) {
+ throw new IllegalArgumentException("Listener cannot be null");
+ }
+ if (closed) {
+ throw new IllegalStateException("Trigger is closed");
+ }
+ if (running) {
+ throw new IllegalStateException("Trigger is already running");
+ }
+
+ this.listener = listener;
+
+ // Create the timer task
+ currentTask = new TimerTask() {
+ @Override
+ public void run() {
+ try {
+ LOGGER.trace("Timer trigger firing");
+ TimerTrigger.this.listener.onTrigger(TimerTrigger.this);
+ } catch (Exception e) {
+ if (LOGGER.isTraceEnabled()) {
+ LOGGER.error("Error in trigger listener", e);
+ } else {
+ LOGGER.error("Error in trigger listener: {}", e.getMessage());
+ }
+ }
+ }
+ };
+
+ // Schedule the task
+ timer.scheduleAtFixedRate(currentTask, initialDelayMs, intervalMs);
+ running = true;
+
+ LOGGER.debug("Started TimerTrigger");
+ }
+
+ @Override
+ public void stop() {
+ if (!running) {
+ return;
+ }
+
+ if (currentTask != null) {
+ currentTask.cancel();
+ currentTask = null;
+ }
+
+ running = false;
+ LOGGER.debug("Stopped TimerTrigger");
+ }
+
+ @Override
+ public boolean isRunning() {
+ return running;
+ }
+
+ @Override
+ public String getType() {
+ return "Timer (interval=" + intervalMs + "ms)";
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+
+ stop();
+
+ if (ownTimer) {
+ timer.cancel();
+ }
+
+ closed = true;
+ LOGGER.debug("Closed TimerTrigger");
+ }
+
+ /**
+ * Get the interval in milliseconds.
+ *
+ * @return The interval in milliseconds
+ */
+ public long getIntervalMs() {
+ return intervalMs;
+ }
+
+ /**
+ * Get the initial delay in milliseconds.
+ *
+ * @return The initial delay in milliseconds
+ */
+ public long getInitialDelayMs() {
+ return initialDelayMs;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/Trigger.java b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/Trigger.java
new file mode 100644
index 0000000000..3d09c538ba
--- /dev/null
+++ b/plc4j/tools/event-pump/src/main/java/org/apache/plc4x/java/tools/eventpump/triggers/Trigger.java
@@ -0,0 +1,79 @@
+/*
+ * 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.tools.eventpump.triggers;
+
+/**
+ * Interface for triggers that determine when a tag batch should be fetched.
+ * <p>
+ * Triggers can be time-based, subscription-based, or expression-based.
+ * When a trigger fires, it notifies its listener to fetch the associated batch.
+ */
+public interface Trigger extends AutoCloseable {
+
+ /**
+ * Start the trigger.
+ * After starting, the trigger will fire events according to its configuration.
+ *
+ * @param listener The listener to notify when the trigger fires
+ * @throws IllegalStateException if the trigger is already started
+ */
+ void start(TriggerListener listener);
+
+ /**
+ * Stop the trigger.
+ * After stopping, the trigger will no longer fire events.
+ * The trigger can be restarted by calling start() again.
+ */
+ void stop();
+
+ /**
+ * Check if the trigger is currently running.
+ *
+ * @return true if the trigger is running, false otherwise
+ */
+ boolean isRunning();
+
+ /**
+ * Get the trigger type (for debugging/logging).
+ *
+ * @return A string describing the trigger type
+ */
+ String getType();
+
+ /**
+ * Close the trigger and release all resources.
+ * After closing, the trigger cannot be restarted.
+ */
+ @Override
+ void close();
+
+ /**
+ * Listener interface for trigger events.
+ */
+ @FunctionalInterface
+ interface TriggerListener {
+ /**
+ * Called when the trigger fires.
+ *
+ * @param trigger The trigger that fired
+ */
+ void onTrigger(Trigger trigger);
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/EventPumpTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/EventPumpTest.java
new file mode 100644
index 0000000000..40d7837325
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/EventPumpTest.java
@@ -0,0 +1,464 @@
+/*
+ * 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.tools.eventpump;
+
+import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.tools.eventpump.triggers.TimerTrigger;
+import org.apache.plc4x.java.tools.eventpump.triggers.Trigger;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for EventPump.
+ */
+class EventPumpTest {
+
+ private PlcConnectionManager connectionManager;
+ private String connectionString;
+ private PlcReadResponse mockReadResponse;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ // Create mocks for the request builders
+ PlcReadRequest.Builder mockReadBuilder = mock(PlcReadRequest.Builder.class);
+ PlcReadRequest mockReadRequest = mock(PlcReadRequest.class);
+ mockReadResponse = mock(PlcReadResponse.class);
+
+ // Setup mock behavior
+ when(mockReadBuilder.addTagAddress(any(), any())).thenReturn(mockReadBuilder);
+ when(mockReadBuilder.build()).thenReturn(mockReadRequest);
+ when(mockReadRequest.execute()).thenAnswer(inv ->
+ CompletableFuture.completedFuture(mockReadResponse));
+
+ // Use stub instead of mocking PlcConnection
+ PlcConnection connection = new StubPlcConnection(mockReadBuilder);
+
+ // Create a mock connection manager that returns our stub connection
+ connectionManager = mock(PlcConnectionManager.class);
+ when(connectionManager.getConnection(any())).thenReturn(connection);
+
+ connectionString = "test://localhost";
+ }
+
+ @Test
+ void testAddAndRemoveBatch() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ pump.addBatch(batch);
+
+ // Assert
+ assertEquals(1, pump.getBatchCount());
+ assertNotNull(pump.getBatch("batch1"));
+
+ // Remove
+ TagBatch removed = pump.removeBatch("batch1");
+ assertEquals(batch, removed);
+ assertEquals(0, pump.getBatchCount());
+ assertNull(pump.getBatch("batch1"));
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testAddDuplicateBatch() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger1 = new TimerTrigger(1, TimeUnit.SECONDS);
+ Trigger trigger2 = new TimerTrigger(1, TimeUnit.SECONDS);
+
+ TagBatch batch1 = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger1)
+ .withListener((b, r) -> {})
+ .build();
+
+ TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch1") // Same ID
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger2)
+ .withListener((b, r) -> {})
+ .build();
+
+ pump.addBatch(batch1);
+
+ // Act & Assert - Adding duplicate should fail
+ assertThrows(IllegalArgumentException.class, () -> {
+ pump.addBatch(batch2);
+ });
+
+ // Cleanup
+ trigger2.close();
+ pump.close();
+ }
+
+ @Test
+ void testStartAndStopBatch() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ pump.addBatch(batch);
+
+ // Act
+ pump.startBatch("batch1");
+
+ // Assert
+ assertTrue(batch.isStarted());
+ assertEquals(1, pump.getStartedBatchCount());
+
+ // Stop
+ pump.stopBatch("batch1");
+ assertFalse(batch.isStarted());
+ assertEquals(0, pump.getStartedBatchCount());
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testStartAll() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger1 = new TimerTrigger(1, TimeUnit.SECONDS);
+ Trigger trigger2 = new TimerTrigger(1, TimeUnit.SECONDS);
+
+ TagBatch batch1 = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger1)
+ .withListener((b, r) -> {})
+ .build();
+
+ TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch2")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger2)
+ .withListener((b, r) -> {})
+ .build();
+
+ pump.addBatch(batch1);
+ pump.addBatch(batch2);
+
+ // Act
+ pump.startAll();
+
+ // Assert
+ assertTrue(batch1.isStarted());
+ assertTrue(batch2.isStarted());
+ assertEquals(2, pump.getStartedBatchCount());
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testStopAll() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger1 = new TimerTrigger(1, TimeUnit.SECONDS);
+ Trigger trigger2 = new TimerTrigger(1, TimeUnit.SECONDS);
+
+ TagBatch batch1 = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger1)
+ .withListener((b, r) -> {})
+ .build();
+
+ TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch2")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger2)
+ .withListener((b, r) -> {})
+ .build();
+
+ pump.addBatch(batch1);
+ pump.addBatch(batch2);
+ pump.startAll();
+
+ // Act
+ pump.stopAll();
+
+ // Assert
+ assertFalse(batch1.isStarted());
+ assertFalse(batch2.isStarted());
+ assertEquals(0, pump.getStartedBatchCount());
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testGetAllBatches() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger1 = new TimerTrigger(1, TimeUnit.SECONDS);
+ Trigger trigger2 = new TimerTrigger(1, TimeUnit.SECONDS);
+
+ TagBatch batch1 = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger1)
+ .withListener((b, r) -> {})
+ .build();
+
+ TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch2")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger2)
+ .withListener((b, r) -> {})
+ .build();
+
+ pump.addBatch(batch1);
+ pump.addBatch(batch2);
+
+ // Act
+ Map<String, TagBatch> allBatches = pump.getAllBatches();
+
+ // Assert
+ assertEquals(2, allBatches.size());
+ assertTrue(allBatches.containsKey("batch1"));
+ assertTrue(allBatches.containsKey("batch2"));
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testGetBatchIds() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger1 = new TimerTrigger(1, TimeUnit.SECONDS);
+ Trigger trigger2 = new TimerTrigger(1, TimeUnit.SECONDS);
+ Trigger trigger3 = new TimerTrigger(1, TimeUnit.SECONDS);
+
+ TagBatch batch1 = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger1)
+ .withListener((b, r) -> {})
+ .build();
+
+ TagBatch batch2 = TagBatch.builder()
+ .withBatchId("batch2")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger2)
+ .withListener((b, r) -> {})
+ .build();
+
+ TagBatch batch3 = TagBatch.builder()
+ .withBatchId("batch3")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger3)
+ .withListener((b, r) -> {})
+ .build();
+
+ pump.addBatch(batch1);
+ pump.addBatch(batch2);
+ pump.addBatch(batch3);
+
+ // Act
+ java.util.Set<String> batchIds = pump.getBatchIds();
+
+ // Assert
+ assertEquals(3, batchIds.size());
+ assertTrue(batchIds.contains("batch1"));
+ assertTrue(batchIds.contains("batch2"));
+ assertTrue(batchIds.contains("batch3"));
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testClose() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ pump.addBatch(batch);
+ pump.startAll();
+
+ // Act
+ pump.close();
+
+ // Assert
+ assertTrue(pump.isClosed());
+ assertEquals(0, pump.getBatchCount());
+ assertFalse(batch.isStarted());
+ }
+
+ @Test
+ void testOperationsAfterClose() {
+ // Arrange
+ EventPump pump = new EventPump();
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "MAIN.tag1");
+
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddresses(tags)
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ pump.close();
+
+ // Act & Assert - Operations after close should fail
+ assertThrows(IllegalStateException.class, () -> pump.addBatch(batch));
+ assertThrows(IllegalStateException.class, pump::startAll);
+
+ // Cleanup
+ trigger.close();
+ }
+
+ @Test
+ void testMultipleClose() {
+ // Arrange
+ EventPump pump = new EventPump();
+
+ // Act - Multiple closes should be safe
+ pump.close();
+ pump.close();
+ pump.close();
+
+ // Assert - Should not throw exception
+ assertTrue(pump.isClosed());
+ }
+
+ @Test
+ void testStartNonExistentBatch() {
+ // Arrange
+ EventPump pump = new EventPump();
+
+ // Act & Assert - Starting non-existent batch should fail
+ assertThrows(IllegalArgumentException.class, () -> {
+ pump.startBatch("nonexistent");
+ });
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testStopNonExistentBatch() {
+ // Arrange
+ EventPump pump = new EventPump();
+
+ // Act & Assert - Stopping non-existent batch should fail
+ assertThrows(IllegalArgumentException.class, () -> {
+ pump.stopBatch("nonexistent");
+ });
+
+ // Cleanup
+ pump.close();
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/StubPlcConnection.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/StubPlcConnection.java
new file mode 100644
index 0000000000..005340b3f7
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/StubPlcConnection.java
@@ -0,0 +1,108 @@
+/*
+ * 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.tools.eventpump;
+
+import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.messages.PlcBrowseRequest;
+import org.apache.plc4x.java.api.messages.PlcPingResponse;
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
+import org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest;
+import org.apache.plc4x.java.api.messages.PlcWriteRequest;
+import org.apache.plc4x.java.api.metadata.PlcConnectionMetadata;
+import org.apache.plc4x.java.api.model.PlcTag;
+import org.apache.plc4x.java.api.value.PlcValue;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Simple stub implementation of PlcConnection for testing.
+ * This avoids the Mockito issues with mocking interfaces that extend AutoCloseable.
+ */
+public class StubPlcConnection implements PlcConnection {
+
+ private final PlcReadRequest.Builder readBuilder;
+ private boolean connected = true;
+
+ public StubPlcConnection(PlcReadRequest.Builder readBuilder) {
+ this.readBuilder = readBuilder;
+ }
+
+ @Override
+ public void connect() {
+ connected = true;
+ }
+
+ @Override
+ public boolean isConnected() {
+ return connected;
+ }
+
+ @Override
+ public void close() {
+ connected = false;
+ }
+
+ @Override
+ public PlcConnectionMetadata getMetadata() {
+ return null;
+ }
+
+ @Override
+ public Optional<PlcTag> parseTagAddress(String tagAddress) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<PlcValue> parseTagValue(PlcTag tag, Object... values) {
+ return Optional.empty();
+ }
+
+ @Override
+ public PlcReadRequest.Builder readRequestBuilder() {
+ return readBuilder;
+ }
+
+ @Override
+ public PlcWriteRequest.Builder writeRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public PlcSubscriptionRequest.Builder subscriptionRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public PlcUnsubscriptionRequest.Builder unsubscriptionRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public PlcBrowseRequest.Builder browseRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public CompletableFuture<? extends PlcPingResponse> ping() {
+ return null;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchListenerTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchListenerTest.java
new file mode 100644
index 0000000000..29c32702cf
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchListenerTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.tools.eventpump;
+
+import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.tools.eventpump.triggers.TimerTrigger;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+
+class TagBatchListenerTest {
+
+ @Test
+ void testDefaultOnErrorMethod() {
+ PlcConnectionManager mockManager = mock(PlcConnectionManager.class);
+
+ // Create a real batch for testing
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("test-batch")
+ .withConnectionManager(mockManager)
+ .withConnectionString("test://localhost")
+ .addTagAddress("test", "MAIN.test")
+ .withTrigger(new TimerTrigger(1, TimeUnit.SECONDS))
+ .build();
+
+ Exception testException = new RuntimeException("Test error");
+
+ // Create a listener that uses the default onError method
+ TagBatch.TagBatchListener listener = new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch b, PlcReadResponse response) {
+ // Not used in this test
+ }
+ };
+
+ // The default onError method should just log and not throw an exception
+ assertDoesNotThrow(() -> {
+ listener.onError(batch, testException);
+ });
+
+ batch.close();
+ }
+
+ @Test
+ void testListenerCanOverrideOnError() {
+ PlcConnectionManager mockManager = mock(PlcConnectionManager.class);
+
+ // Create a real batch for testing
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("test-batch")
+ .withConnectionManager(mockManager)
+ .withConnectionString("test://localhost")
+ .addTagAddress("test", "MAIN.test")
+ .withTrigger(new TimerTrigger(1, TimeUnit.SECONDS))
+ .build();
+
+ Exception testException = new RuntimeException("Test error");
+ final boolean[] onErrorCalled = {false};
+
+ TagBatch.TagBatchListener listener = new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch b, PlcReadResponse response) {
+ // Not used in this test
+ }
+
+ @Override
+ public void onError(TagBatch b, Throwable error) {
+ onErrorCalled[0] = true;
+ assertEquals("test-batch", b.getBatchId());
+ assertEquals(testException, error);
+ }
+ };
+
+ listener.onError(batch, testException);
+
+ assertTrue(onErrorCalled[0], "Custom onError should have been called");
+
+ batch.close();
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchTest.java
new file mode 100644
index 0000000000..561ca51090
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TagBatchTest.java
@@ -0,0 +1,840 @@
+/*
+ * 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.tools.eventpump;
+
+import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.tools.eventpump.triggers.TimerTrigger;
+import org.apache.plc4x.java.tools.eventpump.triggers.Trigger;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+/**
+ * Tests for TagBatch, focusing on tag management functionality.
+ */
+class TagBatchTest {
+
+ private PlcConnectionManager connectionManager;
+ private String connectionString;
+ private PlcReadResponse mockReadResponse;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ // Create mocks for the request builders
+ PlcReadRequest.Builder mockReadBuilder = mock(PlcReadRequest.Builder.class);
+ PlcReadRequest mockReadRequest = mock(PlcReadRequest.class);
+ mockReadResponse = mock(PlcReadResponse.class);
+
+ // Setup mock behavior
+ when(mockReadBuilder.addTagAddress(any(), any())).thenReturn(mockReadBuilder);
+ when(mockReadBuilder.build()).thenReturn(mockReadRequest);
+ when(mockReadRequest.execute()).thenAnswer(inv ->
+ CompletableFuture.completedFuture(mockReadResponse));
+
+ // Use stub instead of mocking PlcConnection
+ PlcConnection connection = new StubPlcConnection(mockReadBuilder);
+
+ // Create mock connection manager that returns our stub connection
+ connectionManager = mock(PlcConnectionManager.class);
+ when(connectionManager.getConnection(any())).thenReturn(connection);
+
+ connectionString = "test://localhost";
+ }
+
+ @Test
+ void testRemoveSingleTag() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .addTagAddress("tag2", "MAIN.tag2")
+ .addTagAddress("tag3", "MAIN.tag3")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ boolean removed = batch.removeTag("tag2");
+
+ // Assert
+ assertTrue(removed);
+ assertEquals(2, batch.getTagCount());
+ assertFalse(batch.hasTag("tag2"));
+ assertTrue(batch.hasTag("tag1"));
+ assertTrue(batch.hasTag("tag3"));
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testRemoveNonExistentTag() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ boolean removed = batch.removeTag("nonexistent");
+
+ // Assert
+ assertFalse(removed);
+ assertEquals(1, batch.getTagCount());
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testRemoveTagFromClosedBatch() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+ batch.close();
+
+ // Act & Assert
+ assertThrows(IllegalStateException.class, () -> {
+ batch.removeTag("tag1");
+ });
+ }
+
+ @Test
+ void testRemoveMultipleTags() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .addTagAddress("tag2", "MAIN.tag2")
+ .addTagAddress("tag3", "MAIN.tag3")
+ .addTagAddress("tag4", "MAIN.tag4")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ Collection<String> tagsToRemove = Arrays.asList("tag1", "tag3", "nonexistent");
+ int removedCount = batch.removeTags(tagsToRemove);
+
+ // Assert
+ assertEquals(2, removedCount);
+ assertEquals(2, batch.getTagCount());
+ assertFalse(batch.hasTag("tag1"));
+ assertTrue(batch.hasTag("tag2"));
+ assertFalse(batch.hasTag("tag3"));
+ assertTrue(batch.hasTag("tag4"));
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testRemoveTagsWithNullCollection() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ int removedCount = batch.removeTags(null);
+
+ // Assert
+ assertEquals(0, removedCount);
+ assertEquals(1, batch.getTagCount());
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testRemoveTagsFromClosedBatch() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+ batch.close();
+
+ // Act & Assert
+ assertThrows(IllegalStateException.class, () -> {
+ batch.removeTags(Arrays.asList("tag1"));
+ });
+ }
+
+ @Test
+ void testClearTags() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .addTagAddress("tag2", "MAIN.tag2")
+ .addTagAddress("tag3", "MAIN.tag3")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ batch.clearTags();
+
+ // Assert
+ assertEquals(0, batch.getTagCount());
+ assertFalse(batch.hasTag("tag1"));
+ assertFalse(batch.hasTag("tag2"));
+ assertFalse(batch.hasTag("tag3"));
+ assertTrue(batch.getTagNames().isEmpty());
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testClearTagsFromClosedBatch() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+ batch.close();
+
+ // Act & Assert
+ assertThrows(IllegalStateException.class, batch::clearTags);
+ }
+
+ @Test
+ void testHasTag() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .addTagAddress("tag2", "MAIN.tag2")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Assert
+ assertTrue(batch.hasTag("tag1"));
+ assertTrue(batch.hasTag("tag2"));
+ assertFalse(batch.hasTag("tag3"));
+ assertFalse(batch.hasTag("nonexistent"));
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testGetTagCount() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .addTagAddress("tag2", "MAIN.tag2")
+ .addTagAddress("tag3", "MAIN.tag3")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Assert
+ assertEquals(3, batch.getTagCount());
+
+ // Add more tags
+ batch.addTag("tag4", "MAIN.tag4");
+ assertEquals(4, batch.getTagCount());
+
+ // Remove tags
+ batch.removeTag("tag1");
+ assertEquals(3, batch.getTagCount());
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testGetTagNames() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .addTagAddress("tag2", "MAIN.tag2")
+ .addTagAddress("tag3", "MAIN.tag3")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ Set<String> tagNames = batch.getTagNames();
+
+ // Assert
+ assertEquals(3, tagNames.size());
+ assertTrue(tagNames.contains("tag1"));
+ assertTrue(tagNames.contains("tag2"));
+ assertTrue(tagNames.contains("tag3"));
+
+ // Test immutability
+ assertThrows(UnsupportedOperationException.class, () -> {
+ tagNames.add("tag4");
+ });
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testAddTagAtRuntime() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ batch.addTag("tag2", "MAIN.tag2");
+
+ // Assert
+ assertEquals(2, batch.getTagCount());
+ assertTrue(batch.hasTag("tag1"));
+ assertTrue(batch.hasTag("tag2"));
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testAddTagWithNullName() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act & Assert
+ assertThrows(IllegalArgumentException.class, () -> {
+ batch.addTag(null, "MAIN.tag2");
+ });
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testAddTagWithEmptyName() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act & Assert
+ assertThrows(IllegalArgumentException.class, () -> {
+ batch.addTag(" ", "MAIN.tag2");
+ });
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testAddTagWithNullAddress() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act & Assert
+ assertThrows(IllegalArgumentException.class, () -> {
+ batch.addTag("tag2", null);
+ });
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testAddTagToClosedBatch() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+ batch.close();
+
+ // Act & Assert
+ assertThrows(IllegalStateException.class, () -> {
+ batch.addTag("tag2", "MAIN.tag2");
+ });
+ }
+
+ @Test
+ void testAddMultipleTagsAtRuntime() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ Map<String, String> newTags = new HashMap<>();
+ newTags.put("tag2", "MAIN.tag2");
+ newTags.put("tag3", "MAIN.tag3");
+
+ // Act
+ batch.addTags(newTags);
+
+ // Assert
+ assertEquals(3, batch.getTagCount());
+ assertTrue(batch.hasTag("tag1"));
+ assertTrue(batch.hasTag("tag2"));
+ assertTrue(batch.hasTag("tag3"));
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testAddTagsWithNullMap() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act & Assert
+ assertThrows(IllegalArgumentException.class, () -> {
+ batch.addTags(null);
+ });
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testAddTagsWithEmptyMap() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act & Assert
+ assertThrows(IllegalArgumentException.class, () -> {
+ batch.addTags(new HashMap<>());
+ });
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testAddTagsToClosedBatch() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+ batch.close();
+
+ Map<String, String> newTags = new HashMap<>();
+ newTags.put("tag2", "MAIN.tag2");
+
+ // Act & Assert
+ assertThrows(IllegalStateException.class, () -> {
+ batch.addTags(newTags);
+ });
+ }
+
+ @Test
+ void testDynamicTagManagementWhileRunning() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ batch.start();
+ assertTrue(batch.isStarted());
+
+ // Act - Add and remove tags while running
+ batch.addTag("tag2", "MAIN.tag2");
+ batch.addTag("tag3", "MAIN.tag3");
+ assertEquals(3, batch.getTagCount());
+
+ batch.removeTag("tag1");
+ assertEquals(2, batch.getTagCount());
+
+ // Assert
+ assertFalse(batch.hasTag("tag1"));
+ assertTrue(batch.hasTag("tag2"));
+ assertTrue(batch.hasTag("tag3"));
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testGetTagsReturnsUnmodifiableMap() {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .build();
+
+ // Act
+ Map<String, String> tags = batch.getTags();
+
+ // Assert - Should not be able to modify
+ assertThrows(UnsupportedOperationException.class, () -> {
+ tags.put("tag2", "MAIN.tag2");
+ });
+
+ // Cleanup
+ batch.close();
+ }
+
+ @Test
+ void testTagBatchListenerInterface() throws Exception {
+ // Arrange
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+
+ // Create listener using anonymous class (not lambda) to ensure interface gets coverage
+ TagBatch.TagBatchListener listener = new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch batch, PlcReadResponse response) {
+ assertNotNull(batch);
+ assertNotNull(response);
+ }
+ };
+
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("batch1")
+ .withConnectionManager(connectionManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener(listener)
+ .build();
+
+ // Act - Manually invoke listener to test interface
+ listener.onTagsFetched(batch, mockReadResponse);
+
+ // Cleanup
+ batch.close();
+ }
+
+ // ========== US3: Exponential Backoff Tests ==========
+
+ /**
+ * T019: Verify that after multiple consecutive fetchTags() failures,
+ * the backoff interval increases exponentially (fewer getConnection calls).
+ */
+ @Test
+ void testBackoffOnConsecutiveFailures() throws Exception {
+ // Arrange — make getConnection throw to simulate network outage
+ PlcConnectionManager failingManager = mock(PlcConnectionManager.class);
+ when(failingManager.getConnection(any())).thenThrow(new RuntimeException("Connection refused"));
+
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("backoff-test")
+ .withConnectionManager(failingManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .withInitialBackoffMs(50)
+ .withMaxBackoffMs(500)
+ .build();
+
+ // Act — first failure: should connect and fail
+ batch.fetchTags();
+ assertEquals(1, batch.getConsecutiveFailures());
+ verify(failingManager, times(1)).getConnection(any());
+
+ // Second call immediately — should be skipped (within backoff window)
+ batch.fetchTags();
+ assertEquals(1, batch.getConsecutiveFailures()); // Unchanged — skipped
+ verify(failingManager, times(1)).getConnection(any()); // No new call
+
+ // Wait for first backoff to expire (50ms), then try again
+ Thread.sleep(100);
+ batch.fetchTags();
+ assertEquals(2, batch.getConsecutiveFailures()); // New failure
+ verify(failingManager, times(2)).getConnection(any());
+
+ // Wait for second backoff (100ms), then try again
+ Thread.sleep(150);
+ batch.fetchTags();
+ assertEquals(3, batch.getConsecutiveFailures()); // New failure
+ verify(failingManager, times(3)).getConnection(any());
+
+ // Verify backoff is increasing — next allowed time is further out
+ assertTrue(batch.getNextAllowedFetchTimeMs() > System.currentTimeMillis());
+
+ // Cleanup
+ batch.close();
+ }
+
+ /**
+ * T020: Verify that after a successful fetchTags() call, the backoff resets to zero
+ * and normal polling resumes immediately.
+ */
+ @Test
+ void testBackoffResetOnSuccess() throws Exception {
+ // Arrange — start with a failing connection, then switch to success
+ PlcConnectionManager switchableManager = mock(PlcConnectionManager.class);
+ when(switchableManager.getConnection(any())).thenThrow(new RuntimeException("Connection refused"));
+
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("backoff-reset-test")
+ .withConnectionManager(switchableManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .withInitialBackoffMs(50)
+ .withMaxBackoffMs(500)
+ .build();
+
+ // Fail a few times to build up backoff
+ batch.fetchTags(); // Fail 1
+ Thread.sleep(100);
+ batch.fetchTags(); // Fail 2
+ assertEquals(2, batch.getConsecutiveFailures());
+ assertTrue(batch.getNextAllowedFetchTimeMs() > 0);
+
+ // Wait for backoff to expire
+ Thread.sleep(200);
+
+ // Switch to a working connection — use doReturn to override existing thenThrow
+ PlcReadRequest.Builder mockBuilder = mock(PlcReadRequest.Builder.class);
+ PlcReadRequest mockRequest = mock(PlcReadRequest.class);
+ when(mockBuilder.addTagAddress(any(), any())).thenReturn(mockBuilder);
+ when(mockBuilder.build()).thenReturn(mockRequest);
+ when(mockRequest.execute()).thenAnswer(inv ->
+ CompletableFuture.completedFuture(mockReadResponse));
+ PlcConnection workingConnection = new StubPlcConnection(mockBuilder);
+ doReturn(workingConnection).when(switchableManager).getConnection(any());
+
+ // Act — successful fetch
+ CompletableFuture<Void> result = batch.fetchTags();
+ result.get(2, TimeUnit.SECONDS); // Wait for completion
+
+ // Assert — backoff should be reset
+ assertEquals(0, batch.getConsecutiveFailures());
+ assertEquals(0, batch.getNextAllowedFetchTimeMs());
+
+ // Cleanup
+ batch.close();
+ }
+
+ /**
+ * T021: Verify that the backoff interval never exceeds the configured maximum.
+ */
+ @Test
+ void testBackoffMaximumCap() throws Exception {
+ // Arrange
+ PlcConnectionManager failingManager = mock(PlcConnectionManager.class);
+ when(failingManager.getConnection(any())).thenThrow(new RuntimeException("Connection refused"));
+
+ long maxBackoffMs = 200;
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("backoff-cap-test")
+ .withConnectionManager(failingManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .withInitialBackoffMs(50)
+ .withMaxBackoffMs(maxBackoffMs)
+ .build();
+
+ // Act — fail many times to hit the cap
+ for (int i = 0; i < 10; i++) {
+ batch.fetchTags();
+ // Wait for the backoff to expire before the next attempt
+ Thread.sleep(maxBackoffMs + 50);
+ }
+
+ // Assert — consecutive failures should be 10 but backoff capped at maxBackoffMs
+ assertEquals(10, batch.getConsecutiveFailures());
+ // The next allowed time should be at most maxBackoffMs from now
+ long maxAllowed = System.currentTimeMillis() + maxBackoffMs;
+ assertTrue(batch.getNextAllowedFetchTimeMs() <= maxAllowed,
+ "Backoff should not exceed maxBackoffMs (" + maxBackoffMs + "ms)");
+
+ // Cleanup
+ batch.close();
+ }
+
+ /**
+ * T022: Verify that trigger invocations falling within the backoff window are skipped
+ * (return immediately without attempting connection).
+ */
+ @Test
+ void testBackoffSkipsTrigger() throws Exception {
+ // Arrange
+ PlcConnectionManager failingManager = mock(PlcConnectionManager.class);
+ when(failingManager.getConnection(any())).thenThrow(new RuntimeException("Connection refused"));
+
+ Trigger trigger = new TimerTrigger(1, TimeUnit.SECONDS);
+ TagBatch batch = TagBatch.builder()
+ .withBatchId("backoff-skip-test")
+ .withConnectionManager(failingManager)
+ .withConnectionString(connectionString)
+ .addTagAddress("tag1", "MAIN.tag1")
+ .withTrigger(trigger)
+ .withListener((b, r) -> {})
+ .withInitialBackoffMs(500) // Long backoff window
+ .withMaxBackoffMs(5000)
+ .build();
+
+ // First call fails and enters backoff
+ batch.fetchTags();
+ verify(failingManager, times(1)).getConnection(any());
+ assertEquals(1, batch.getConsecutiveFailures());
+
+ // Multiple rapid calls — all should be skipped
+ for (int i = 0; i < 5; i++) {
+ CompletableFuture<Void> result = batch.fetchTags();
+ // Should complete immediately with null (skipped)
+ assertNotNull(result);
+ assertNull(result.get(1, TimeUnit.SECONDS));
+ }
+
+ // Verify no additional connection attempts were made
+ verify(failingManager, times(1)).getConnection(any());
+ // Failure count should still be 1 — skipped triggers don't increment it
+ assertEquals(1, batch.getConsecutiveFailures());
+
+ // Cleanup
+ batch.close();
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TransformedPlcReadResponseTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TransformedPlcReadResponseTest.java
new file mode 100644
index 0000000000..22c5317428
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/TransformedPlcReadResponseTest.java
@@ -0,0 +1,882 @@
+/*
+ * 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.tools.eventpump;
+
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.api.metadata.Metadata;
+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.values.PlcBOOL;
+import org.apache.plc4x.java.spi.values.PlcBYTE;
+import org.apache.plc4x.java.spi.values.PlcDATE;
+import org.apache.plc4x.java.spi.values.PlcDATE_AND_TIME;
+import org.apache.plc4x.java.spi.values.PlcDINT;
+import org.apache.plc4x.java.spi.values.PlcINT;
+import org.apache.plc4x.java.spi.values.PlcLINT;
+import org.apache.plc4x.java.spi.values.PlcLREAL;
+import org.apache.plc4x.java.spi.values.PlcREAL;
+import org.apache.plc4x.java.spi.values.PlcSTRING;
+import org.apache.plc4x.java.spi.values.PlcTIME_OF_DAY;
+import org.apache.plc4x.java.spi.values.PlcULINT;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.*;
+
+class TransformedPlcReadResponseTest {
+
+ private PlcReadResponse mockDelegate;
+ private Map<String, PlcValue> transformedValues;
+ private TransformedPlcReadResponse transformedResponse;
+
+ @BeforeEach
+ void setUp() {
+ mockDelegate = mock(PlcReadResponse.class);
+ transformedValues = new HashMap<>();
+ }
+
+ @Test
+ void testGetRequest() {
+ PlcReadRequest mockRequest = mock(PlcReadRequest.class);
+ when(mockDelegate.getRequest()).thenReturn(mockRequest);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(mockRequest, transformedResponse.getRequest());
+ verify(mockDelegate).getRequest();
+ }
+
+ @Test
+ void testGetTagNames() {
+ Collection<String> tagNames = Arrays.asList("tag1", "tag2");
+ when(mockDelegate.getTagNames()).thenReturn(tagNames);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(tagNames, transformedResponse.getTagNames());
+ verify(mockDelegate).getTagNames();
+ }
+
+ @Test
+ void testGetTag() {
+ PlcTag mockTag = mock(PlcTag.class);
+ when(mockDelegate.getTag("tag1")).thenReturn(mockTag);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(mockTag, transformedResponse.getTag("tag1"));
+ verify(mockDelegate).getTag("tag1");
+ }
+
+ @Test
+ void testGetResponseCode() {
+ when(mockDelegate.getResponseCode("tag1")).thenReturn(PlcResponseCode.OK);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(PlcResponseCode.OK, transformedResponse.getResponseCode("tag1"));
+ verify(mockDelegate).getResponseCode("tag1");
+ }
+
+ @Test
+ void testGetTagMetadata() {
+ Metadata mockMetadata = mock(Metadata.class);
+ when(mockDelegate.getTagMetadata("tag1")).thenReturn(mockMetadata);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(mockMetadata, transformedResponse.getTagMetadata("tag1"));
+ verify(mockDelegate).getTagMetadata("tag1");
+ }
+
+ @Test
+ void testGetPlcValueWithTransformation() {
+ PlcValue originalValue = new PlcDINT(10);
+ PlcValue transformedValue = new PlcDINT(20);
+
+ when(mockDelegate.getPlcValue("tag1")).thenReturn(originalValue);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(transformedValue, transformedResponse.getPlcValue("tag1"));
+ verify(mockDelegate, never()).getPlcValue("tag1");
+ }
+
+ @Test
+ void testGetPlcValueWithoutTransformation() {
+ PlcValue originalValue = new PlcDINT(10);
+
+ when(mockDelegate.getPlcValue("tag1")).thenReturn(originalValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(originalValue, transformedResponse.getPlcValue("tag1"));
+ verify(mockDelegate).getPlcValue("tag1");
+ }
+
+ @Test
+ void testGetAsPlcValue() {
+ PlcValue mockValue = mock(PlcValue.class);
+ when(mockDelegate.getAsPlcValue()).thenReturn(mockValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(mockValue, transformedResponse.getAsPlcValue());
+ verify(mockDelegate).getAsPlcValue();
+ }
+
+ @Test
+ void testGetNumberOfValues() {
+ when(mockDelegate.getNumberOfValues("tag1")).thenReturn(5);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(5, transformedResponse.getNumberOfValues("tag1"));
+ verify(mockDelegate).getNumberOfValues("tag1");
+ }
+
+ @Test
+ void testGetObjectWithTransformation() {
+ PlcValue transformedValue = new PlcDINT(42);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(42, transformedResponse.getObject("tag1"));
+ }
+
+ @Test
+ void testGetObjectWithoutTransformation() {
+ when(mockDelegate.getObject("tag1")).thenReturn(42);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(42, transformedResponse.getObject("tag1"));
+ verify(mockDelegate).getObject("tag1");
+ }
+
+ @Test
+ void testGetObjectWithIndex() {
+ when(mockDelegate.getObject("tag1", 0)).thenReturn(42);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(42, transformedResponse.getObject("tag1", 0));
+ verify(mockDelegate).getObject("tag1", 0);
+ }
+
+ @Test
+ void testGetAllObjects() {
+ Collection<Object> objects = Arrays.asList(1, 2, 3);
+ when(mockDelegate.getAllObjects("tag1")).thenReturn(objects);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(objects, transformedResponse.getAllObjects("tag1"));
+ verify(mockDelegate).getAllObjects("tag1");
+ }
+
+ // Boolean tests
+ @Test
+ void testIsValidBoolean() {
+ when(mockDelegate.isValidBoolean("tag1")).thenReturn(true);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertTrue(transformedResponse.isValidBoolean("tag1"));
+ verify(mockDelegate).isValidBoolean("tag1");
+ }
+
+ @Test
+ void testIsValidBooleanWithIndex() {
+ when(mockDelegate.isValidBoolean("tag1", 0)).thenReturn(true);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertTrue(transformedResponse.isValidBoolean("tag1", 0));
+ verify(mockDelegate).isValidBoolean("tag1", 0);
+ }
+
+ @Test
+ void testGetBooleanWithTransformation() {
+ PlcValue transformedValue = new PlcBOOL(true);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertTrue(transformedResponse.getBoolean("tag1"));
+ }
+
+ @Test
+ void testGetBooleanWithoutTransformation() {
+ when(mockDelegate.getBoolean("tag1")).thenReturn(true);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertTrue(transformedResponse.getBoolean("tag1"));
+ verify(mockDelegate).getBoolean("tag1");
+ }
+
+ @Test
+ void testGetBooleanWithIndex() {
+ when(mockDelegate.getBoolean("tag1", 0)).thenReturn(true);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertTrue(transformedResponse.getBoolean("tag1", 0));
+ verify(mockDelegate).getBoolean("tag1", 0);
+ }
+
+ @Test
+ void testGetAllBooleans() {
+ Collection<Boolean> booleans = Arrays.asList(true, false);
+ when(mockDelegate.getAllBooleans("tag1")).thenReturn(booleans);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(booleans, transformedResponse.getAllBooleans("tag1"));
+ verify(mockDelegate).getAllBooleans("tag1");
+ }
+
+ // Byte tests
+ @Test
+ void testIsValidByte() {
+ when(mockDelegate.isValidByte("tag1")).thenReturn(true);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertTrue(transformedResponse.isValidByte("tag1"));
+ }
+
+ @Test
+ void testIsValidByteWithIndex() {
+ when(mockDelegate.isValidByte("tag1", 0)).thenReturn(true);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertTrue(transformedResponse.isValidByte("tag1", 0));
+ }
+
+ @Test
+ void testGetByteWithTransformation() {
+ PlcValue transformedValue = new PlcBYTE((byte) 42);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals((byte) 42, transformedResponse.getByte("tag1"));
+ }
+
+ @Test
+ void testGetByteWithoutTransformation() {
+ when(mockDelegate.getByte("tag1")).thenReturn((byte) 42);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals((byte) 42, transformedResponse.getByte("tag1"));
+ }
+
+ @Test
+ void testGetByteWithIndex() {
+ when(mockDelegate.getByte("tag1", 0)).thenReturn((byte) 42);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals((byte) 42, transformedResponse.getByte("tag1", 0));
+ }
+
+ @Test
+ void testGetAllBytes() {
+ Collection<Byte> bytes = Arrays.asList((byte) 1, (byte) 2);
+ when(mockDelegate.getAllBytes("tag1")).thenReturn(bytes);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(bytes, transformedResponse.getAllBytes("tag1"));
+ }
+
+ // Short tests
+ @Test
+ void testGetShortWithTransformation() {
+ PlcValue transformedValue = new PlcINT((short) 100);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals((short) 100, transformedResponse.getShort("tag1"));
+ }
+
+ @Test
+ void testGetAllShorts() {
+ Collection<Short> shorts = Arrays.asList((short) 1, (short) 2);
+ when(mockDelegate.getAllShorts("tag1")).thenReturn(shorts);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(shorts, transformedResponse.getAllShorts("tag1"));
+ }
+
+ // Integer tests
+ @Test
+ void testGetIntegerWithTransformation() {
+ PlcValue transformedValue = new PlcDINT(1000);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(1000, transformedResponse.getInteger("tag1"));
+ }
+
+ @Test
+ void testGetAllIntegers() {
+ Collection<Integer> integers = Arrays.asList(1, 2, 3);
+ when(mockDelegate.getAllIntegers("tag1")).thenReturn(integers);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(integers, transformedResponse.getAllIntegers("tag1"));
+ }
+
+ // Long tests
+ @Test
+ void testGetLongWithTransformation() {
+ PlcValue transformedValue = new PlcLINT(10000L);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(10000L, transformedResponse.getLong("tag1"));
+ }
+
+ @Test
+ void testGetAllLongs() {
+ Collection<Long> longs = Arrays.asList(1L, 2L, 3L);
+ when(mockDelegate.getAllLongs("tag1")).thenReturn(longs);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(longs, transformedResponse.getAllLongs("tag1"));
+ }
+
+ // Float tests
+ @Test
+ void testGetFloatWithTransformation() {
+ PlcValue transformedValue = new PlcREAL(3.14f);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(3.14f, transformedResponse.getFloat("tag1"), 0.001);
+ }
+
+ @Test
+ void testGetAllFloats() {
+ Collection<Float> floats = Arrays.asList(1.0f, 2.0f);
+ when(mockDelegate.getAllFloats("tag1")).thenReturn(floats);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(floats, transformedResponse.getAllFloats("tag1"));
+ }
+
+ // Double tests
+ @Test
+ void testGetDoubleWithTransformation() {
+ PlcValue transformedValue = new PlcLREAL(3.14159);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(3.14159, transformedResponse.getDouble("tag1"), 0.00001);
+ }
+
+ @Test
+ void testGetAllDoubles() {
+ Collection<Double> doubles = Arrays.asList(1.0, 2.0);
+ when(mockDelegate.getAllDoubles("tag1")).thenReturn(doubles);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(doubles, transformedResponse.getAllDoubles("tag1"));
+ }
+
+ // BigInteger tests
+ @Test
+ void testGetBigIntegerWithTransformation() {
+ PlcValue transformedValue = new PlcULINT(BigInteger.valueOf(123456789));
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(BigInteger.valueOf(123456789), transformedResponse.getBigInteger("tag1"));
+ }
+
+ @Test
+ void testGetAllBigIntegers() {
+ Collection<BigInteger> bigIntegers = Arrays.asList(BigInteger.ONE, BigInteger.TEN);
+ when(mockDelegate.getAllBigIntegers("tag1")).thenReturn(bigIntegers);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(bigIntegers, transformedResponse.getAllBigIntegers("tag1"));
+ }
+
+ // BigDecimal tests
+ @Test
+ void testGetBigDecimalWithTransformation() {
+ PlcValue transformedValue = new PlcLREAL(3.14159);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ BigDecimal result = transformedResponse.getBigDecimal("tag1");
+ assertEquals(3.14159, result.doubleValue(), 0.00001);
+ }
+
+ @Test
+ void testGetAllBigDecimals() {
+ Collection<BigDecimal> bigDecimals = Arrays.asList(BigDecimal.ONE, BigDecimal.TEN);
+ when(mockDelegate.getAllBigDecimals("tag1")).thenReturn(bigDecimals);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(bigDecimals, transformedResponse.getAllBigDecimals("tag1"));
+ }
+
+ // String tests
+ @Test
+ void testGetStringWithTransformation() {
+ PlcValue transformedValue = new PlcSTRING("transformed");
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals("transformed", transformedResponse.getString("tag1"));
+ }
+
+ @Test
+ void testGetAllStrings() {
+ Collection<String> strings = Arrays.asList("a", "b", "c");
+ when(mockDelegate.getAllStrings("tag1")).thenReturn(strings);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(strings, transformedResponse.getAllStrings("tag1"));
+ }
+
+ // Time tests
+ @Test
+ void testGetTimeWithTransformation() {
+ LocalTime time = LocalTime.of(12, 30);
+ PlcValue transformedValue = new PlcTIME_OF_DAY(time);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(time, transformedResponse.getTime("tag1"));
+ }
+
+ @Test
+ void testGetAllTimes() {
+ Collection<LocalTime> times = Arrays.asList(LocalTime.now(), LocalTime.NOON);
+ when(mockDelegate.getAllTimes("tag1")).thenReturn(times);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(times, transformedResponse.getAllTimes("tag1"));
+ }
+
+ // Date tests
+ @Test
+ void testGetDateWithTransformation() {
+ LocalDate date = LocalDate.of(2024, 10, 24);
+ PlcValue transformedValue = new PlcDATE(date);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(date, transformedResponse.getDate("tag1"));
+ }
+
+ @Test
+ void testGetAllDates() {
+ Collection<LocalDate> dates = Arrays.asList(LocalDate.now(), LocalDate.of(2024, 1, 1));
+ when(mockDelegate.getAllDates("tag1")).thenReturn(dates);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(dates, transformedResponse.getAllDates("tag1"));
+ }
+
+ // DateTime tests
+ @Test
+ void testGetDateTimeWithTransformation() {
+ LocalDateTime dateTime = LocalDateTime.of(2024, 10, 24, 12, 30);
+ PlcValue transformedValue = new PlcDATE_AND_TIME(dateTime);
+ transformedValues.put("tag1", transformedValue);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(dateTime, transformedResponse.getDateTime("tag1"));
+ }
+
+ @Test
+ void testGetAllDateTimes() {
+ Collection<LocalDateTime> dateTimes = Arrays.asList(LocalDateTime.now(), LocalDateTime.of(2024, 1, 1, 0, 0));
+ when(mockDelegate.getAllDateTimes("tag1")).thenReturn(dateTimes);
+
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+
+ assertEquals(dateTimes, transformedResponse.getAllDateTimes("tag1"));
+ }
+
+ // Validation methods
+ @Test
+ void testIsValidShort() {
+ when(mockDelegate.isValidShort("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidShort("tag1"));
+ }
+
+ @Test
+ void testIsValidInteger() {
+ when(mockDelegate.isValidInteger("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidInteger("tag1"));
+ }
+
+ @Test
+ void testIsValidBigInteger() {
+ when(mockDelegate.isValidBigInteger("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidBigInteger("tag1"));
+ }
+
+ @Test
+ void testIsValidLong() {
+ when(mockDelegate.isValidLong("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidLong("tag1"));
+ }
+
+ @Test
+ void testIsValidFloat() {
+ when(mockDelegate.isValidFloat("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidFloat("tag1"));
+ }
+
+ @Test
+ void testIsValidDouble() {
+ when(mockDelegate.isValidDouble("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidDouble("tag1"));
+ }
+
+ @Test
+ void testIsValidBigDecimal() {
+ when(mockDelegate.isValidBigDecimal("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidBigDecimal("tag1"));
+ }
+
+ @Test
+ void testIsValidString() {
+ when(mockDelegate.isValidString("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidString("tag1"));
+ }
+
+ @Test
+ void testIsValidTime() {
+ when(mockDelegate.isValidTime("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidTime("tag1"));
+ }
+
+ @Test
+ void testIsValidDate() {
+ when(mockDelegate.isValidDate("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidDate("tag1"));
+ }
+
+ @Test
+ void testIsValidDateTime() {
+ when(mockDelegate.isValidDateTime("tag1")).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidDateTime("tag1"));
+ }
+
+ // Validation with index methods
+ @Test
+ void testIsValidShortWithIndex() {
+ when(mockDelegate.isValidShort("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidShort("tag1", 0));
+ }
+
+ @Test
+ void testIsValidIntegerWithIndex() {
+ when(mockDelegate.isValidInteger("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidInteger("tag1", 0));
+ }
+
+ @Test
+ void testIsValidBigIntegerWithIndex() {
+ when(mockDelegate.isValidBigInteger("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidBigInteger("tag1", 0));
+ }
+
+ @Test
+ void testIsValidLongWithIndex() {
+ when(mockDelegate.isValidLong("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidLong("tag1", 0));
+ }
+
+ @Test
+ void testIsValidFloatWithIndex() {
+ when(mockDelegate.isValidFloat("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidFloat("tag1", 0));
+ }
+
+ @Test
+ void testIsValidDoubleWithIndex() {
+ when(mockDelegate.isValidDouble("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidDouble("tag1", 0));
+ }
+
+ @Test
+ void testIsValidBigDecimalWithIndex() {
+ when(mockDelegate.isValidBigDecimal("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidBigDecimal("tag1", 0));
+ }
+
+ @Test
+ void testIsValidStringWithIndex() {
+ when(mockDelegate.isValidString("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidString("tag1", 0));
+ }
+
+ @Test
+ void testIsValidTimeWithIndex() {
+ when(mockDelegate.isValidTime("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidTime("tag1", 0));
+ }
+
+ @Test
+ void testIsValidDateWithIndex() {
+ when(mockDelegate.isValidDate("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidDate("tag1", 0));
+ }
+
+ @Test
+ void testIsValidDateTimeWithIndex() {
+ when(mockDelegate.isValidDateTime("tag1", 0)).thenReturn(true);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertTrue(transformedResponse.isValidDateTime("tag1", 0));
+ }
+
+ // Get with index methods
+ @Test
+ void testGetShortWithIndex() {
+ when(mockDelegate.getShort("tag1", 0)).thenReturn((short) 100);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals((short) 100, transformedResponse.getShort("tag1", 0));
+ }
+
+ @Test
+ void testGetIntegerWithIndex() {
+ when(mockDelegate.getInteger("tag1", 0)).thenReturn(1000);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(1000, transformedResponse.getInteger("tag1", 0));
+ }
+
+ @Test
+ void testGetBigIntegerWithIndex() {
+ when(mockDelegate.getBigInteger("tag1", 0)).thenReturn(BigInteger.TEN);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(BigInteger.TEN, transformedResponse.getBigInteger("tag1", 0));
+ }
+
+ @Test
+ void testGetLongWithIndex() {
+ when(mockDelegate.getLong("tag1", 0)).thenReturn(10000L);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(10000L, transformedResponse.getLong("tag1", 0));
+ }
+
+ @Test
+ void testGetFloatWithIndex() {
+ when(mockDelegate.getFloat("tag1", 0)).thenReturn(3.14f);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(3.14f, transformedResponse.getFloat("tag1", 0), 0.001);
+ }
+
+ @Test
+ void testGetDoubleWithIndex() {
+ when(mockDelegate.getDouble("tag1", 0)).thenReturn(3.14159);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(3.14159, transformedResponse.getDouble("tag1", 0), 0.00001);
+ }
+
+ @Test
+ void testGetBigDecimalWithIndex() {
+ when(mockDelegate.getBigDecimal("tag1", 0)).thenReturn(BigDecimal.TEN);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(BigDecimal.TEN, transformedResponse.getBigDecimal("tag1", 0));
+ }
+
+ @Test
+ void testGetStringWithIndex() {
+ when(mockDelegate.getString("tag1", 0)).thenReturn("test");
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals("test", transformedResponse.getString("tag1", 0));
+ }
+
+ @Test
+ void testGetTimeWithIndex() {
+ LocalTime time = LocalTime.NOON;
+ when(mockDelegate.getTime("tag1", 0)).thenReturn(time);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(time, transformedResponse.getTime("tag1", 0));
+ }
+
+ @Test
+ void testGetDateWithIndex() {
+ LocalDate date = LocalDate.now();
+ when(mockDelegate.getDate("tag1", 0)).thenReturn(date);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(date, transformedResponse.getDate("tag1", 0));
+ }
+
+ @Test
+ void testGetDateTimeWithIndex() {
+ LocalDateTime dateTime = LocalDateTime.now();
+ when(mockDelegate.getDateTime("tag1", 0)).thenReturn(dateTime);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(dateTime, transformedResponse.getDateTime("tag1", 0));
+ }
+
+ // Get without transformation methods
+ @Test
+ void testGetShortWithoutTransformation() {
+ when(mockDelegate.getShort("tag1")).thenReturn((short) 100);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals((short) 100, transformedResponse.getShort("tag1"));
+ }
+
+ @Test
+ void testGetIntegerWithoutTransformation() {
+ when(mockDelegate.getInteger("tag1")).thenReturn(1000);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(1000, transformedResponse.getInteger("tag1"));
+ }
+
+ @Test
+ void testGetBigIntegerWithoutTransformation() {
+ when(mockDelegate.getBigInteger("tag1")).thenReturn(BigInteger.TEN);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(BigInteger.TEN, transformedResponse.getBigInteger("tag1"));
+ }
+
+ @Test
+ void testGetLongWithoutTransformation() {
+ when(mockDelegate.getLong("tag1")).thenReturn(10000L);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(10000L, transformedResponse.getLong("tag1"));
+ }
+
+ @Test
+ void testGetFloatWithoutTransformation() {
+ when(mockDelegate.getFloat("tag1")).thenReturn(3.14f);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(3.14f, transformedResponse.getFloat("tag1"), 0.001);
+ }
+
+ @Test
+ void testGetDoubleWithoutTransformation() {
+ when(mockDelegate.getDouble("tag1")).thenReturn(3.14159);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(3.14159, transformedResponse.getDouble("tag1"), 0.00001);
+ }
+
+ @Test
+ void testGetBigDecimalWithoutTransformation() {
+ when(mockDelegate.getBigDecimal("tag1")).thenReturn(BigDecimal.TEN);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(BigDecimal.TEN, transformedResponse.getBigDecimal("tag1"));
+ }
+
+ @Test
+ void testGetStringWithoutTransformation() {
+ when(mockDelegate.getString("tag1")).thenReturn("test");
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals("test", transformedResponse.getString("tag1"));
+ }
+
+ @Test
+ void testGetTimeWithoutTransformation() {
+ LocalTime time = LocalTime.NOON;
+ when(mockDelegate.getTime("tag1")).thenReturn(time);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(time, transformedResponse.getTime("tag1"));
+ }
+
+ @Test
+ void testGetDateWithoutTransformation() {
+ LocalDate date = LocalDate.now();
+ when(mockDelegate.getDate("tag1")).thenReturn(date);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(date, transformedResponse.getDate("tag1"));
+ }
+
+ @Test
+ void testGetDateTimeWithoutTransformation() {
+ LocalDateTime dateTime = LocalDateTime.now();
+ when(mockDelegate.getDateTime("tag1")).thenReturn(dateTime);
+ transformedResponse = new TransformedPlcReadResponse(mockDelegate, transformedValues);
+ assertEquals(dateTime, transformedResponse.getDateTime("tag1"));
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpConfigurationTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpConfigurationTest.java
new file mode 100644
index 0000000000..61882ebeb4
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpConfigurationTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.tools.eventpump.config;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for EventPumpConfiguration serialization/deserialization.
+ */
+class EventPumpConfigurationTest {
+
+ @TempDir
+ File tempDir;
+
+ @Test
+ void testJsonSerialization() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = createTestConfiguration();
+ File jsonFile = new File(tempDir, "config.json");
+
+ // Act - Save
+ config.toJson(jsonFile);
+
+ // Assert - File exists
+ assertTrue(jsonFile.exists());
+
+ // Act - Load
+ EventPumpConfiguration loaded = EventPumpConfiguration.fromJson(jsonFile);
+
+ // Assert - Content matches
+ assertConfigurationEquals(config, loaded);
+ }
+
+ @Test
+ void testYamlSerialization() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = createTestConfiguration();
+ File yamlFile = new File(tempDir, "config.yaml");
+
+ // Act - Save
+ config.toYaml(yamlFile);
+
+ // Assert - File exists
+ assertTrue(yamlFile.exists());
+
+ // Act - Load
+ EventPumpConfiguration loaded = EventPumpConfiguration.fromYaml(yamlFile);
+
+ // Assert - Content matches
+ assertConfigurationEquals(config, loaded);
+ }
+
+ @Test
+ void testXmlSerialization() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = createTestConfiguration();
+ File xmlFile = new File(tempDir, "config.xml");
+
+ // Act - Save
+ config.toXml(xmlFile);
+
+ // Assert - File exists
+ assertTrue(xmlFile.exists());
+
+ // Act - Load
+ EventPumpConfiguration loaded = EventPumpConfiguration.fromXml(xmlFile);
+
+ // Assert - Content matches
+ assertConfigurationEquals(config, loaded);
+ }
+
+ private EventPumpConfiguration createTestConfiguration() {
+ EventPumpConfiguration config = new EventPumpConfiguration();
+
+ // Add connection
+ ConnectionConfiguration conn = new ConnectionConfiguration();
+ conn.setId("plc1");
+ conn.setUrl("ads:tcp://192.168.1.1");
+ conn.setUsername("admin");
+ conn.setPassword("secret");
+ config.getConnections().add(conn);
+
+ // Add batch with timer trigger
+ BatchConfiguration batch1 = new BatchConfiguration();
+ batch1.setId("batch1");
+ batch1.setConnectionId("plc1");
+
+ Map<String, String> tags1 = new HashMap<>();
+ tags1.put("temperature", "MAIN.temperature");
+ tags1.put("pressure", "MAIN.pressure");
+ batch1.setSimpleTags(tags1);
+
+ TriggerConfiguration trigger1 = new TriggerConfiguration();
+ trigger1.setType("timer");
+ trigger1.setIntervalSeconds(5L);
+ trigger1.setInitialDelaySeconds(2L);
+ batch1.setTrigger(trigger1);
+
+ config.getBatches().add(batch1);
+
+ // Add batch with subscription trigger
+ BatchConfiguration batch2 = new BatchConfiguration();
+ batch2.setId("batch2");
+ batch2.setConnectionId("plc1");
+
+ Map<String, String> tags2 = new HashMap<>();
+ tags2.put("status", "MAIN.status");
+ batch2.setSimpleTags(tags2);
+
+ TriggerConfiguration trigger2 = new TriggerConfiguration();
+ trigger2.setType("subscription");
+ trigger2.setTagName("triggerTag");
+ trigger2.setTagAddress("MAIN.trigger");
+ trigger2.setCondition("value == true");
+ batch2.setTrigger(trigger2);
+
+ config.getBatches().add(batch2);
+
+ return config;
+ }
+
+ private void assertConfigurationEquals(EventPumpConfiguration expected, EventPumpConfiguration actual) {
+ // Check connections
+ assertEquals(expected.getConnections().size(), actual.getConnections().size());
+
+ ConnectionConfiguration expectedConn = expected.getConnections().get(0);
+ ConnectionConfiguration actualConn = actual.getConnections().get(0);
+
+ assertEquals(expectedConn.getId(), actualConn.getId());
+ assertEquals(expectedConn.getUrl(), actualConn.getUrl());
+ assertEquals(expectedConn.getUsername(), actualConn.getUsername());
+ assertEquals(expectedConn.getPassword(), actualConn.getPassword());
+
+ // Check batches
+ assertEquals(expected.getBatches().size(), actual.getBatches().size());
+
+ // Check batch 1 (timer trigger)
+ BatchConfiguration expectedBatch1 = expected.getBatches().get(0);
+ BatchConfiguration actualBatch1 = actual.getBatches().get(0);
+
+ assertEquals(expectedBatch1.getId(), actualBatch1.getId());
+ assertEquals(expectedBatch1.getConnectionId(), actualBatch1.getConnectionId());
+ assertEquals(expectedBatch1.getSimpleTagMap(), actualBatch1.getSimpleTagMap());
+
+ TriggerConfiguration expectedTrigger1 = expectedBatch1.getTrigger();
+ TriggerConfiguration actualTrigger1 = actualBatch1.getTrigger();
+
+ assertEquals(expectedTrigger1.getType(), actualTrigger1.getType());
+ assertEquals(expectedTrigger1.getIntervalSeconds(), actualTrigger1.getIntervalSeconds());
+ assertEquals(expectedTrigger1.getInitialDelaySeconds(), actualTrigger1.getInitialDelaySeconds());
+
+ // Check batch 2 (subscription trigger)
+ BatchConfiguration expectedBatch2 = expected.getBatches().get(1);
+ BatchConfiguration actualBatch2 = actual.getBatches().get(1);
+
+ assertEquals(expectedBatch2.getId(), actualBatch2.getId());
+ assertEquals(expectedBatch2.getConnectionId(), actualBatch2.getConnectionId());
+ assertEquals(expectedBatch2.getSimpleTagMap(), actualBatch2.getSimpleTagMap());
+
+ TriggerConfiguration expectedTrigger2 = expectedBatch2.getTrigger();
+ TriggerConfiguration actualTrigger2 = actualBatch2.getTrigger();
+
+ assertEquals(expectedTrigger2.getType(), actualTrigger2.getType());
+ assertEquals(expectedTrigger2.getTagName(), actualTrigger2.getTagName());
+ assertEquals(expectedTrigger2.getTagAddress(), actualTrigger2.getTagAddress());
+ assertEquals(expectedTrigger2.getCondition(), actualTrigger2.getCondition());
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactoryTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactoryTest.java
new file mode 100644
index 0000000000..d4d2f64d74
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/EventPumpFactoryTest.java
@@ -0,0 +1,452 @@
+/*
+ * 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.tools.eventpump.config;
+
+import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.PlcConnectionManager;
+import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
+import org.apache.plc4x.java.api.messages.PlcBrowseRequest;
+import org.apache.plc4x.java.api.messages.PlcPingResponse;
+import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
+import org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest;
+import org.apache.plc4x.java.api.messages.PlcWriteRequest;
+import org.apache.plc4x.java.api.metadata.PlcConnectionMetadata;
+import org.apache.plc4x.java.api.model.PlcTag;
+import org.apache.plc4x.java.api.value.PlcValue;
+import org.apache.plc4x.java.tools.eventpump.EventPump;
+import org.apache.plc4x.java.tools.eventpump.TagBatch;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Tests for EventPumpFactory.
+ */
+class EventPumpFactoryTest {
+
+ @TempDir
+ File tempDir;
+
+ private PlcConnectionManager connectionManager;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ // Create a stub connection manager
+ connectionManager = new PlcConnectionManager() {
+ @Override
+ public PlcConnection getConnection(String connectionString) throws PlcConnectionException {
+ return new StubPlcConnection();
+ }
+
+ @Override
+ public PlcConnection getConnection(String connectionString, org.apache.plc4x.java.api.authentication.PlcAuthentication authentication) throws PlcConnectionException {
+ return new StubPlcConnection();
+ }
+ };
+ }
+
+ /**
+ * Stub implementation of PlcConnection for testing.
+ */
+ private static class StubPlcConnection implements PlcConnection {
+ @Override
+ public void connect() throws PlcConnectionException {
+ }
+
+ @Override
+ public void close() throws Exception {
+ }
+
+ @Override
+ public boolean isConnected() {
+ return true;
+ }
+
+ @Override
+ public Optional<PlcTag> parseTagAddress(String tagAddress) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<PlcValue> parseTagValue(PlcTag tag, Object... values) {
+ return Optional.empty();
+ }
+
+ @Override
+ public CompletableFuture<? extends PlcPingResponse> ping() {
+ return null;
+ }
+
+ @Override
+ public PlcReadRequest.Builder readRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public PlcWriteRequest.Builder writeRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public PlcSubscriptionRequest.Builder subscriptionRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public PlcUnsubscriptionRequest.Builder unsubscriptionRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public PlcBrowseRequest.Builder browseRequestBuilder() {
+ return null;
+ }
+
+ @Override
+ public PlcConnectionMetadata getMetadata() {
+ return null;
+ }
+ }
+
+ @Test
+ void testCreateFromConfiguration() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = createTestConfiguration();
+ TagBatch.TagBatchListener listener = (batch, response) -> {};
+
+ // Act
+ EventPump pump = EventPumpFactory.create(config, connectionManager, listener);
+
+ // Assert
+ assertNotNull(pump);
+ assertEquals(2, pump.getAllBatches().size());
+ assertTrue(pump.getAllBatches().containsKey("batch1"));
+ assertTrue(pump.getAllBatches().containsKey("batch2"));
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testCreateWithoutListener() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = createTestConfiguration();
+
+ // Act
+ EventPump pump = EventPumpFactory.create(config, connectionManager, null);
+
+ // Assert
+ assertNotNull(pump);
+ assertEquals(2, pump.getAllBatches().size());
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testCreateWithMissingConnection() {
+ // Arrange
+ EventPumpConfiguration config = new EventPumpConfiguration();
+
+ // Add batch without adding the connection it references
+ BatchConfiguration batch = new BatchConfiguration();
+ batch.setId("batch1");
+ batch.setConnectionId("nonexistent");
+
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "address1");
+ batch.setSimpleTags(tags);
+
+ TriggerConfiguration trigger = new TriggerConfiguration();
+ trigger.setType("timer");
+ trigger.setIntervalSeconds(1L);
+ batch.setTrigger(trigger);
+
+ config.getBatches().add(batch);
+
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> EventPumpFactory.create(config, connectionManager, null)
+ );
+ assertTrue(exception.getMessage().contains("Connection 'nonexistent' not found"));
+ }
+
+ @Test
+ void testCreateWithUnknownTriggerType() {
+ // Arrange
+ EventPumpConfiguration config = new EventPumpConfiguration();
+
+ ConnectionConfiguration conn = new ConnectionConfiguration();
+ conn.setId("conn1");
+ conn.setUrl("test://localhost");
+ config.getConnections().add(conn);
+
+ BatchConfiguration batch = new BatchConfiguration();
+ batch.setId("batch1");
+ batch.setConnectionId("conn1");
+
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "address1");
+ batch.setSimpleTags(tags);
+
+ TriggerConfiguration trigger = new TriggerConfiguration();
+ trigger.setType("unknown");
+ batch.setTrigger(trigger);
+
+ config.getBatches().add(batch);
+
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> EventPumpFactory.create(config, connectionManager, null)
+ );
+ assertTrue(exception.getMessage().contains("Unknown trigger type"));
+ }
+
+ @Test
+ void testCreateSubscriptionTriggerWithMissingTagName() {
+ // Arrange
+ EventPumpConfiguration config = new EventPumpConfiguration();
+
+ ConnectionConfiguration conn = new ConnectionConfiguration();
+ conn.setId("conn1");
+ conn.setUrl("test://localhost");
+ config.getConnections().add(conn);
+
+ BatchConfiguration batch = new BatchConfiguration();
+ batch.setId("batch1");
+ batch.setConnectionId("conn1");
+
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "address1");
+ batch.setSimpleTags(tags);
+
+ TriggerConfiguration trigger = new TriggerConfiguration();
+ trigger.setType("subscription");
+ trigger.setTagAddress("MAIN.trigger");
+ // Missing tagName
+ batch.setTrigger(trigger);
+
+ config.getBatches().add(batch);
+
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> EventPumpFactory.create(config, connectionManager, null)
+ );
+ assertTrue(exception.getMessage().contains("Subscription trigger requires tagName and tagAddress"));
+ }
+
+ @Test
+ void testFromYaml() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = createTestConfiguration();
+ File yamlFile = new File(tempDir, "config.yaml");
+ config.toYaml(yamlFile);
+
+ TagBatch.TagBatchListener listener = (batch, response) -> {};
+
+ // Act
+ EventPump pump = EventPumpFactory.fromYaml(yamlFile, connectionManager, listener);
+
+ // Assert
+ assertNotNull(pump);
+ assertEquals(2, pump.getAllBatches().size());
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testFromJson() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = createTestConfiguration();
+ File jsonFile = new File(tempDir, "config.json");
+ config.toJson(jsonFile);
+
+ TagBatch.TagBatchListener listener = (batch, response) -> {};
+
+ // Act
+ EventPump pump = EventPumpFactory.fromJson(jsonFile, connectionManager, listener);
+
+ // Assert
+ assertNotNull(pump);
+ assertEquals(2, pump.getAllBatches().size());
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testFromXml() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = createTestConfiguration();
+ File xmlFile = new File(tempDir, "config.xml");
+ config.toXml(xmlFile);
+
+ TagBatch.TagBatchListener listener = (batch, response) -> {};
+
+ // Act
+ EventPump pump = EventPumpFactory.fromXml(xmlFile, connectionManager, listener);
+
+ // Assert
+ assertNotNull(pump);
+ assertEquals(2, pump.getAllBatches().size());
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testCreateTimerTriggerWithInitialDelay() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = new EventPumpConfiguration();
+
+ ConnectionConfiguration conn = new ConnectionConfiguration();
+ conn.setId("conn1");
+ conn.setUrl("test://localhost");
+ config.getConnections().add(conn);
+
+ BatchConfiguration batch = new BatchConfiguration();
+ batch.setId("batch1");
+ batch.setConnectionId("conn1");
+
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "address1");
+ batch.setSimpleTags(tags);
+
+ TriggerConfiguration trigger = new TriggerConfiguration();
+ trigger.setType("timer");
+ trigger.setIntervalSeconds(5L);
+ trigger.setInitialDelaySeconds(2L);
+ batch.setTrigger(trigger);
+
+ config.getBatches().add(batch);
+
+ // Act
+ EventPump pump = EventPumpFactory.create(config, connectionManager, null);
+
+ // Assert
+ assertNotNull(pump);
+ TagBatch createdBatch = pump.getAllBatches().get("batch1");
+ assertNotNull(createdBatch);
+
+ // Cleanup
+ pump.close();
+ }
+
+ @Test
+ void testCreateTimerTriggerWithoutInitialDelay() throws Exception {
+ // Arrange
+ EventPumpConfiguration config = new EventPumpConfiguration();
+
+ ConnectionConfiguration conn = new ConnectionConfiguration();
+ conn.setId("conn1");
+ conn.setUrl("test://localhost");
+ config.getConnections().add(conn);
+
+ BatchConfiguration batch = new BatchConfiguration();
+ batch.setId("batch1");
+ batch.setConnectionId("conn1");
+
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag1", "address1");
+ batch.setSimpleTags(tags);
+
+ TriggerConfiguration trigger = new TriggerConfiguration();
+ trigger.setType("timer");
+ trigger.setIntervalSeconds(5L);
+ // No initial delay - should default to 0
+ batch.setTrigger(trigger);
+
+ config.getBatches().add(batch);
+
+ // Act
+ EventPump pump = EventPumpFactory.create(config, connectionManager, null);
+
+ // Assert
+ assertNotNull(pump);
+ TagBatch createdBatch = pump.getAllBatches().get("batch1");
+ assertNotNull(createdBatch);
+
+ // Cleanup
+ pump.close();
+ }
+
+ private EventPumpConfiguration createTestConfiguration() {
+ EventPumpConfiguration config = new EventPumpConfiguration();
+
+ // Add connection
+ ConnectionConfiguration conn = new ConnectionConfiguration();
+ conn.setId("plc1");
+ conn.setUrl("ads:tcp://192.168.1.1");
+ conn.setUsername("admin");
+ conn.setPassword("secret");
+ config.getConnections().add(conn);
+
+ // Add batch with timer trigger
+ BatchConfiguration batch1 = new BatchConfiguration();
+ batch1.setId("batch1");
+ batch1.setConnectionId("plc1");
+
+ Map<String, String> tags1 = new HashMap<>();
+ tags1.put("temperature", "MAIN.temperature");
+ tags1.put("pressure", "MAIN.pressure");
+ batch1.setSimpleTags(tags1);
+
+ TriggerConfiguration trigger1 = new TriggerConfiguration();
+ trigger1.setType("timer");
+ trigger1.setIntervalSeconds(5L);
+ trigger1.setInitialDelaySeconds(2L);
+ batch1.setTrigger(trigger1);
+
+ config.getBatches().add(batch1);
+
+ // Add batch with subscription trigger
+ BatchConfiguration batch2 = new BatchConfiguration();
+ batch2.setId("batch2");
+ batch2.setConnectionId("plc1");
+
+ Map<String, String> tags2 = new HashMap<>();
+ tags2.put("status", "MAIN.status");
+ batch2.setSimpleTags(tags2);
+
+ TriggerConfiguration trigger2 = new TriggerConfiguration();
+ trigger2.setType("subscription");
+ trigger2.setTagName("triggerTag");
+ trigger2.setTagAddress("MAIN.trigger");
+ trigger2.setCondition("value == true");
+ batch2.setTrigger(trigger2);
+
+ config.getBatches().add(batch2);
+
+ return config;
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/TagConfigurationTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/TagConfigurationTest.java
new file mode 100644
index 0000000000..33b4c471a4
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/TagConfigurationTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.tools.eventpump.config;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class TagConfigurationTest {
+
+ @Test
+ void testDefaultConstructor() {
+ TagConfiguration config = new TagConfiguration();
+
+ assertNull(config.getAddress());
+ assertNull(config.getTransform());
+ assertFalse(config.hasTransform());
+ }
+
+ @Test
+ void testConstructorWithAddress() {
+ TagConfiguration config = new TagConfiguration("MAIN.temperature");
+
+ assertEquals("MAIN.temperature", config.getAddress());
+ assertNull(config.getTransform());
+ assertFalse(config.hasTransform());
+ }
+
+ @Test
+ void testConstructorWithAddressAndTransform() {
+ TagConfiguration config = new TagConfiguration("MAIN.temperature", "value * 1.8 + 32");
+
+ assertEquals("MAIN.temperature", config.getAddress());
+ assertEquals("value * 1.8 + 32", config.getTransform());
+ assertTrue(config.hasTransform());
+ }
+
+ @Test
+ void testConstructorWithAddressAndNullTransform() {
+ TagConfiguration config = new TagConfiguration("MAIN.temperature", null);
+
+ assertEquals("MAIN.temperature", config.getAddress());
+ assertNull(config.getTransform());
+ assertFalse(config.hasTransform());
+ }
+
+ @Test
+ void testSetAndGetAddress() {
+ TagConfiguration config = new TagConfiguration();
+
+ config.setAddress("MAIN.pressure");
+
+ assertEquals("MAIN.pressure", config.getAddress());
+ }
+
+ @Test
+ void testSetAndGetTransform() {
+ TagConfiguration config = new TagConfiguration();
+
+ config.setTransform("value + 100");
+
+ assertEquals("value + 100", config.getTransform());
+ assertTrue(config.hasTransform());
+ }
+
+ @Test
+ void testHasTransformWithEmptyString() {
+ TagConfiguration config = new TagConfiguration("MAIN.temp", "");
+
+ assertFalse(config.hasTransform());
+ }
+
+ @Test
+ void testHasTransformWithWhitespace() {
+ TagConfiguration config = new TagConfiguration("MAIN.temp", " ");
+
+ assertFalse(config.hasTransform());
+ }
+
+ @Test
+ void testToStringWithoutTransform() {
+ TagConfiguration config = new TagConfiguration("MAIN.temperature");
+
+ String result = config.toString();
+
+ assertEquals("TagConfiguration{address='MAIN.temperature'}", result);
+ }
+
+ @Test
+ void testToStringWithTransform() {
+ TagConfiguration config = new TagConfiguration("MAIN.temperature", "value * 1.8 + 32");
+
+ String result = config.toString();
+
+ assertEquals("TagConfiguration{address='MAIN.temperature', transform='value * 1.8 + 32'}", result);
+ }
+
+ @Test
+ void testEquals_Same() {
+ TagConfiguration config = new TagConfiguration("MAIN.temp", "value * 2");
+
+ assertTrue(config.equals(config));
+ }
+
+ @Test
+ void testEquals_Null() {
+ TagConfiguration config = new TagConfiguration("MAIN.temp");
+
+ assertFalse(config.equals(null));
+ }
+
+ @Test
+ void testEquals_DifferentClass() {
+ TagConfiguration config = new TagConfiguration("MAIN.temp");
+
+ assertFalse(config.equals("not a TagConfiguration"));
+ }
+
+ @Test
+ void testEquals_SameAddressNoTransform() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp");
+ TagConfiguration config2 = new TagConfiguration("MAIN.temp");
+
+ assertTrue(config1.equals(config2));
+ assertTrue(config2.equals(config1));
+ }
+
+ @Test
+ void testEquals_SameAddressAndTransform() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp", "value * 2");
+ TagConfiguration config2 = new TagConfiguration("MAIN.temp", "value * 2");
+
+ assertTrue(config1.equals(config2));
+ assertTrue(config2.equals(config1));
+ }
+
+ @Test
+ void testEquals_DifferentAddress() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp1");
+ TagConfiguration config2 = new TagConfiguration("MAIN.temp2");
+
+ assertFalse(config1.equals(config2));
+ assertFalse(config2.equals(config1));
+ }
+
+ @Test
+ void testEquals_DifferentTransform() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp", "value * 2");
+ TagConfiguration config2 = new TagConfiguration("MAIN.temp", "value * 3");
+
+ assertFalse(config1.equals(config2));
+ assertFalse(config2.equals(config1));
+ }
+
+ @Test
+ void testEquals_OneHasTransformOtherDoesNot() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp", "value * 2");
+ TagConfiguration config2 = new TagConfiguration("MAIN.temp");
+
+ assertFalse(config1.equals(config2));
+ assertFalse(config2.equals(config1));
+ }
+
+ @Test
+ void testEquals_BothAddressesNull() {
+ TagConfiguration config1 = new TagConfiguration();
+ TagConfiguration config2 = new TagConfiguration();
+
+ assertTrue(config1.equals(config2));
+ }
+
+ @Test
+ void testEquals_OneAddressNull() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp");
+ TagConfiguration config2 = new TagConfiguration();
+
+ assertFalse(config1.equals(config2));
+ }
+
+ @Test
+ void testHashCode_Equal() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp", "value * 2");
+ TagConfiguration config2 = new TagConfiguration("MAIN.temp", "value * 2");
+
+ assertEquals(config1.hashCode(), config2.hashCode());
+ }
+
+ @Test
+ void testHashCode_DifferentAddress() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp1");
+ TagConfiguration config2 = new TagConfiguration("MAIN.temp2");
+
+ assertNotEquals(config1.hashCode(), config2.hashCode());
+ }
+
+ @Test
+ void testHashCode_DifferentTransform() {
+ TagConfiguration config1 = new TagConfiguration("MAIN.temp", "value * 2");
+ TagConfiguration config2 = new TagConfiguration("MAIN.temp", "value * 3");
+
+ assertNotEquals(config1.hashCode(), config2.hashCode());
+ }
+
+ @Test
+ void testHashCode_NullFields() {
+ TagConfiguration config = new TagConfiguration();
+
+ // Should not throw exception
+ int hashCode = config.hashCode();
+
+ assertTrue(hashCode == 0); // 31 * 0 + 0 = 0
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/TagMapDeserializerTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/TagMapDeserializerTest.java
new file mode 100644
index 0000000000..8d12ce287f
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/config/TagMapDeserializerTest.java
@@ -0,0 +1,216 @@
+/*
+ * 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.tools.eventpump.config;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class TagMapDeserializerTest {
+
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ @Test
+ void testDeserializeSimpleFormat() throws Exception {
+ String json = """
+ {
+ "temperature": "MAIN.temperature",
+ "pressure": "MAIN.pressure"
+ }
+ """;
+
+ JsonParser parser = mapper.getFactory().createParser(json);
+ parser.nextToken(); // Move to start of object
+
+ TagMapDeserializer deserializer = new TagMapDeserializer();
+ Map<String, TagConfiguration> result = deserializer.deserialize(parser, mapper.getDeserializationContext());
+
+ assertNotNull(result);
+ assertEquals(2, result.size());
+
+ TagConfiguration tempConfig = result.get("temperature");
+ assertNotNull(tempConfig);
+ assertEquals("MAIN.temperature", tempConfig.getAddress());
+ assertFalse(tempConfig.hasTransform());
+
+ TagConfiguration pressConfig = result.get("pressure");
+ assertNotNull(pressConfig);
+ assertEquals("MAIN.pressure", pressConfig.getAddress());
+ assertFalse(pressConfig.hasTransform());
+ }
+
+ @Test
+ void testDeserializeExtendedFormat() throws Exception {
+ String json = """
+ {
+ "temperature": {
+ "address": "MAIN.temperature",
+ "transform": "value * 1.8 + 32"
+ }
+ }
+ """;
+
+ JsonParser parser = mapper.getFactory().createParser(json);
+ parser.nextToken(); // Move to start of object
+
+ TagMapDeserializer deserializer = new TagMapDeserializer();
+ Map<String, TagConfiguration> result = deserializer.deserialize(parser, mapper.getDeserializationContext());
+
+ assertNotNull(result);
+ assertEquals(1, result.size());
+
+ TagConfiguration tempConfig = result.get("temperature");
+ assertNotNull(tempConfig);
+ assertEquals("MAIN.temperature", tempConfig.getAddress());
+ assertEquals("value * 1.8 + 32", tempConfig.getTransform());
+ assertTrue(tempConfig.hasTransform());
+ }
+
+ @Test
+ void testDeserializeMixedFormat() throws Exception {
+ String json = """
+ {
+ "temperature": {
+ "address": "MAIN.temperature",
+ "transform": "value * 1.8 + 32"
+ },
+ "pressure": "MAIN.pressure",
+ "humidity": {
+ "address": "MAIN.humidity"
+ }
+ }
+ """;
+
+ JsonParser parser = mapper.getFactory().createParser(json);
+ parser.nextToken(); // Move to start of object
+
+ TagMapDeserializer deserializer = new TagMapDeserializer();
+ Map<String, TagConfiguration> result = deserializer.deserialize(parser, mapper.getDeserializationContext());
+
+ assertNotNull(result);
+ assertEquals(3, result.size());
+
+ // Temperature with transform
+ TagConfiguration tempConfig = result.get("temperature");
+ assertNotNull(tempConfig);
+ assertEquals("MAIN.temperature", tempConfig.getAddress());
+ assertEquals("value * 1.8 + 32", tempConfig.getTransform());
+ assertTrue(tempConfig.hasTransform());
+
+ // Pressure simple format
+ TagConfiguration pressConfig = result.get("pressure");
+ assertNotNull(pressConfig);
+ assertEquals("MAIN.pressure", pressConfig.getAddress());
+ assertFalse(pressConfig.hasTransform());
+
+ // Humidity extended format without transform
+ TagConfiguration humidConfig = result.get("humidity");
+ assertNotNull(humidConfig);
+ assertEquals("MAIN.humidity", humidConfig.getAddress());
+ assertFalse(humidConfig.hasTransform());
+ }
+
+ @Test
+ void testDeserializeInvalidFormat() throws Exception {
+ String json = """
+ {
+ "temperature": 123
+ }
+ """;
+
+ JsonParser parser = mapper.getFactory().createParser(json);
+ parser.nextToken(); // Move to start of object
+
+ TagMapDeserializer deserializer = new TagMapDeserializer();
+
+ assertThrows(Exception.class, () -> {
+ deserializer.deserialize(parser, mapper.getDeserializationContext());
+ });
+ }
+
+ @Test
+ void testDeserializeEmptyMap() throws Exception {
+ String json = "{}";
+
+ JsonParser parser = mapper.getFactory().createParser(json);
+ parser.nextToken(); // Move to start of object
+
+ TagMapDeserializer deserializer = new TagMapDeserializer();
+ Map<String, TagConfiguration> result = deserializer.deserialize(parser, mapper.getDeserializationContext());
+
+ assertNotNull(result);
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ void testDeserializeExtendedFormatWithoutTransform() throws Exception {
+ String json = """
+ {
+ "temperature": {
+ "address": "MAIN.temperature"
+ }
+ }
+ """;
+
+ JsonParser parser = mapper.getFactory().createParser(json);
+ parser.nextToken(); // Move to start of object
+
+ TagMapDeserializer deserializer = new TagMapDeserializer();
+ Map<String, TagConfiguration> result = deserializer.deserialize(parser, mapper.getDeserializationContext());
+
+ assertNotNull(result);
+ assertEquals(1, result.size());
+
+ TagConfiguration tempConfig = result.get("temperature");
+ assertNotNull(tempConfig);
+ assertEquals("MAIN.temperature", tempConfig.getAddress());
+ assertFalse(tempConfig.hasTransform());
+ }
+
+ @Test
+ void testDeserializePreservesOrder() throws Exception {
+ String json = """
+ {
+ "tag1": "MAIN.tag1",
+ "tag2": "MAIN.tag2",
+ "tag3": "MAIN.tag3"
+ }
+ """;
+
+ JsonParser parser = mapper.getFactory().createParser(json);
+ parser.nextToken(); // Move to start of object
+
+ TagMapDeserializer deserializer = new TagMapDeserializer();
+ Map<String, TagConfiguration> result = deserializer.deserialize(parser, mapper.getDeserializationContext());
+
+ assertNotNull(result);
+ assertEquals(3, result.size());
+
+ // LinkedHashMap should preserve insertion order
+ var keys = result.keySet().toArray(new String[0]);
+ assertEquals("tag1", keys[0]);
+ assertEquals("tag2", keys[1]);
+ assertEquals("tag3", keys[2]);
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/SimpleExpressionEvaluatorTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/SimpleExpressionEvaluatorTest.java
new file mode 100644
index 0000000000..2066ad635a
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/SimpleExpressionEvaluatorTest.java
@@ -0,0 +1,248 @@
+/*
+ * 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.tools.eventpump.transform;
+
+import org.apache.plc4x.java.api.value.PlcValue;
+import org.apache.plc4x.java.spi.values.PlcDINT;
+import org.apache.plc4x.java.spi.values.PlcREAL;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class SimpleExpressionEvaluatorTest {
+
+ private SimpleExpressionEvaluator evaluator;
+
+ @BeforeEach
+ void setUp() {
+ evaluator = new SimpleExpressionEvaluator();
+ }
+
+ @Test
+ void testGetName() {
+ assertEquals("simple", evaluator.getName());
+ }
+
+ @Test
+ void testSimpleVariable() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(42));
+
+ PlcValue result = evaluator.transform("value", context);
+ assertEquals(42, result.getInteger());
+ }
+
+ @Test
+ void testArithmeticAddition() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result = evaluator.transform("value + 5", context);
+ assertEquals(15L, result.getLong());
+ }
+
+ @Test
+ void testArithmeticSubtraction() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result = evaluator.transform("value - 3", context);
+ assertEquals(7L, result.getLong());
+ }
+
+ @Test
+ void testArithmeticMultiplication() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result = evaluator.transform("value * 2", context);
+ assertEquals(20L, result.getLong());
+ }
+
+ @Test
+ void testArithmeticDivision() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result = evaluator.transform("value / 2", context);
+ assertEquals(5.0, result.getDouble(), 0.001);
+ }
+
+ @Test
+ void testArithmeticModulo() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result = evaluator.transform("value % 3", context);
+ assertEquals(1L, result.getLong());
+ }
+
+ @Test
+ void testFloatingPointArithmetic() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcREAL(100.0));
+
+ // Celsius to Fahrenheit: (C * 1.8) + 32
+ PlcValue result = evaluator.transform("value * 1.8 + 32", context);
+ assertEquals(212.0, result.getDouble(), 0.001);
+ }
+
+ @Test
+ void testParentheses() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result = evaluator.transform("(value + 5) * 2", context);
+ assertEquals(30L, result.getLong());
+ }
+
+ @Test
+ void testComparison() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result1 = evaluator.transform("value > 5", context);
+ assertTrue(result1.getBoolean());
+
+ PlcValue result2 = evaluator.transform("value < 5", context);
+ assertFalse(result2.getBoolean());
+
+ PlcValue result3 = evaluator.transform("value == 10", context);
+ assertTrue(result3.getBoolean());
+
+ PlcValue result4 = evaluator.transform("value != 10", context);
+ assertFalse(result4.getBoolean());
+ }
+
+ @Test
+ void testLogicalOperators() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result1 = evaluator.transform("value > 5 && value < 15", context);
+ assertTrue(result1.getBoolean());
+
+ PlcValue result2 = evaluator.transform("value > 15 || value < 5", context);
+ assertFalse(result2.getBoolean());
+
+ PlcValue result3 = evaluator.transform("!(value > 15)", context);
+ assertTrue(result3.getBoolean());
+ }
+
+ @Test
+ void testUnaryNegation() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result = evaluator.transform("-value", context);
+ assertEquals(-10L, result.getLong());
+ }
+
+ @Test
+ void testBooleanLiterals() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+
+ PlcValue result1 = evaluator.transform("true", context);
+ assertTrue(result1.getBoolean());
+
+ PlcValue result2 = evaluator.transform("false", context);
+ assertFalse(result2.getBoolean());
+ }
+
+ @Test
+ void testMultipleVariables() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("temperature", new PlcDINT(25));
+ context.put("humidity", new PlcDINT(60));
+
+ PlcValue result = evaluator.transform("temperature + humidity", context);
+ assertEquals(85L, result.getLong());
+ }
+
+ @Test
+ void testComplexExpression() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("pressure", new PlcREAL(1050.0));
+
+ // Pressure deviation from standard: (P - 1013.25) / 1013.25 * 100
+ PlcValue result = evaluator.transform("(pressure - 1013.25) / 1013.25 * 100", context);
+ assertEquals(3.626, result.getDouble(), 0.001);
+ }
+
+ @Test
+ void testNullExpression() {
+ Map<String, PlcValue> context = new HashMap<>();
+
+ assertThrows(TransformException.class, () -> {
+ evaluator.transform(null, context);
+ });
+ }
+
+ @Test
+ void testEmptyExpression() {
+ Map<String, PlcValue> context = new HashMap<>();
+
+ assertThrows(TransformException.class, () -> {
+ evaluator.transform("", context);
+ });
+ }
+
+ @Test
+ void testUnknownVariable() {
+ Map<String, PlcValue> context = new HashMap<>();
+
+ assertThrows(TransformException.class, () -> {
+ evaluator.transform("unknownVar + 5", context);
+ });
+ }
+
+ @Test
+ void testDivisionByZero() {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ assertThrows(TransformException.class, () -> {
+ evaluator.transform("value / 0", context);
+ });
+ }
+
+ @Test
+ void testModuloByZero() {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ assertThrows(TransformException.class, () -> {
+ evaluator.transform("value % 0", context);
+ });
+ }
+
+ @Test
+ void testWhitespace() throws TransformException {
+ Map<String, PlcValue> context = new HashMap<>();
+ context.put("value", new PlcDINT(10));
+
+ PlcValue result = evaluator.transform(" value + 5 ", context);
+ assertEquals(15L, result.getLong());
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerRegistryTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerRegistryTest.java
new file mode 100644
index 0000000000..4db92e1f64
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerRegistryTest.java
@@ -0,0 +1,221 @@
+/*
+ * 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.tools.eventpump.transform;
+
+import org.apache.plc4x.java.api.value.PlcValue;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class ValueTransformerRegistryTest {
+
+ private ValueTransformerRegistry registry;
+
+ @BeforeEach
+ void setUp() {
+ // Create a fresh registry instance for each test to ensure isolation
+ registry = ValueTransformerRegistry.createDefault();
+ }
+
+ @Test
+ void testCreateDefaultRegistersSimpleTransformer() {
+ ValueTransformer defaultTransformer = registry.getDefault();
+
+ assertNotNull(defaultTransformer);
+ assertEquals("simple", defaultTransformer.getName());
+ }
+
+ @Test
+ void testEmptyRegistryCanBeCreated() {
+ ValueTransformerRegistry emptyRegistry = new ValueTransformerRegistry();
+
+ // Empty registry should have no transformers
+ assertTrue(emptyRegistry.getRegisteredNames().isEmpty());
+ }
+
+ @Test
+ void testCanRegisterToEmptyRegistry() {
+ ValueTransformerRegistry emptyRegistry = new ValueTransformerRegistry();
+ MockValueTransformer mockTransformer = new MockValueTransformer("test");
+
+ emptyRegistry.register(mockTransformer);
+
+ assertEquals(1, emptyRegistry.getRegisteredNames().size());
+ assertSame(mockTransformer, emptyRegistry.get("test"));
+ }
+
+ @Test
+ void testRegisterNewTransformer() {
+ MockValueTransformer mockTransformer = new MockValueTransformer("test");
+
+ registry.register(mockTransformer);
+
+ ValueTransformer retrieved = registry.get("test");
+ assertNotNull(retrieved);
+ assertEquals("test", retrieved.getName());
+ assertSame(mockTransformer, retrieved);
+ }
+
+ @Test
+ void testRegisterNullTransformerThrowsException() {
+ assertThrows(IllegalArgumentException.class, () -> {
+ registry.register(null);
+ });
+ }
+
+ @Test
+ void testRegisterTransformerWithNullNameThrowsException() {
+ MockValueTransformer mockTransformer = new MockValueTransformer(null);
+
+ assertThrows(IllegalArgumentException.class, () -> {
+ registry.register(mockTransformer);
+ });
+ }
+
+ @Test
+ void testRegisterTransformerWithEmptyNameThrowsException() {
+ MockValueTransformer mockTransformer = new MockValueTransformer("");
+
+ assertThrows(IllegalArgumentException.class, () -> {
+ registry.register(mockTransformer);
+ });
+ }
+
+ @Test
+ void testRegisterDuplicateTransformerKeepsExisting() {
+ MockValueTransformer first = new MockValueTransformer("duplicate");
+ MockValueTransformer second = new MockValueTransformer("duplicate");
+
+ registry.register(first);
+ registry.register(second);
+
+ ValueTransformer retrieved = registry.get("duplicate");
+ assertSame(first, retrieved, "Should keep the first registered transformer");
+ }
+
+ @Test
+ void testGetNonExistentTransformerReturnsNull() {
+ ValueTransformer result = registry.get("nonexistent");
+
+ assertNull(result);
+ }
+
+ @Test
+ void testSetDefault() {
+ MockValueTransformer mockTransformer = new MockValueTransformer("custom");
+ registry.register(mockTransformer);
+
+ registry.setDefault("custom");
+
+ ValueTransformer defaultTransformer = registry.getDefault();
+ assertEquals("custom", defaultTransformer.getName());
+ assertSame(mockTransformer, defaultTransformer);
+ }
+
+ @Test
+ void testSetDefaultToNonExistentTransformerThrowsException() {
+ assertThrows(IllegalArgumentException.class, () -> {
+ registry.setDefault("nonexistent");
+ });
+ }
+
+ @Test
+ void testGetDefaultWhenDefaultNotRegisteredThrowsException() {
+ // Create an empty registry and try to get default without registering anything
+ ValueTransformerRegistry emptyRegistry = new ValueTransformerRegistry();
+
+ assertThrows(IllegalStateException.class, () -> {
+ emptyRegistry.getDefault();
+ });
+ }
+
+ @Test
+ void testUnregister() {
+ MockValueTransformer mockTransformer = new MockValueTransformer("toremove");
+ registry.register(mockTransformer);
+
+ boolean removed = registry.unregister("toremove");
+
+ assertTrue(removed);
+ assertNull(registry.get("toremove"));
+ }
+
+ @Test
+ void testUnregisterNonExistentReturnsFalse() {
+ boolean removed = registry.unregister("nonexistent");
+
+ assertFalse(removed);
+ }
+
+ @Test
+ void testUnregisterDefaultTransformerThrowsException() {
+ assertThrows(IllegalStateException.class, () -> {
+ registry.unregister("simple");
+ });
+ }
+
+ @Test
+ void testGetRegisteredNames() {
+ MockValueTransformer mock1 = new MockValueTransformer("test1");
+ MockValueTransformer mock2 = new MockValueTransformer("test2");
+
+ registry.register(mock1);
+ registry.register(mock2);
+
+ java.util.Set<String> names = registry.getRegisteredNames();
+
+ assertTrue(names.contains("simple"));
+ assertTrue(names.contains("test1"));
+ assertTrue(names.contains("test2"));
+ }
+
+ @Test
+ void testGetRegisteredNamesIsUnmodifiable() {
+ java.util.Set<String> names = registry.getRegisteredNames();
+
+ assertThrows(UnsupportedOperationException.class, () -> {
+ names.add("new");
+ });
+ }
+
+ /**
+ * Mock ValueTransformer for testing
+ */
+ private static class MockValueTransformer implements ValueTransformer {
+ private final String name;
+
+ public MockValueTransformer(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public PlcValue transform(String expression, Map<String, PlcValue> context) throws TransformException {
+ return null; // Not used in registry tests
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerTest.java
new file mode 100644
index 0000000000..50375783e7
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/transform/ValueTransformerTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.tools.eventpump.transform;
+
+import org.apache.plc4x.java.api.value.PlcValue;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ValueTransformerTest {
+
+ @Test
+ void testDefaultSupportsMethod() throws TransformException {
+ ValueTransformer transformer = new ValueTransformer() {
+ @Override
+ public PlcValue transform(String expression, Map<String, PlcValue> context) throws TransformException {
+ return null;
+ }
+
+ @Override
+ public String getName() {
+ return "test";
+ }
+ };
+
+ // Test the default supports() method
+ assertTrue(transformer.supports("any expression"));
+ assertTrue(transformer.supports(""));
+ assertTrue(transformer.supports(null));
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/SubscriptionTriggerTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/SubscriptionTriggerTest.java
new file mode 100644
index 0000000000..d97e370cb1
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/SubscriptionTriggerTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.tools.eventpump.triggers;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Tests for SubscriptionTrigger.
+ */
+class SubscriptionTriggerTest {
+
+ @Test
+ void testConstructor() {
+ // Act
+ SubscriptionTrigger trigger = new SubscriptionTrigger("testTag", "MAIN.test");
+
+ // Assert
+ assertEquals("testTag", trigger.getTagName());
+ assertEquals("MAIN.test", trigger.getTagAddress());
+ assertFalse(trigger.isRunning());
+ }
+
+ @Test
+ void testConstructorWithNullTagName() {
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new SubscriptionTrigger(null, "MAIN.test")
+ );
+ assertTrue(exception.getMessage().contains("Tag name cannot be null or empty"));
+ }
+
+ @Test
+ void testConstructorWithEmptyTagName() {
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new SubscriptionTrigger("", "MAIN.test")
+ );
+ assertTrue(exception.getMessage().contains("Tag name cannot be null or empty"));
+ }
+
+ @Test
+ void testConstructorWithWhitespaceTagName() {
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new SubscriptionTrigger(" ", "MAIN.test")
+ );
+ assertTrue(exception.getMessage().contains("Tag name cannot be null or empty"));
+ }
+
+ @Test
+ void testConstructorWithNullTagAddress() {
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new SubscriptionTrigger("testTag", null)
+ );
+ assertTrue(exception.getMessage().contains("Tag address cannot be null or empty"));
+ }
+
+ @Test
+ void testConstructorWithEmptyTagAddress() {
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new SubscriptionTrigger("testTag", "")
+ );
+ assertTrue(exception.getMessage().contains("Tag address cannot be null or empty"));
+ }
+
+ @Test
+ void testConstructorWithWhitespaceTagAddress() {
+ // Act & Assert
+ IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new SubscriptionTrigger("testTag", " ")
+ );
+ assertTrue(exception.getMessage().contains("Tag address cannot be null or empty"));
+ }
+
+ @Test
+ void testGetTagName() {
+ // Arrange
+ SubscriptionTrigger trigger = new SubscriptionTrigger("myTag", "MAIN.value");
+
+ // Act & Assert
+ assertEquals("myTag", trigger.getTagName());
+ }
+
+ @Test
+ void testGetTagAddress() {
+ // Arrange
+ SubscriptionTrigger trigger = new SubscriptionTrigger("myTag", "MAIN.value");
+
+ // Act & Assert
+ assertEquals("MAIN.value", trigger.getTagAddress());
+ }
+
+ @Test
+ void testStartThrowsUnsupportedOperationException() {
+ // Arrange
+ SubscriptionTrigger trigger = new SubscriptionTrigger("testTag", "MAIN.test");
+ Trigger.TriggerListener listener = (t) -> {};
+
+ // Act & Assert
+ UnsupportedOperationException exception = assertThrows(
+ UnsupportedOperationException.class,
+ () -> trigger.start(listener)
+ );
+ assertTrue(exception.getMessage().contains("SubscriptionTrigger is not yet fully implemented"));
+ assertTrue(exception.getMessage().contains("Use TimerTrigger"));
+ }
+
+ @Test
+ void testStop() {
+ // Arrange
+ SubscriptionTrigger trigger = new SubscriptionTrigger("testTag", "MAIN.test");
+
+ // Act - should not throw
+ assertDoesNotThrow(() -> trigger.stop());
+ }
+
+ @Test
+ void testIsRunning() {
+ // Arrange
+ SubscriptionTrigger trigger = new SubscriptionTrigger("testTag", "MAIN.test");
+
+ // Act & Assert - placeholder never runs
+ assertFalse(trigger.isRunning());
+ }
+
+ @Test
+ void testGetType() {
+ // Arrange
+ SubscriptionTrigger trigger = new SubscriptionTrigger("myTag", "MAIN.value");
+
+ // Act
+ String type = trigger.getType();
+
+ // Assert
+ assertNotNull(type);
+ assertTrue(type.contains("Subscription"));
+ assertTrue(type.contains("PLACEHOLDER"));
+ assertTrue(type.contains("myTag"));
+ assertTrue(type.contains("MAIN.value"));
+ }
+
+ @Test
+ void testClose() {
+ // Arrange
+ SubscriptionTrigger trigger = new SubscriptionTrigger("testTag", "MAIN.test");
+
+ // Act - should not throw
+ assertDoesNotThrow(() -> trigger.close());
+ }
+
+ @Test
+ void testMultipleCalls() {
+ // Arrange
+ SubscriptionTrigger trigger = new SubscriptionTrigger("testTag", "MAIN.test");
+
+ // Act & Assert - multiple calls should be safe
+ assertDoesNotThrow(() -> {
+ trigger.stop();
+ trigger.stop();
+ trigger.close();
+ trigger.close();
+ });
+
+ assertFalse(trigger.isRunning());
+ }
+}
diff --git a/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTriggerTest.java b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTriggerTest.java
new file mode 100644
index 0000000000..340ce61a93
--- /dev/null
+++ b/plc4j/tools/event-pump/src/test/java/org/apache/plc4x/java/tools/eventpump/triggers/TimerTriggerTest.java
@@ -0,0 +1,234 @@
+/*
+ * 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.tools.eventpump.triggers;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Tests for TimerTrigger.
+ */
+class TimerTriggerTest {
+
+ @Test
+ void testBasicFunctionality() throws Exception {
+ // Arrange
+ CountDownLatch latch = new CountDownLatch(3); // Wait for 3 firings
+ AtomicInteger fireCount = new AtomicInteger(0);
+
+ TimerTrigger trigger = new TimerTrigger(100, TimeUnit.MILLISECONDS);
+
+ // Act
+ trigger.start(t -> {
+ fireCount.incrementAndGet();
+ latch.countDown();
+ });
+
+ // Wait for at least 3 firings
+ assertTrue(latch.await(2, TimeUnit.SECONDS), "Trigger should fire at least 3 times");
+
+ // Assert
+ assertTrue(trigger.isRunning());
+ assertTrue(fireCount.get() >= 3, "Fire count should be at least 3, was: " + fireCount.get());
+
+ // Cleanup
+ trigger.close();
+ assertFalse(trigger.isRunning());
+ }
+
+ @Test
+ void testStopAndRestart() throws Exception {
+ // Arrange
+ CountDownLatch latch1 = new CountDownLatch(2);
+ CountDownLatch latch2 = new CountDownLatch(2);
... 181 lines suppressed ...