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 2024/10/06 13:50:16 UTC
(plc4x) 01/01: fix: Updated the subscription API to match the changes of the Read- and Write-API. Also replaced the single-item handlers with an all items handler.
This is an automated email from the ASF dual-hosted git repository.
cdutz pushed a commit to branch refactor/update-subscription-api
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 0b6ddd290f169fa2deceaa2d076f8efbbcaa2cee
Author: Christofer Dutz <cd...@apache.org>
AuthorDate: Sun Oct 6 07:50:02 2024 -0600
fix: Updated the subscription API to match the changes of the Read- and Write-API. Also replaced the single-item handlers with an all items handler.
---
.../java/api/messages/PlcSubscriptionRequest.java | 21 +++---
.../api/messages/PlcSubscriptionTagRequest.java | 6 +-
.../plc4x/java/ads/protocol/AdsProtocolLogic.java | 77 ++++++++++++----------
.../java/firmata/readwrite/FirmataManualTest.java | 2 +-
.../apache/plc4x/java/iec608705104/ManualTest.java | 4 +-
.../modbus/base/optimizer/ModbusOptimizer.java | 16 ++---
.../modbus/base/optimizer/ModbusOptimizerTest.java | 4 +-
.../plc4x/java/opcua/optimizer/OpcuaOptimizer.java | 5 +-
.../java/s7/readwrite/optimizer/S7Optimizer.java | 21 +++---
.../s7/readwrite/protocol/S7ProtocolLogic.java | 20 +++---
.../readwrite/utils/S7PlcSubscriptionRequest.java | 66 ++++++++++---------
.../java/spi/connection/AbstractPlcConnection.java | 47 ++++++++++++-
.../java/spi/messages/DefaultPlcReadRequest.java | 24 +++----
.../java/spi/messages/DefaultPlcReadResponse.java | 6 +-
.../messages/DefaultPlcSubscriptionRequest.java | 63 ++++++++++--------
.../messages/DefaultPlcSubscriptionResponse.java | 10 +--
.../java/spi/messages/DefaultPlcWriteRequest.java | 26 ++++----
.../spi/messages/utils/DefaultPlcResponseItem.java | 2 +-
.../spi/messages/utils/DefaultPlcTagErrorItem.java | 3 +-
.../java/spi/messages/utils/DefaultPlcTagItem.java | 8 +--
.../spi/messages/utils/DefaultPlcTagValueItem.java | 4 +-
.../java/spi/messages/utils/PlcResponseItem.java | 4 +-
.../plc4x/java/spi/messages/utils/PlcTagItem.java | 4 +-
.../java/spi/messages/utils/PlcTagValueItem.java | 3 +-
.../java/spi/optimizer/SingleTagOptimizer.java | 4 +-
.../java/utils/cache/LeasedPlcConnection.java | 19 ++++--
26 files changed, 270 insertions(+), 199 deletions(-)
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcSubscriptionRequest.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcSubscriptionRequest.java
index a87a14696c..73f025a2fd 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcSubscriptionRequest.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcSubscriptionRequest.java
@@ -18,7 +18,6 @@
*/
package org.apache.plc4x.java.api.messages;
-import org.apache.plc4x.java.api.model.PlcConsumerRegistration;
import org.apache.plc4x.java.api.model.PlcTag;
import java.time.Duration;
@@ -35,6 +34,14 @@ public interface PlcSubscriptionRequest extends PlcSubscriptionTagRequest {
@Override
PlcSubscriptionRequest build();
+ /**
+ * Define the consumer that will be receiving subscription events.
+ *
+ * @param consumer Consumer
+ * @return builder.
+ */
+ PlcSubscriptionRequest.Builder setConsumer(Consumer<PlcSubscriptionEvent> consumer);
+
/**
* Adds a new tag to the to be constructed request which should be polled cyclically.
*
@@ -99,18 +106,6 @@ public interface PlcSubscriptionRequest extends PlcSubscriptionTagRequest {
*/
PlcSubscriptionRequest.Builder addEventTag(String name, PlcTag tag);
- /**
- * Convenience method which attaches the {@link Consumer<PlcSubscriptionEvent>} directly to the handles once the
- * requests succeeds.
- * Note: opposed to register on the {@link org.apache.plc4x.java.api.model.PlcSubscriptionHandle} directly you
- * won't retrieve a {@link PlcConsumerRegistration} which is useful to cancel registrations.
- *
- * @param name alias of the tag.
- * @param preRegisteredConsumer {@link Consumer<PlcSubscriptionEvent>} to be attached
- * @return builder.
- */
- PlcSubscriptionRequest.Builder addPreRegisteredConsumer(String name, Consumer<PlcSubscriptionEvent> preRegisteredConsumer);
-
}
}
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcSubscriptionTagRequest.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcSubscriptionTagRequest.java
index 323fa83679..bf2ccaae8a 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcSubscriptionTagRequest.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcSubscriptionTagRequest.java
@@ -19,10 +19,10 @@
package org.apache.plc4x.java.api.messages;
import org.apache.plc4x.java.api.model.PlcSubscriptionTag;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
import java.util.LinkedHashSet;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
@@ -35,10 +35,12 @@ public interface PlcSubscriptionTagRequest extends PlcRequest {
LinkedHashSet<String> getTagNames();
+ PlcResponseCode getTagResponseCode(String tagName);
+
PlcSubscriptionTag getTag(String name);
List<PlcSubscriptionTag> getTags();
- Map<String, List<Consumer<PlcSubscriptionEvent>>> getPreRegisteredConsumers();
+ Consumer<PlcSubscriptionEvent> getConsumer();
}
diff --git a/plc4j/drivers/ads/src/main/java/org/apache/plc4x/java/ads/protocol/AdsProtocolLogic.java b/plc4j/drivers/ads/src/main/java/org/apache/plc4x/java/ads/protocol/AdsProtocolLogic.java
index 5a7e9e47bc..ca18d76ce4 100644
--- a/plc4j/drivers/ads/src/main/java/org/apache/plc4x/java/ads/protocol/AdsProtocolLogic.java
+++ b/plc4j/drivers/ads/src/main/java/org/apache/plc4x/java/ads/protocol/AdsProtocolLogic.java
@@ -45,7 +45,9 @@ import org.apache.plc4x.java.spi.connection.PlcTagHandler;
import org.apache.plc4x.java.spi.generation.*;
import org.apache.plc4x.java.spi.messages.*;
import org.apache.plc4x.java.spi.messages.utils.DefaultPlcResponseItem;
+import org.apache.plc4x.java.spi.messages.utils.DefaultPlcTagItem;
import org.apache.plc4x.java.spi.messages.utils.PlcResponseItem;
+import org.apache.plc4x.java.spi.messages.utils.PlcTagItem;
import org.apache.plc4x.java.spi.model.DefaultArrayInfo;
import org.apache.plc4x.java.spi.model.DefaultPlcConsumerRegistration;
import org.apache.plc4x.java.spi.model.DefaultPlcSubscriptionTag;
@@ -408,50 +410,57 @@ public class AdsProtocolLogic extends Plc4xProtocolBase<AmsTCPPacket> implements
}
}
- LinkedHashMap<String, PlcSubscriptionTag> subscriptionTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagItem<PlcSubscriptionTag>> subscriptionTags = new LinkedHashMap<>();
// Subscribe to online-version changes (get the address from the collected data for symbol: "TwinCAT_SystemInfoVarList._AppInfo.OnlineChangeCnt")
- subscriptionTags.put("onlineVersion", new DefaultPlcSubscriptionTag(
+ subscriptionTags.put("onlineVersion", new DefaultPlcTagItem<>(new DefaultPlcSubscriptionTag(
PlcSubscriptionType.CHANGE_OF_STATE,
new SymbolicAdsTag("TwinCAT_SystemInfoVarList._AppInfo.OnlineChangeCnt", org.apache.plc4x.java.api.types.PlcValueType.UDINT, Collections.emptyList()),
- Duration.ofMillis(1000)));
+ Duration.ofMillis(1000))));
// Subscribe to symbol-version changes (Address: GroupID: 0xF008, Offset: 0, Read length: 1)
- subscriptionTags.put("symbolVersion", new DefaultPlcSubscriptionTag(
+ subscriptionTags.put("symbolVersion", new DefaultPlcTagItem<>(new DefaultPlcSubscriptionTag(
PlcSubscriptionType.CHANGE_OF_STATE,
new DirectAdsTag(0xF008, 0x0000, "USINT", 1),
- Duration.ofMillis(1000)));
- LinkedHashMap<String, List<Consumer<PlcSubscriptionEvent>>> consumer = new LinkedHashMap<>();
- consumer.put("onlineVersion", Collections.singletonList(plcSubscriptionEvent -> {
- long oldVersion = onlineVersion;
- long newVersion = plcSubscriptionEvent.getPlcValue("onlineVersion").getLong();
- if (oldVersion != newVersion) {
- if (invalidationLock.tryLock()) {
- LOGGER.info("Detected change of the 'online-version', invalidating data type and symbol information.");
- CompletableFuture<Void> reloadingFuture = readSymbolTableAndDatatypeTable(context);
- reloadingFuture.whenComplete((unused, throwable) -> {
- if (throwable != null) {
- LOGGER.error("Error reloading data type and symbol data", throwable);
+ Duration.ofMillis(1000))));
+ Consumer<PlcSubscriptionEvent> consumer = plcSubscriptionEvent -> {
+ for (String tagName : plcSubscriptionEvent.getTagNames()) {
+ switch (tagName) {
+ case "onlineVersion": {
+ long oldVersion = onlineVersion;
+ long newVersion = plcSubscriptionEvent.getPlcValue("onlineVersion").getLong();
+ if (oldVersion != newVersion) {
+ if (invalidationLock.tryLock()) {
+ LOGGER.info("Detected change of the 'online-version', invalidating data type and symbol information.");
+ CompletableFuture<Void> reloadingFuture = readSymbolTableAndDatatypeTable(context);
+ reloadingFuture.whenComplete((unused, throwable) -> {
+ if (throwable != null) {
+ LOGGER.error("Error reloading data type and symbol data", throwable);
+ }
+ invalidationLock.unlock();
+ });
+ }
}
- invalidationLock.unlock();
- });
- }
- }
- }));
- consumer.put("symbolVersion", Collections.singletonList(plcSubscriptionEvent -> {
- int oldVersion = symbolVersion;
- int newVersion = plcSubscriptionEvent.getPlcValue("symbolVersion").getInteger();
- if (oldVersion != newVersion) {
- if (invalidationLock.tryLock()) {
- LOGGER.info("Detected change of the 'symbol-version', invalidating data type and symbol information.");
- CompletableFuture<Void> reloadingFuture = readSymbolTableAndDatatypeTable(context);
- reloadingFuture.whenComplete((unused, throwable) -> {
- if (throwable != null) {
- LOGGER.error("Error reloading data type and symbol data", throwable);
+ break;
+ }
+ case "symbolVersion": {
+ int oldVersion = symbolVersion;
+ int newVersion = plcSubscriptionEvent.getPlcValue("symbolVersion").getInteger();
+ if (oldVersion != newVersion) {
+ if (invalidationLock.tryLock()) {
+ LOGGER.info("Detected change of the 'symbol-version', invalidating data type and symbol information.");
+ CompletableFuture<Void> reloadingFuture = readSymbolTableAndDatatypeTable(context);
+ reloadingFuture.whenComplete((unused, throwable) -> {
+ if (throwable != null) {
+ LOGGER.error("Error reloading data type and symbol data", throwable);
+ }
+ invalidationLock.unlock();
+ });
+ }
}
- invalidationLock.unlock();
- });
+ }
}
}
- }));
+ };
+
PlcSubscriptionRequest subscriptionRequest = new DefaultPlcSubscriptionRequest(this, subscriptionTags, consumer);
CompletableFuture<PlcSubscriptionResponse> subscriptionResponseCompletableFuture = subscribe(subscriptionRequest);
diff --git a/plc4j/drivers/firmata/src/test/java/org/apache/plc4x/java/firmata/readwrite/FirmataManualTest.java b/plc4j/drivers/firmata/src/test/java/org/apache/plc4x/java/firmata/readwrite/FirmataManualTest.java
index ccf354ff67..5d4c25bffe 100644
--- a/plc4j/drivers/firmata/src/test/java/org/apache/plc4x/java/firmata/readwrite/FirmataManualTest.java
+++ b/plc4j/drivers/firmata/src/test/java/org/apache/plc4x/java/firmata/readwrite/FirmataManualTest.java
@@ -30,7 +30,7 @@ public class FirmataManualTest {
// Default for the Firmata Arduino sketch is, 57600 baud, 8 data-bits, no parity and one stop bit (Which is the default for the Serial transport)
try(PlcConnection connection = PlcDriverManager.getDefault().getConnectionManager().getConnection("firmata:///dev/tty.usbmodem2114401")) {
- PlcSubscriptionRequest subscriptionRequest = connection.subscriptionRequestBuilder().addEventTagAddress("analog1", "analog:1").addPreRegisteredConsumer("analog1", plcSubscriptionEvent -> {
+ PlcSubscriptionRequest subscriptionRequest = connection.subscriptionRequestBuilder().addEventTagAddress("analog1", "analog:1").setConsumer(plcSubscriptionEvent -> {
System.out.println("Incoming Event: " + plcSubscriptionEvent.getPlcValue("analog1").getInteger());
}).build();
PlcSubscriptionResponse plcSubscriptionResponse = subscriptionRequest.execute().get();
diff --git a/plc4j/drivers/iec-60870/src/test/java/org/apache/plc4x/java/iec608705104/ManualTest.java b/plc4j/drivers/iec-60870/src/test/java/org/apache/plc4x/java/iec608705104/ManualTest.java
index 0cbfb28ae7..887d4689dd 100644
--- a/plc4j/drivers/iec-60870/src/test/java/org/apache/plc4x/java/iec608705104/ManualTest.java
+++ b/plc4j/drivers/iec-60870/src/test/java/org/apache/plc4x/java/iec608705104/ManualTest.java
@@ -37,10 +37,10 @@ public class ManualTest {
throw new RuntimeException("Subscription not supported");
}
- plcConnection.subscriptionRequestBuilder().addChangeOfStateTagAddress("all", "*").addPreRegisteredConsumer("all", plcSubscriptionEvent -> {
+ plcConnection.subscriptionRequestBuilder().addChangeOfStateTagAddress("all", "*").setConsumer(plcSubscriptionEvent -> {
for (String tagName : plcSubscriptionEvent.getTagNames()) {
Iec608705104Tag tag = (Iec608705104Tag) plcSubscriptionEvent.getTag(tagName);
- System.out.println(String.format("TS: %s, Addr: %d:%d, Value; %s", plcSubscriptionEvent.getTimestamp().toString(), tag.getAdsuAddress(), tag.getObjectAddress(), plcSubscriptionEvent.getPlcValue(tagName).toString()));
+ System.out.printf("TS: %s, Addr: %d:%d, Value; %s%n", plcSubscriptionEvent.getTimestamp().toString(), tag.getAdsuAddress(), tag.getObjectAddress(), plcSubscriptionEvent.getPlcValue(tagName).toString());
}
}).build().execute();
diff --git a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/base/optimizer/ModbusOptimizer.java b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/base/optimizer/ModbusOptimizer.java
index 78b092282c..db838c6a0b 100644
--- a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/base/optimizer/ModbusOptimizer.java
+++ b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/base/optimizer/ModbusOptimizer.java
@@ -305,8 +305,8 @@ public class ModbusOptimizer extends SingleTagOptimizer {
// finish this one and start a new one.
if (tag.getAddress() + (sizeInCoils * tag.getNumberOfElements()) > maxCoilCurRequest) {
// Finish the current sub-request
- LinkedHashMap<String, PlcTagItem> subTags = new LinkedHashMap<>();
- subTags.put("coils" + subRequests.size(), new DefaultPlcTagItem(new ModbusTagCoil(firstCoil, lastCoil - firstCoil, ModbusDataType.BYTE, Collections.emptyMap())));
+ LinkedHashMap<String, PlcTagItem<PlcTag>> subTags = new LinkedHashMap<>();
+ subTags.put("coils" + subRequests.size(), new DefaultPlcTagItem<>(new ModbusTagCoil(firstCoil, lastCoil - firstCoil, ModbusDataType.BYTE, Collections.emptyMap())));
subRequests.add(new DefaultPlcReadRequest(reader, subTags));
// Re-initialize the structures for the next request.
@@ -321,8 +321,8 @@ public class ModbusOptimizer extends SingleTagOptimizer {
}
// Finish the last sub-request
- LinkedHashMap<String, PlcTagItem> subTags = new LinkedHashMap<>();
- subTags.put("coils" + subRequests.size(), new DefaultPlcTagItem(new ModbusTagCoil(firstCoil, lastCoil - firstCoil, ModbusDataType.BYTE, Collections.emptyMap())));
+ LinkedHashMap<String, PlcTagItem<PlcTag>> subTags = new LinkedHashMap<>();
+ subTags.put("coils" + subRequests.size(), new DefaultPlcTagItem<>(new ModbusTagCoil(firstCoil, lastCoil - firstCoil, ModbusDataType.BYTE, Collections.emptyMap())));
subRequests.add(new DefaultPlcReadRequest(reader, subTags));
return subRequests;
}
@@ -346,8 +346,8 @@ public class ModbusOptimizer extends SingleTagOptimizer {
// finish this one and start a new one.
if (tag.getAddress() + (sizeInRegisters * tag.getNumberOfElements()) > maxRegisterCurRequest) {
// Finish the current sub-request
- LinkedHashMap<String, PlcTagItem> subTags = new LinkedHashMap<>();
- subTags.put("registers" + subRequests.size(), new DefaultPlcTagItem(tagFactory.createTag(firstRegister, lastRegister - firstRegister, ModbusDataType.WORD)));
+ LinkedHashMap<String, PlcTagItem<PlcTag>> subTags = new LinkedHashMap<>();
+ subTags.put("registers" + subRequests.size(), new DefaultPlcTagItem<>(tagFactory.createTag(firstRegister, lastRegister - firstRegister, ModbusDataType.WORD)));
subRequests.add(new DefaultPlcReadRequest(reader, subTags));
// Re-initialize the structures for the next request.
@@ -362,8 +362,8 @@ public class ModbusOptimizer extends SingleTagOptimizer {
}
// Finish the last sub-request
- LinkedHashMap<String, PlcTagItem> subTags = new LinkedHashMap<>();
- subTags.put("registers" + subRequests.size(), new DefaultPlcTagItem(tagFactory.createTag(firstRegister, lastRegister - firstRegister, ModbusDataType.WORD)));
+ LinkedHashMap<String, PlcTagItem<PlcTag>> subTags = new LinkedHashMap<>();
+ subTags.put("registers" + subRequests.size(), new DefaultPlcTagItem<>(tagFactory.createTag(firstRegister, lastRegister - firstRegister, ModbusDataType.WORD)));
subRequests.add(new DefaultPlcReadRequest(reader, subTags));
return subRequests;
}
diff --git a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/base/optimizer/ModbusOptimizerTest.java b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/base/optimizer/ModbusOptimizerTest.java
index be18764f70..6f2559e286 100644
--- a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/base/optimizer/ModbusOptimizerTest.java
+++ b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/base/optimizer/ModbusOptimizerTest.java
@@ -227,10 +227,10 @@ class ModbusOptimizerTest {
Mockito.when(driverContext.getByteOrder()).thenReturn(ModbusByteOrder.BIG_ENDIAN);
Mockito.when(driverContext.getMaxCoilsPerRequest()).thenReturn(2000);
Mockito.when(driverContext.getMaxRegistersPerRequest()).thenReturn(125);
- LinkedHashMap<String, PlcTagItem> tagMap = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagItem<PlcTag>> tagMap = new LinkedHashMap<>();
int i = 0;
for (PlcTag tag : tags) {
- tagMap.put("tag" + i++, new DefaultPlcTagItem(tag));
+ tagMap.put("tag" + i++, new DefaultPlcTagItem<>(tag));
}
ModbusOptimizer sut = new ModbusOptimizer();
List<PlcReadRequest> plcReadRequests = sut.processReadRequest(new DefaultPlcReadRequest(reader, tagMap), driverContext);
diff --git a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/optimizer/OpcuaOptimizer.java b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/optimizer/OpcuaOptimizer.java
index 7b61baddaf..522440f984 100644
--- a/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/optimizer/OpcuaOptimizer.java
+++ b/plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/optimizer/OpcuaOptimizer.java
@@ -19,6 +19,7 @@
package org.apache.plc4x.java.opcua.optimizer;
import org.apache.plc4x.java.api.messages.PlcReadRequest;
+import org.apache.plc4x.java.api.model.PlcTag;
import org.apache.plc4x.java.opcua.tag.OpcuaTag;
import org.apache.plc4x.java.spi.context.DriverContext;
import org.apache.plc4x.java.spi.messages.DefaultPlcReadRequest;
@@ -37,11 +38,11 @@ public class OpcuaOptimizer extends BaseOptimizer{
List<PlcReadRequest> processedRequests = new LinkedList<>();
// List of all items in the current request.
- LinkedHashMap<String, PlcTagItem> curTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagItem<PlcTag>> curTags = new LinkedHashMap<>();
for (String tagName : readRequest.getTagNames()) {
OpcuaTag tag = (OpcuaTag) readRequest.getTag(tagName);
- curTags.put(tagName, new DefaultPlcTagItem(tag));
+ curTags.put(tagName, new DefaultPlcTagItem<>(tag));
}
// Create a new PlcReadRequest from the remaining tag items.
diff --git a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/optimizer/S7Optimizer.java b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/optimizer/S7Optimizer.java
index b48512c9a0..108ae80b38 100644
--- a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/optimizer/S7Optimizer.java
+++ b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/optimizer/S7Optimizer.java
@@ -22,6 +22,7 @@ import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
import org.apache.plc4x.java.api.messages.PlcReadRequest;
import org.apache.plc4x.java.api.messages.PlcReadResponse;
import org.apache.plc4x.java.api.messages.PlcWriteRequest;
+import org.apache.plc4x.java.api.model.PlcTag;
import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.api.types.PlcValueType;
import org.apache.plc4x.java.api.value.PlcValue;
@@ -81,7 +82,7 @@ public class S7Optimizer extends BaseOptimizer {
// List of all items in the current request.
- LinkedHashMap<String, PlcTagItem> curTagItems = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagItem<PlcTag>> curTagItems = new LinkedHashMap<>();
for (String tagName : readRequest.getTagNames()) {
@@ -89,7 +90,7 @@ public class S7Optimizer extends BaseOptimizer {
if ((readRequest.getTag(tagName) instanceof S7SzlTag) ||
(readRequest.getTag(tagName) instanceof S7ClkTag)) {
// We are only expecting valid tagValueItems being passed in.
- curTagItems.put(tagName, new DefaultPlcTagItem(readRequest.getTag(tagName)));
+ curTagItems.put(tagName, new DefaultPlcTagItem<>(readRequest.getTag(tagName)));
continue;
}
@@ -124,7 +125,7 @@ public class S7Optimizer extends BaseOptimizer {
// Add the tag to the current request
// We are only expecting valid tagValueItems being passed in.
- curTagItems.put(tagName, new DefaultPlcTagItem(tag));
+ curTagItems.put(tagName, new DefaultPlcTagItem<>(tag));
}
// If the current item would exceed the PDU size in the response, even if this is a new request, split
// up the array to fill up the last request and create so many sub-requests that read all in total.
@@ -137,8 +138,8 @@ public class S7Optimizer extends BaseOptimizer {
for(int curRequest = 0; curRequest < numRequests; curRequest++) {
int numCurRequestItems = Math.min(numItemsPerRequest, itemsLeft);
S7Tag tagFragment = new S7Tag(tag.getDataType(), tag.getMemoryArea(), tag.getBlockNumber(), curByteOffset, (byte) 0, numCurRequestItems);
- LinkedHashMap<String, PlcTagItem> tagFragments = new LinkedHashMap<>();
- tagFragments.put(tagName, new DefaultPlcTagItem(tagFragment));
+ LinkedHashMap<String, PlcTagItem<PlcTag>> tagFragments = new LinkedHashMap<>();
+ tagFragments.put(tagName, new DefaultPlcTagItem<>(tagFragment));
processedRequests.add(new DefaultPlcReadRequest(((DefaultPlcReadRequest) readRequest).getReader(), tagFragments));
curByteOffset += numItemsPerRequest * tag.getDataType().getSizeInBytes();
itemsLeft -= numCurRequestItems;
@@ -163,7 +164,7 @@ public class S7Optimizer extends BaseOptimizer {
// Add the tag to the new current request
// We are only expecting valid tagValueItems being passed in.
- curTagItems.put(tagName, new DefaultPlcTagItem(tag));
+ curTagItems.put(tagName, new DefaultPlcTagItem<>(tag));
}
}
@@ -199,7 +200,7 @@ public class S7Optimizer extends BaseOptimizer {
if (tagValues.containsKey(tagName)) {
PlcResponseItem<PlcValue> existingTagItem = tagValues.get(tagName);
// If a previous response was invalid, we'll discard this part of it too.
- if(existingTagItem.getCode() != PlcResponseCode.OK) {
+ if(existingTagItem.getResponseCode() != PlcResponseCode.OK) {
continue;
}
existingItem = (PlcRawByteArray) existingTagItem.getValue();
@@ -218,7 +219,7 @@ public class S7Optimizer extends BaseOptimizer {
if (tagValues.containsKey(tagName)) {
PlcResponseItem<PlcValue> existingTagItem = tagValues.get(tagName);
// If a previous response was invalid, we'll discard this part of it too.
- if(existingTagItem.getCode() != PlcResponseCode.OK) {
+ if(existingTagItem.getResponseCode() != PlcResponseCode.OK) {
continue;
}
existingItems = (List<PlcValue>) existingTagItem.getValue().getList();
@@ -267,7 +268,7 @@ public class S7Optimizer extends BaseOptimizer {
int curResponseSize = EMPTY_WRITE_RESPONSE_SIZE;
// List of all items in the current request.
- LinkedHashMap<String, PlcTagValueItem> curTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagValueItem<PlcTag>> curTags = new LinkedHashMap<>();
for (String tagName : writeRequest.getTagNames()) {
S7Tag tag = (S7Tag) writeRequest.getTag(tagName);
@@ -333,7 +334,7 @@ public class S7Optimizer extends BaseOptimizer {
}
}
// We are only expecting valid tagValueItems being passed in.
- curTags.put(tagName, new DefaultPlcTagValueItem(tag, value));
+ curTags.put(tagName, new DefaultPlcTagValueItem<>(tag, value));
}
// Create a new PlcWriteRequest from the remaining tag items.
diff --git a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java
index c50c1bd39c..8f4eee69c2 100644
--- a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java
+++ b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java
@@ -1300,17 +1300,17 @@ public class S7ProtocolLogic extends Plc4xProtocolBase<TPKTPacket> {
// Replace the var-length string fields with requests to read the
// length information instead of the string content.
int numVarLengthStrings = 0;
- LinkedHashMap<String, PlcTagItem> updatedRequestItems = new LinkedHashMap<>(request.getNumberOfTags());
+ LinkedHashMap<String, PlcTagItem<PlcTag>> updatedRequestItems = new LinkedHashMap<>(request.getNumberOfTags());
for (String tagName : request.getTagNames()) {
- PlcTagItem plcTagItem = request.getTagItem(tagName);
+ PlcTagItem<PlcTag> plcTagItem = request.getTagItem(tagName);
if(plcTagItem.getTag() instanceof S7StringVarLengthTag) {
S7Tag s7Tag = (S7Tag) plcTagItem.getTag();
TransportSize dataType = s7Tag.getDataType();
if(dataType == TransportSize.STRING) {
- updatedRequestItems.put(tagName, new DefaultPlcTagItem(new S7Tag(TransportSize.BYTE, s7Tag.getMemoryArea(), s7Tag.getBlockNumber(), s7Tag.getByteOffset(), s7Tag.getBitOffset(), 2)));
+ updatedRequestItems.put(tagName, new DefaultPlcTagItem<>(new S7Tag(TransportSize.BYTE, s7Tag.getMemoryArea(), s7Tag.getBlockNumber(), s7Tag.getByteOffset(), s7Tag.getBitOffset(), 2)));
numVarLengthStrings++;
} else if(dataType == TransportSize.WSTRING) {
- updatedRequestItems.put(tagName, new DefaultPlcTagItem(new S7Tag(TransportSize.BYTE, s7Tag.getMemoryArea(), s7Tag.getBlockNumber(), s7Tag.getByteOffset(), s7Tag.getBitOffset(), 4)));
+ updatedRequestItems.put(tagName, new DefaultPlcTagItem<>(new S7Tag(TransportSize.BYTE, s7Tag.getMemoryArea(), s7Tag.getBlockNumber(), s7Tag.getByteOffset(), s7Tag.getBitOffset(), 4)));
numVarLengthStrings++;
}
} else {
@@ -1326,7 +1326,7 @@ public class S7ProtocolLogic extends Plc4xProtocolBase<TPKTPacket> {
return;
}
// Collect the responses for the var-length strings and read them separately.
- LinkedHashMap<String, PlcTagItem> varLengthStringTags = new LinkedHashMap<>(finalNumVarLengthStrings);
+ LinkedHashMap<String, PlcTagItem<PlcTag>> varLengthStringTags = new LinkedHashMap<>(finalNumVarLengthStrings);
int curItem = 0;
for (String tagName : request.getTagNames()) {
S7Tag s7tag = (S7Tag) request.getTag(tagName);
@@ -1339,11 +1339,11 @@ public class S7ProtocolLogic extends Plc4xProtocolBase<TPKTPacket> {
if (s7tag.getDataType() == TransportSize.STRING) {
rb.readShort(8);
int stringLength = rb.readShort(8);
- varLengthStringTags.put(tagName, new DefaultPlcTagItem(new S7StringFixedLengthTag(TransportSize.STRING, s7tag.getMemoryArea(), s7tag.getBlockNumber(), s7tag.getByteOffset(), s7tag.getBitOffset(), 1, stringLength)));
+ varLengthStringTags.put(tagName, new DefaultPlcTagItem<>(new S7StringFixedLengthTag(TransportSize.STRING, s7tag.getMemoryArea(), s7tag.getBlockNumber(), s7tag.getByteOffset(), s7tag.getBitOffset(), 1, stringLength)));
} else if (s7tag.getDataType() == TransportSize.WSTRING) {
rb.readInt(16);
int stringLength = rb.readInt(16);
- varLengthStringTags.put(tagName, new DefaultPlcTagItem(new S7StringFixedLengthTag(TransportSize.WSTRING, s7tag.getMemoryArea(), s7tag.getBlockNumber(), s7tag.getByteOffset(), s7tag.getBitOffset(), 1, stringLength)));
+ varLengthStringTags.put(tagName, new DefaultPlcTagItem<>(new S7StringFixedLengthTag(TransportSize.WSTRING, s7tag.getMemoryArea(), s7tag.getBlockNumber(), s7tag.getByteOffset(), s7tag.getBitOffset(), 1, stringLength)));
}
} catch (Exception e) {
logger.warn("Error parsing string size for tag {}", tagName, e);
@@ -1438,7 +1438,7 @@ public class S7ProtocolLogic extends Plc4xProtocolBase<TPKTPacket> {
} else {
// Create an alternative list of request items, where all var-length string tags are replaced with
// fixed-length string tags using the string length returned by the previous request.
- LinkedHashMap<String, PlcTagValueItem> updatedRequestItems = new LinkedHashMap<>(request.getNumberOfTags());
+ LinkedHashMap<String, PlcTagValueItem<PlcTag>> updatedRequestItems = new LinkedHashMap<>(request.getNumberOfTags());
for (String tagName : request.getTagNames()) {
PlcTag tag = request.getTag(tagName);
PlcValue value = request.getPlcValue(tagName);
@@ -1448,9 +1448,9 @@ public class S7ProtocolLogic extends Plc4xProtocolBase<TPKTPacket> {
S7StringFixedLengthTag newTag = new S7StringFixedLengthTag(varLengthTag.getDataType(), varLengthTag.getMemoryArea(),
varLengthTag.getBlockNumber(), varLengthTag.getByteOffset(), varLengthTag.getBitOffset(),
varLengthTag.getNumberOfElements(), stringLength);
- updatedRequestItems.put(tagName, new DefaultPlcTagValueItem(newTag, value));
+ updatedRequestItems.put(tagName, new DefaultPlcTagValueItem<>(newTag, value));
} else {
- updatedRequestItems.put(tagName, new DefaultPlcTagValueItem(tag, value));
+ updatedRequestItems.put(tagName, new DefaultPlcTagValueItem<>(tag, value));
}
}
diff --git a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/S7PlcSubscriptionRequest.java b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/S7PlcSubscriptionRequest.java
index e8b8296f7f..5bba1d31bd 100644
--- a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/S7PlcSubscriptionRequest.java
+++ b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/S7PlcSubscriptionRequest.java
@@ -24,10 +24,13 @@ import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse;
import org.apache.plc4x.java.api.model.PlcSubscriptionTag;
import org.apache.plc4x.java.api.model.PlcTag;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.api.types.PlcSubscriptionType;
import org.apache.plc4x.java.spi.connection.PlcTagHandler;
import org.apache.plc4x.java.spi.generation.SerializationException;
import org.apache.plc4x.java.spi.generation.WriteBuffer;
+import org.apache.plc4x.java.spi.messages.utils.DefaultPlcTagItem;
+import org.apache.plc4x.java.spi.messages.utils.PlcTagItem;
import org.apache.plc4x.java.spi.model.DefaultPlcSubscriptionTag;
import org.apache.plc4x.java.spi.utils.Serializable;
@@ -36,6 +39,8 @@ import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
import org.apache.plc4x.java.s7.readwrite.TimeBase;
import org.apache.plc4x.java.s7.readwrite.tag.S7SubscriptionTag;
import org.apache.plc4x.java.s7.readwrite.tag.S7Tag;
@@ -51,16 +56,16 @@ public class S7PlcSubscriptionRequest implements PlcSubscriptionRequest, Seriali
private final PlcSubscriber subscriber;
- private final LinkedHashMap<String, PlcSubscriptionTag> tags;
+ private final LinkedHashMap<String, PlcTagItem<PlcSubscriptionTag>> tags;
- private final LinkedHashMap<String, List<Consumer<PlcSubscriptionEvent>>> preRegisteredConsumers;
+ private final Consumer<PlcSubscriptionEvent> consumer;
public S7PlcSubscriptionRequest(PlcSubscriber subscriber,
- LinkedHashMap<String, PlcSubscriptionTag> tags,
- LinkedHashMap<String, List<Consumer<PlcSubscriptionEvent>>> preRegisteredConsumers) {
+ LinkedHashMap<String, PlcTagItem<PlcSubscriptionTag>> tags,
+ Consumer<PlcSubscriptionEvent> consumer) {
this.subscriber = subscriber;
this.tags = tags;
- this.preRegisteredConsumers = preRegisteredConsumers;
+ this.consumer = consumer;
}
@Override
@@ -80,17 +85,22 @@ public class S7PlcSubscriptionRequest implements PlcSubscriptionRequest, Seriali
@Override
public PlcSubscriptionTag getTag(String name) {
- return tags.get(name);
+ return tags.get(name).getTag();
}
@Override
- public List<PlcSubscriptionTag> getTags() {
- return new ArrayList<>(tags.values());
+ public PlcResponseCode getTagResponseCode(String tagName) {
+ return tags.get(tagName).getResponseCode();
+ }
+
+ @Override
+ public Consumer<PlcSubscriptionEvent> getConsumer() {
+ return consumer;
}
@Override
- public Map<String, List<Consumer<PlcSubscriptionEvent>>> getPreRegisteredConsumers() {
- return new LinkedHashMap<>(preRegisteredConsumers);
+ public List<PlcSubscriptionTag> getTags() {
+ return tags.values().stream().map(PlcTagItem::getTag).collect(Collectors.toCollection(LinkedList::new));
}
@Override
@@ -98,14 +108,14 @@ public class S7PlcSubscriptionRequest implements PlcSubscriptionRequest, Seriali
writeBuffer.pushContext("PlcSubscriptionRequest");
writeBuffer.pushContext("tags");
- for (Map.Entry<String, PlcSubscriptionTag> tagEntry : tags.entrySet()) {
+ for (Map.Entry<String, PlcTagItem<PlcSubscriptionTag>> tagEntry : tags.entrySet()) {
String tagName = tagEntry.getKey();
writeBuffer.pushContext(tagName);
- PlcTag tag = tagEntry.getValue();
- if (!(tag instanceof Serializable)) {
+ PlcTagItem<PlcSubscriptionTag> tagItem = tagEntry.getValue();
+ if (!(tagItem instanceof Serializable)) {
throw new PlcRuntimeException("Error serializing. Tag doesn't implement XmlSerializable");
}
- ((Serializable) tag).serialize(writeBuffer);
+ ((Serializable) tagItem).serialize(writeBuffer);
writeBuffer.popContext(tagName);
}
writeBuffer.popContext("tags");
@@ -118,13 +128,18 @@ public class S7PlcSubscriptionRequest implements PlcSubscriptionRequest, Seriali
private final PlcSubscriber subscriber;
private final PlcTagHandler tagHandler;
private final Map<String, BuilderItem> tags;
- private final LinkedHashMap<String, List<Consumer<PlcSubscriptionEvent>>> preRegisteredConsumers;
+ private Consumer<PlcSubscriptionEvent> consumer;
public Builder(PlcSubscriber subscriber, PlcTagHandler tagHandler) {
this.subscriber = subscriber;
this.tagHandler = tagHandler;
this.tags = new TreeMap<>();
- this.preRegisteredConsumers = new LinkedHashMap<>();
+ }
+
+ @Override
+ public PlcSubscriptionRequest.Builder setConsumer(Consumer<PlcSubscriptionEvent> consumer) {
+ this.consumer = Objects.requireNonNull(consumer);
+ return this;
}
/*
@@ -274,27 +289,16 @@ public class S7PlcSubscriptionRequest implements PlcSubscriptionRequest, Seriali
return this;
}
- @Override
- public PlcSubscriptionRequest.Builder addPreRegisteredConsumer(String name, Consumer<PlcSubscriptionEvent> consumer) {
- preRegisteredConsumers.putIfAbsent(name, new LinkedList<>());
- preRegisteredConsumers.get(name).add(consumer);
- return this;
- }
-
@Override
public PlcSubscriptionRequest build() {
- LinkedHashMap<String, PlcSubscriptionTag> parsedTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagItem<PlcSubscriptionTag>> parsedTags = new LinkedHashMap<>();
tags.forEach((name, builderItem) -> {
PlcTag parsedTag = builderItem.tag.get();
- parsedTags.put(name, new DefaultPlcSubscriptionTag(builderItem.plcSubscriptionType, parsedTag, builderItem.duration));
+ parsedTags.put(name, new DefaultPlcTagItem<>(new DefaultPlcSubscriptionTag(builderItem.plcSubscriptionType, parsedTag, builderItem.duration)));
});
- preRegisteredConsumers.forEach((tagName, ignored) -> {
- if (!tags.containsKey(tagName)) {
- throw new PlcRuntimeException("tagName " + tagName + "for preRegisteredConsumer not found");
- }
- });
- return new S7PlcSubscriptionRequest(subscriber, parsedTags, preRegisteredConsumers);
+
+ return new S7PlcSubscriptionRequest(subscriber, parsedTags, consumer);
}
private static class BuilderItem {
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/connection/AbstractPlcConnection.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/connection/AbstractPlcConnection.java
index fb321b60f1..52d5603c2d 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/connection/AbstractPlcConnection.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/connection/AbstractPlcConnection.java
@@ -27,6 +27,7 @@ import org.apache.plc4x.java.api.messages.*;
import org.apache.plc4x.java.api.metadata.PlcConnectionMetadata;
import org.apache.plc4x.java.api.model.PlcConsumerRegistration;
import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
+import org.apache.plc4x.java.api.model.PlcSubscriptionTag;
import org.apache.plc4x.java.api.model.PlcTag;
import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.api.value.PlcValue;
@@ -34,6 +35,7 @@ import org.apache.plc4x.java.spi.Plc4xProtocolBase;
import org.apache.plc4x.java.spi.generation.Message;
import org.apache.plc4x.java.spi.messages.*;
import org.apache.plc4x.java.spi.messages.utils.DefaultPlcResponseItem;
+import org.apache.plc4x.java.spi.messages.utils.DefaultPlcTagItem;
import org.apache.plc4x.java.spi.messages.utils.PlcResponseItem;
import org.apache.plc4x.java.spi.messages.utils.PlcTagItem;
import org.apache.plc4x.java.spi.messages.utils.PlcTagValueItem;
@@ -269,12 +271,43 @@ public abstract class AbstractPlcConnection implements PlcConnection, PlcConnect
@Override
public CompletableFuture<PlcSubscriptionResponse> subscribe(PlcSubscriptionRequest subscriptionRequest) {
+ PlcSubscriptionRequest filteredSubscriptionRequest = getFilteredSubscriptionRequest((DefaultPlcSubscriptionRequest) subscriptionRequest);
+ return internalSubscribe(filteredSubscriptionRequest)
+ .thenApply(filteredSubscriptionResponse -> {
+ // Shortcut for the case that all tags were valid.
+ if(subscriptionRequest.getNumberOfTags() == filteredSubscriptionRequest.getNumberOfTags()) {
+ return filteredSubscriptionResponse;
+ }
+
+ Map<String, PlcResponseItem<PlcSubscriptionHandle>> values = new HashMap<>();
+ for (String tagName : subscriptionRequest.getTagNames()) {
+ // If the tag was correct, then we expect a response in the response.
+ if(filteredSubscriptionRequest.getTagResponseCode(tagName) != null) {
+ values.put(tagName, new DefaultPlcResponseItem<>(filteredSubscriptionRequest.getTagResponseCode(tagName), filteredSubscriptionResponse.getSubscriptionHandle(tagName)));
+ }
+ // In all other cases forward the initial error to the final output.
+ else {
+ values.put(tagName, new DefaultPlcResponseItem<>(subscriptionRequest.getTagResponseCode(tagName), null));
+ }
+ }
+ return new DefaultPlcSubscriptionResponse(subscriptionRequest, values);
+ });
+ }
+
+ public CompletableFuture<PlcSubscriptionResponse> internalSubscribe(PlcSubscriptionRequest subscriptionRequest) {
if(optimizer != null) {
return optimizer.optimizedSubscribe(subscriptionRequest, protocol);
}
return protocol.subscribe(subscriptionRequest);
}
+ /**
+ * No need to do the whole innerUnsubscribe thing here, as the request only contains handles
+ * returned by the previous subscription request. So we've already filtered out the invalid
+ * tags.
+ * @param unsubscriptionRequest unsubscription request containing at least one unsubscription request item.
+ * @return future for the response.
+ */
@Override
public CompletableFuture<PlcUnsubscriptionResponse> unsubscribe(PlcUnsubscriptionRequest unsubscriptionRequest) {
if(optimizer != null) {
@@ -326,7 +359,7 @@ public abstract class AbstractPlcConnection implements PlcConnection, PlcConnect
}
protected PlcReadRequest getFilteredReadRequest(DefaultPlcReadRequest readRequest) {
- LinkedHashMap<String, PlcTagItem> filteredTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagItem<PlcTag>> filteredTags = new LinkedHashMap<>();
for (String tagName : readRequest.getTagNames()) {
if(readRequest.getTagResponseCode(tagName) == PlcResponseCode.OK) {
filteredTags.put(tagName, readRequest.getTagItem(tagName));
@@ -336,7 +369,7 @@ public abstract class AbstractPlcConnection implements PlcConnection, PlcConnect
}
protected PlcWriteRequest getFilteredWriteRequest(DefaultPlcWriteRequest writeRequest) {
- LinkedHashMap<String, PlcTagValueItem> filteredTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagValueItem<PlcTag>> filteredTags = new LinkedHashMap<>();
for (String tagName : writeRequest.getTagNames()) {
if(writeRequest.getTagResponseCode(tagName) == PlcResponseCode.OK) {
filteredTags.put(tagName, writeRequest.getTagValueItem(tagName));
@@ -345,4 +378,14 @@ public abstract class AbstractPlcConnection implements PlcConnection, PlcConnect
return new DefaultPlcWriteRequest(writeRequest.getWriter(), filteredTags);
}
+ protected PlcSubscriptionRequest getFilteredSubscriptionRequest(DefaultPlcSubscriptionRequest subscriptionRequest) {
+ LinkedHashMap<String, PlcTagItem<PlcSubscriptionTag>> filteredTags = new LinkedHashMap<>();
+ for (String tagName : subscriptionRequest.getTagNames()) {
+ if(subscriptionRequest.getTagResponseCode(tagName) == PlcResponseCode.OK) {
+ filteredTags.put(tagName, new DefaultPlcTagItem<>(subscriptionRequest.getTag(tagName)));
+ }
+ }
+ return new DefaultPlcSubscriptionRequest(subscriptionRequest.getSubscriber(), filteredTags, subscriptionRequest.getConsumer());
+ }
+
}
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcReadRequest.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcReadRequest.java
index 92c64c4ebb..ebc112b26a 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcReadRequest.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcReadRequest.java
@@ -48,10 +48,10 @@ public class DefaultPlcReadRequest implements PlcReadRequest, PlcTagRequest, Ser
private final PlcReader reader;
// This is intentionally a linked hash map in order to keep the order of how elements were added.
- private final LinkedHashMap<String, PlcTagItem> tags;
+ private final LinkedHashMap<String, PlcTagItem<PlcTag>> tags;
public DefaultPlcReadRequest(PlcReader reader,
- LinkedHashMap<String, PlcTagItem> tags) {
+ LinkedHashMap<String, PlcTagItem<PlcTag>> tags) {
this.reader = reader;
this.tags = tags;
}
@@ -72,7 +72,7 @@ public class DefaultPlcReadRequest implements PlcReadRequest, PlcTagRequest, Ser
return new LinkedHashSet<>(tags.keySet());
}
- public PlcTagItem getTagItem(String tagName) {
+ public PlcTagItem<PlcTag> getTagItem(String tagName) {
return tags.get(tagName);
}
@@ -101,10 +101,10 @@ public class DefaultPlcReadRequest implements PlcReadRequest, PlcTagRequest, Ser
writeBuffer.pushContext("PlcTagRequest");
writeBuffer.pushContext("tags", WithRenderAsList(true));
- for (Map.Entry<String, PlcTagItem> tagEntry : tags.entrySet()) {
+ for (Map.Entry<String, PlcTagItem<PlcTag>> tagEntry : tags.entrySet()) {
String tagName = tagEntry.getKey();
writeBuffer.pushContext(tagName);
- PlcTagItem tagItem = tagEntry.getValue();
+ PlcTagItem<PlcTag> tagItem = tagEntry.getValue();
if (!(tagItem instanceof Serializable)) {
throw new RuntimeException("Error serializing. PlcTagItem doesn't implement Serializable");
}
@@ -121,7 +121,7 @@ public class DefaultPlcReadRequest implements PlcReadRequest, PlcTagRequest, Ser
private final PlcReader reader;
private final PlcTagHandler tagHandler;
- private final Map<String, Supplier<PlcTagItem>> tagItems;
+ private final Map<String, Supplier<PlcTagItem<PlcTag>>> tagItems;
public Builder(PlcReader reader, PlcTagHandler tagHandler) {
this.reader = Objects.requireNonNull(reader);
@@ -137,11 +137,11 @@ public class DefaultPlcReadRequest implements PlcReadRequest, PlcTagRequest, Ser
tagItems.put(name, () -> {
try {
PlcTag tag = tagHandler.parseTag(tagAddress);
- return new DefaultPlcTagItem(tag);
+ return new DefaultPlcTagItem<>(tag);
} catch (PlcTagNotFoundException e) {
- return new DefaultPlcTagErrorItem(PlcResponseCode.NOT_FOUND);
+ return new DefaultPlcTagErrorItem<>(PlcResponseCode.NOT_FOUND);
} catch (Exception e) {
- return new DefaultPlcTagErrorItem(PlcResponseCode.INVALID_ADDRESS);
+ return new DefaultPlcTagErrorItem<>(PlcResponseCode.INVALID_ADDRESS);
}
});
return this;
@@ -152,15 +152,15 @@ public class DefaultPlcReadRequest implements PlcReadRequest, PlcTagRequest, Ser
if (tagItems.containsKey(name)) {
throw new PlcRuntimeException("Duplicate tag definition '" + name + "'");
}
- tagItems.put(name, () -> new DefaultPlcTagItem(tag));
+ tagItems.put(name, () -> new DefaultPlcTagItem<>(tag));
return this;
}
@Override
public PlcReadRequest build() {
- LinkedHashMap<String, PlcTagItem> parsedTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagItem<PlcTag>> parsedTags = new LinkedHashMap<>();
tagItems.forEach((name, tagItemSupplier) -> {
- PlcTagItem plcTagItem = tagItemSupplier.get();
+ PlcTagItem<PlcTag> plcTagItem = tagItemSupplier.get();
parsedTags.put(name, plcTagItem);
});
return new DefaultPlcReadRequest(reader, parsedTags);
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcReadResponse.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcReadResponse.java
index a4de7c7c99..5983e519f6 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcReadResponse.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcReadResponse.java
@@ -103,7 +103,7 @@ public class DefaultPlcReadResponse implements PlcReadResponse, Serializable {
if (values.get(name) == null) {
throw new PlcInvalidTagException(name);
}
- return values.get(name).getCode();
+ return values.get(name).getResponseCode();
}
public Map<String, PlcResponseItem<PlcValue>> getValues() {
@@ -623,9 +623,9 @@ public class DefaultPlcReadResponse implements PlcReadResponse, Serializable {
if (values.get(name) == null) {
throw new PlcInvalidTagException(name);
}
- if (values.get(name).getCode() != PlcResponseCode.OK) {
+ if (values.get(name).getResponseCode() != PlcResponseCode.OK) {
throw new PlcRuntimeException(
- "Tag '" + name + "' could not be fetched, response was " + values.get(name).getCode());
+ "Tag '" + name + "' could not be fetched, response was " + values.get(name).getResponseCode());
}
// No need to check for "null" as this is already captured by the constructors.
return values.get(name).getValue();
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcSubscriptionRequest.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcSubscriptionRequest.java
index acd6bd23dd..a0555b5a0f 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcSubscriptionRequest.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcSubscriptionRequest.java
@@ -24,10 +24,13 @@ import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse;
import org.apache.plc4x.java.api.model.PlcSubscriptionTag;
import org.apache.plc4x.java.api.model.PlcTag;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.api.types.PlcSubscriptionType;
import org.apache.plc4x.java.spi.connection.PlcTagHandler;
import org.apache.plc4x.java.spi.generation.SerializationException;
import org.apache.plc4x.java.spi.generation.WriteBuffer;
+import org.apache.plc4x.java.spi.messages.utils.DefaultPlcTagItem;
+import org.apache.plc4x.java.spi.messages.utils.PlcTagItem;
import org.apache.plc4x.java.spi.model.DefaultPlcSubscriptionTag;
import org.apache.plc4x.java.spi.utils.Serializable;
@@ -36,21 +39,22 @@ import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Supplier;
+import java.util.stream.Collectors;
public class DefaultPlcSubscriptionRequest implements PlcSubscriptionRequest, Serializable {
private final PlcSubscriber subscriber;
- private final LinkedHashMap<String, PlcSubscriptionTag> tags;
+ private final LinkedHashMap<String, PlcTagItem<PlcSubscriptionTag>> tags;
- private final LinkedHashMap<String, List<Consumer<PlcSubscriptionEvent>>> preRegisteredConsumers;
+ private final Consumer<PlcSubscriptionEvent> consumer;
public DefaultPlcSubscriptionRequest(PlcSubscriber subscriber,
- LinkedHashMap<String, PlcSubscriptionTag> tags,
- LinkedHashMap<String, List<Consumer<PlcSubscriptionEvent>>> preRegisteredConsumers) {
+ LinkedHashMap<String, PlcTagItem<PlcSubscriptionTag>> tags,
+ Consumer<PlcSubscriptionEvent> consumer) {
this.subscriber = subscriber;
this.tags = tags;
- this.preRegisteredConsumers = preRegisteredConsumers;
+ this.consumer = consumer;
}
@Override
@@ -70,17 +74,26 @@ public class DefaultPlcSubscriptionRequest implements PlcSubscriptionRequest, Se
@Override
public PlcSubscriptionTag getTag(String name) {
- return tags.get(name);
+ return tags.get(name).getTag();
+ }
+
+ @Override
+ public PlcResponseCode getTagResponseCode(String tagName) {
+ return tags.get(tagName).getResponseCode();
}
@Override
public List<PlcSubscriptionTag> getTags() {
- return new ArrayList<>(tags.values());
+ return tags.values().stream().map(PlcTagItem::getTag).collect(Collectors.toCollection(LinkedList::new));
}
@Override
- public Map<String, List<Consumer<PlcSubscriptionEvent>>> getPreRegisteredConsumers() {
- return new LinkedHashMap<>(preRegisteredConsumers);
+ public Consumer<PlcSubscriptionEvent> getConsumer() {
+ return consumer;
+ }
+
+ public PlcSubscriber getSubscriber() {
+ return subscriber;
}
@Override
@@ -88,10 +101,10 @@ public class DefaultPlcSubscriptionRequest implements PlcSubscriptionRequest, Se
writeBuffer.pushContext("PlcSubscriptionRequest");
writeBuffer.pushContext("tags");
- for (Map.Entry<String, PlcSubscriptionTag> tagEntry : tags.entrySet()) {
+ for (Map.Entry<String, PlcTagItem<PlcSubscriptionTag>> tagEntry : tags.entrySet()) {
String tagName = tagEntry.getKey();
writeBuffer.pushContext(tagName);
- PlcTag tag = tagEntry.getValue();
+ PlcTagItem<PlcSubscriptionTag> tag = tagEntry.getValue();
if (!(tag instanceof Serializable)) {
throw new RuntimeException("Error serializing. Tag doesn't implement XmlSerializable");
}
@@ -108,13 +121,18 @@ public class DefaultPlcSubscriptionRequest implements PlcSubscriptionRequest, Se
private final PlcSubscriber subscriber;
private final PlcTagHandler tagHandler;
private final Map<String, BuilderItem> tags;
- private final LinkedHashMap<String, List<Consumer<PlcSubscriptionEvent>>> preRegisteredConsumers;
+ private Consumer<PlcSubscriptionEvent> consumer;
public Builder(PlcSubscriber subscriber, PlcTagHandler tagHandler) {
this.subscriber = Objects.requireNonNull(subscriber);
this.tagHandler = Objects.requireNonNull(tagHandler);
this.tags = new TreeMap<>();
- this.preRegisteredConsumers = new LinkedHashMap<>();
+ }
+
+ @Override
+ public PlcSubscriptionRequest.Builder setConsumer(Consumer<PlcSubscriptionEvent> consumer) {
+ this.consumer = Objects.requireNonNull(consumer);
+ return this;
}
@Override
@@ -171,27 +189,16 @@ public class DefaultPlcSubscriptionRequest implements PlcSubscriptionRequest, Se
return this;
}
- @Override
- public PlcSubscriptionRequest.Builder addPreRegisteredConsumer(String name, Consumer<PlcSubscriptionEvent> consumer) {
- preRegisteredConsumers.putIfAbsent(name, new LinkedList<>());
- preRegisteredConsumers.get(name).add(consumer);
- return this;
- }
-
@Override
public PlcSubscriptionRequest build() {
- LinkedHashMap<String, PlcSubscriptionTag> parsedTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagItem<PlcSubscriptionTag>> parsedTags = new LinkedHashMap<>();
tags.forEach((name, builderItem) -> {
PlcTag parsedTag = builderItem.tag.get();
- parsedTags.put(name, new DefaultPlcSubscriptionTag(builderItem.plcSubscriptionType, parsedTag, builderItem.duration));
- });
- preRegisteredConsumers.forEach((tagName, ignored) -> {
- if (!tags.containsKey(tagName)) {
- throw new RuntimeException("tagName " + tagName + "for preRegisteredConsumer not found");
- }
+ parsedTags.put(name, new DefaultPlcTagItem<>(new DefaultPlcSubscriptionTag(builderItem.plcSubscriptionType, parsedTag, builderItem.duration)));
});
- return new DefaultPlcSubscriptionRequest(subscriber, parsedTags, preRegisteredConsumers);
+
+ return new DefaultPlcSubscriptionRequest(subscriber, parsedTags, consumer);
}
private static class BuilderItem {
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcSubscriptionResponse.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcSubscriptionResponse.java
index 10595de831..c85ff37cc7 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcSubscriptionResponse.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcSubscriptionResponse.java
@@ -45,13 +45,13 @@ public class DefaultPlcSubscriptionResponse implements PlcSubscriptionResponse,
Map<String, PlcResponseItem<PlcSubscriptionHandle>> values) {
this.request = request;
this.values = values;
- request.getPreRegisteredConsumers().forEach((subscriptionTagName, consumers) -> {
+ /*request.getPreRegisteredConsumers().forEach((subscriptionTagName, consumers) -> {
PlcSubscriptionHandle subscriptionHandle = getSubscriptionHandle(subscriptionTagName);
if (subscriptionHandle == null) {
throw new PlcRuntimeException("PlcSubscriptionHandle for " + subscriptionTagName + " not found");
}
consumers.forEach(subscriptionHandle::register);
- });
+ });*/
}
@Override
@@ -60,8 +60,8 @@ public class DefaultPlcSubscriptionResponse implements PlcSubscriptionResponse,
if (response == null) {
return null;
}
- if (response.getCode() != PlcResponseCode.OK) {
- throw new PlcRuntimeException("Item " + name + " failed to subscribe: " + response.getCode());
+ if (response.getResponseCode() != PlcResponseCode.OK) {
+ throw new PlcRuntimeException("Item " + name + " failed to subscribe: " + response.getResponseCode());
}
return response.getValue();
}
@@ -82,7 +82,7 @@ public class DefaultPlcSubscriptionResponse implements PlcSubscriptionResponse,
if (response == null) {
return null;
}
- return response.getCode();
+ return response.getResponseCode();
}
@Override
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcWriteRequest.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcWriteRequest.java
index d68762c211..d8530196f2 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcWriteRequest.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/DefaultPlcWriteRequest.java
@@ -47,10 +47,10 @@ public class DefaultPlcWriteRequest implements PlcWriteRequest, Serializable {
private final PlcWriter writer;
- private final LinkedHashMap<String, PlcTagValueItem> tags;
+ private final LinkedHashMap<String, PlcTagValueItem<PlcTag>> tags;
public DefaultPlcWriteRequest(PlcWriter writer,
- LinkedHashMap<String, PlcTagValueItem> tags) {
+ LinkedHashMap<String, PlcTagValueItem<PlcTag>> tags) {
this.writer = writer;
this.tags = tags;
}
@@ -71,7 +71,7 @@ public class DefaultPlcWriteRequest implements PlcWriteRequest, Serializable {
return new LinkedHashSet<>(tags.keySet());
}
- public PlcTagValueItem getTagValueItem(String tagName) {
+ public PlcTagValueItem<PlcTag> getTagValueItem(String tagName) {
return tags.get(tagName);
}
@@ -118,10 +118,10 @@ public class DefaultPlcWriteRequest implements PlcWriteRequest, Serializable {
writeBuffer.pushContext("PlcTagRequest");
writeBuffer.pushContext("tags", WithRenderAsList(true));
- for (Map.Entry<String, PlcTagValueItem> tagEntry : tags.entrySet()) {
+ for (Map.Entry<String, PlcTagValueItem<PlcTag>> tagEntry : tags.entrySet()) {
String tagName = tagEntry.getKey();
writeBuffer.pushContext(tagName);
- PlcTagValueItem plcTagValueItem = tagEntry.getValue();
+ PlcTagValueItem<PlcTag> plcTagValueItem = tagEntry.getValue();
if (!(plcTagValueItem instanceof Serializable)) {
throw new RuntimeException("Error serializing. PlcTagValueItem doesn't implement Serializable");
}
@@ -151,7 +151,7 @@ public class DefaultPlcWriteRequest implements PlcWriteRequest, Serializable {
private final PlcWriter writer;
private final PlcTagHandler tagHandler;
private final PlcValueHandler valueHandler;
- private final Map<String, Supplier<PlcTagValueItem>> tagValues;
+ private final Map<String, Supplier<PlcTagValueItem<PlcTag>>> tagValues;
public Builder(PlcWriter writer, PlcTagHandler tagHandler, PlcValueHandler valueHandler) {
this.writer = Objects.requireNonNull(writer);
@@ -170,12 +170,12 @@ public class DefaultPlcWriteRequest implements PlcWriteRequest, Serializable {
PlcTag tag = tagHandler.parseTag(tagAddress);
try {
PlcValue plcValue = parsePlcValue(tag, values);
- return new DefaultPlcTagValueItem(tag, plcValue);
+ return new DefaultPlcTagValueItem<>(tag, plcValue);
} catch (Exception e) {
- return new DefaultPlcTagErrorItem(PlcResponseCode.INVALID_DATA);
+ return new DefaultPlcTagErrorItem<>(PlcResponseCode.INVALID_DATA);
}
} catch (Exception e) {
- return new DefaultPlcTagErrorItem(PlcResponseCode.INVALID_ADDRESS);
+ return new DefaultPlcTagErrorItem<>(PlcResponseCode.INVALID_ADDRESS);
}
});
return this;
@@ -189,9 +189,9 @@ public class DefaultPlcWriteRequest implements PlcWriteRequest, Serializable {
tagValues.put(name, () -> {
try {
PlcValue plcValue = parsePlcValue(tag, values);
- return new DefaultPlcTagValueItem(tag, plcValue);
+ return new DefaultPlcTagValueItem<>(tag, plcValue);
} catch (Exception e) {
- return new DefaultPlcTagErrorItem(PlcResponseCode.INVALID_DATA);
+ return new DefaultPlcTagErrorItem<>(PlcResponseCode.INVALID_DATA);
}
});
return this;
@@ -203,9 +203,9 @@ public class DefaultPlcWriteRequest implements PlcWriteRequest, Serializable {
@Override
public PlcWriteRequest build() {
- LinkedHashMap<String, PlcTagValueItem> parsedTags = new LinkedHashMap<>();
+ LinkedHashMap<String, PlcTagValueItem<PlcTag>> parsedTags = new LinkedHashMap<>();
tagValues.forEach((name, tagValueItemSupplier) -> {
- PlcTagValueItem plcTagValueItem = tagValueItemSupplier.get();
+ PlcTagValueItem<PlcTag> plcTagValueItem = tagValueItemSupplier.get();
parsedTags.put(name, plcTagValueItem);
});
return new DefaultPlcWriteRequest(writer, parsedTags);
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcResponseItem.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcResponseItem.java
index 87e4dec387..cbaaea7786 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcResponseItem.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcResponseItem.java
@@ -37,7 +37,7 @@ public class DefaultPlcResponseItem<T> implements Serializable, PlcResponseItem<
}
@Override
- public PlcResponseCode getCode() {
+ public PlcResponseCode getResponseCode() {
return code;
}
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagErrorItem.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagErrorItem.java
index 211497ec63..de8392a922 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagErrorItem.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagErrorItem.java
@@ -19,12 +19,13 @@
package org.apache.plc4x.java.spi.messages.utils;
+import org.apache.plc4x.java.api.model.PlcTag;
import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.spi.generation.SerializationException;
import org.apache.plc4x.java.spi.generation.WriteBuffer;
import org.apache.plc4x.java.spi.utils.Serializable;
-public class DefaultPlcTagErrorItem implements PlcTagItem, PlcTagValueItem, Serializable {
+public class DefaultPlcTagErrorItem<T extends PlcTag> implements PlcTagItem<T>, PlcTagValueItem<T>, Serializable {
private final PlcResponseCode responseCode;
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagItem.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagItem.java
index f1b15bb4a4..ea94a88f7b 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagItem.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagItem.java
@@ -24,15 +24,15 @@ import org.apache.plc4x.java.spi.generation.SerializationException;
import org.apache.plc4x.java.spi.generation.WriteBuffer;
import org.apache.plc4x.java.spi.utils.Serializable;
-public class DefaultPlcTagItem implements PlcTagItem, Serializable {
+public class DefaultPlcTagItem<T extends PlcTag> implements PlcTagItem<T>, Serializable {
- private final PlcTag tag;
+ private final T tag;
- public DefaultPlcTagItem(PlcTag tag) {
+ public DefaultPlcTagItem(T tag) {
this.tag = tag;
}
- public PlcTag getTag() {
+ public T getTag() {
return tag;
}
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagValueItem.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagValueItem.java
index 315b9c62b3..c9a94f3fe4 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagValueItem.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/DefaultPlcTagValueItem.java
@@ -24,11 +24,11 @@ import org.apache.plc4x.java.spi.generation.SerializationException;
import org.apache.plc4x.java.spi.generation.WriteBuffer;
import org.apache.plc4x.java.spi.utils.Serializable;
-public class DefaultPlcTagValueItem extends DefaultPlcTagItem implements PlcTagValueItem, Serializable {
+public class DefaultPlcTagValueItem<T extends PlcTag> extends DefaultPlcTagItem<T> implements PlcTagValueItem<T>, Serializable {
private final PlcValue value;
- public DefaultPlcTagValueItem(PlcTag tag, PlcValue value) {
+ public DefaultPlcTagValueItem(T tag, PlcValue value) {
super(tag);
this.value = value;
}
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcResponseItem.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcResponseItem.java
index 582ff604fe..5d8756c262 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcResponseItem.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcResponseItem.java
@@ -22,7 +22,9 @@ package org.apache.plc4x.java.spi.messages.utils;
import org.apache.plc4x.java.api.types.PlcResponseCode;
public interface PlcResponseItem<T> {
- PlcResponseCode getCode();
+
+ PlcResponseCode getResponseCode();
T getValue();
+
}
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcTagItem.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcTagItem.java
index 17f6b92ee6..3146fe3658 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcTagItem.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcTagItem.java
@@ -22,13 +22,13 @@ package org.apache.plc4x.java.spi.messages.utils;
import org.apache.plc4x.java.api.model.PlcTag;
import org.apache.plc4x.java.api.types.PlcResponseCode;
-public interface PlcTagItem {
+public interface PlcTagItem<T extends PlcTag> {
default PlcResponseCode getResponseCode() {
return PlcResponseCode.OK;
}
- default PlcTag getTag() {
+ default T getTag() {
return null;
}
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcTagValueItem.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcTagValueItem.java
index fcadabec13..e845377a13 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcTagValueItem.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/messages/utils/PlcTagValueItem.java
@@ -18,9 +18,10 @@
*/
package org.apache.plc4x.java.spi.messages.utils;
+import org.apache.plc4x.java.api.model.PlcTag;
import org.apache.plc4x.java.api.value.PlcValue;
-public interface PlcTagValueItem extends PlcTagItem {
+public interface PlcTagValueItem<T extends PlcTag> extends PlcTagItem<T> {
default PlcValue getValue() {
return null;
diff --git a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/optimizer/SingleTagOptimizer.java b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/optimizer/SingleTagOptimizer.java
index 79b8b93ef7..f9429251f7 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/optimizer/SingleTagOptimizer.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/optimizer/SingleTagOptimizer.java
@@ -49,7 +49,7 @@ public class SingleTagOptimizer extends BaseOptimizer {
PlcReadRequest subRequest = new DefaultPlcReadRequest(
((DefaultPlcReadRequest) readRequest).getReader(),
// We are only expecting valid tagItems being passed in.
- new LinkedHashMap<>(Collections.singletonMap(tagName, new DefaultPlcTagItem(tag))));
+ new LinkedHashMap<>(Collections.singletonMap(tagName, new DefaultPlcTagItem<>(tag))));
subRequests.add(subRequest);
}
return subRequests;
@@ -67,7 +67,7 @@ public class SingleTagOptimizer extends BaseOptimizer {
PlcWriteRequest subRequest = new DefaultPlcWriteRequest(
((DefaultPlcWriteRequest) writeRequest).getWriter(),
// We are only expecting valid tagValueItems being passed in.
- new LinkedHashMap<>(Collections.singletonMap(tagName, new DefaultPlcTagValueItem(tag, value))));
+ new LinkedHashMap<>(Collections.singletonMap(tagName, new DefaultPlcTagValueItem<>(tag, value))));
subRequests.add(subRequest);
}
return subRequests;
diff --git a/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/LeasedPlcConnection.java b/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/LeasedPlcConnection.java
index ddcb70e2f9..d6468c62a3 100644
--- a/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/LeasedPlcConnection.java
+++ b/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/LeasedPlcConnection.java
@@ -337,18 +337,28 @@ public class LeasedPlcConnection implements EventPlcConnection {
return innerPlcSubscriptionRequest.getTag(name);
}
+ @Override
+ public PlcResponseCode getTagResponseCode(String tagName) {
+ return innerPlcSubscriptionRequest.getTagResponseCode(tagName);
+ }
+
@Override
public List<PlcSubscriptionTag> getTags() {
return innerPlcSubscriptionRequest.getTags();
}
@Override
- public Map<String, List<Consumer<PlcSubscriptionEvent>>> getPreRegisteredConsumers() {
- return innerPlcSubscriptionRequest.getPreRegisteredConsumers();
+ public Consumer<PlcSubscriptionEvent> getConsumer() {
+ return innerPlcSubscriptionRequest.getConsumer();
}
};
}
+ @Override
+ public PlcSubscriptionRequest.Builder setConsumer(Consumer<PlcSubscriptionEvent> consumer) {
+ return innerBuilder.setConsumer(consumer);
+ }
+
@Override
public PlcSubscriptionRequest.Builder addCyclicTagAddress(String name, String tagAddress, Duration pollingInterval) {
return innerBuilder.addCyclicTagAddress(name, tagAddress, pollingInterval);
@@ -378,11 +388,6 @@ public class LeasedPlcConnection implements EventPlcConnection {
public PlcSubscriptionRequest.Builder addEventTag(String name, PlcTag tag) {
return innerBuilder.addEventTag(name, tag);
}
-
- @Override
- public PlcSubscriptionRequest.Builder addPreRegisteredConsumer(String name, Consumer<PlcSubscriptionEvent> preRegisteredConsumer) {
- return innerBuilder.addPreRegisteredConsumer(name, preRegisteredConsumer);
- }
};
}