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 7f851ab2592082850b2467db330fa54a7a4cb9a6 Author: Christofer Dutz <[email protected]> AuthorDate: Mon Jul 27 10:48:04 2026 +0200 feat: Updated all drivers that currently didn't support subscriptions natively to use the new subscription emulation. --- plc4j/drivers/ab-eth/pom.xml | 5 ++ .../apache/plc4x/java/abeth/AbEthConnection.java | 4 +- .../abeth/manual/ManualAbEthSubscriptionTest.java | 89 +++++++++++++++++++++ plc4j/drivers/eip/pom.xml | 5 ++ .../plc4x/java/eip/base/EipTcpConnection.java | 4 +- .../eip/base/manual/ManualEipSubscriptionTest.java | 80 +++++++++++++++++++ plc4j/drivers/modbus/pom.xml | 5 ++ .../java/modbus/ascii/ModbusAsciiConnection.java | 4 +- .../plc4x/java/modbus/rtu/ModbusRtuConnection.java | 4 +- .../plc4x/java/modbus/tcp/ModbusTcpConnection.java | 4 +- .../manual/ManualModbusASCIISubscriptionTest.java | 81 +++++++++++++++++++ .../manual/ManualModbusRTUSubscriptionTest.java | 81 +++++++++++++++++++ .../manual/ManualModbusTCPSubscriptionTest.java | 88 +++++++++++++++++++++ plc4j/drivers/slmp/pom.xml | 5 ++ .../org/apache/plc4x/java/slmp/SlmpConnection.java | 4 +- .../slmp/manual/ManualSlmpSubscriptionTest.java | 89 +++++++++++++++++++++ plc4j/drivers/umas/pom.xml | 5 ++ .../org/apache/plc4x/java/umas/UmasConnection.java | 4 +- .../umas/manual/ManualUmasSubscriptionTest.java | 90 ++++++++++++++++++++++ 19 files changed, 637 insertions(+), 14 deletions(-) diff --git a/plc4j/drivers/ab-eth/pom.xml b/plc4j/drivers/ab-eth/pom.xml index f5e0a7c766..e889fbd85c 100644 --- a/plc4j/drivers/ab-eth/pom.xml +++ b/plc4j/drivers/ab-eth/pom.xml @@ -157,6 +157,11 @@ <artifactId>plc4j-spi-drivers</artifactId> <version>1.0.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.plc4x</groupId> + <artifactId>plc4j-utils-subscription-emulation</artifactId> + <version>1.0.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.plc4x</groupId> <artifactId>plc4j-transports-api</artifactId> diff --git a/plc4j/drivers/ab-eth/src/main/java/org/apache/plc4x/java/abeth/AbEthConnection.java b/plc4j/drivers/ab-eth/src/main/java/org/apache/plc4x/java/abeth/AbEthConnection.java index e8c7e115a4..07a531797f 100644 --- a/plc4j/drivers/ab-eth/src/main/java/org/apache/plc4x/java/abeth/AbEthConnection.java +++ b/plc4j/drivers/ab-eth/src/main/java/org/apache/plc4x/java/abeth/AbEthConnection.java @@ -37,7 +37,7 @@ import org.apache.plc4x.java.api.messages.PlcReadResponse; import org.apache.plc4x.java.api.types.ConnectionStateChangeType; import org.apache.plc4x.java.api.types.PlcResponseCode; import org.apache.plc4x.java.api.value.PlcValue; -import org.apache.plc4x.java.spi.drivers.ConnectionBase; +import org.apache.plc4x.java.utils.subscriptionemulation.PollingSubscriptionConnectionBase; import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException; import org.apache.plc4x.java.spi.drivers.messages.DefaultPlcReadResponse; import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem; @@ -72,7 +72,7 @@ import java.util.concurrent.atomic.AtomicInteger; * handle; every subsequent read carries that handle and a fresh * 16-bit transaction counter used to match responses to in-flight reads.</p> */ -public class AbEthConnection extends ConnectionBase<AbEthConfiguration> { +public class AbEthConnection extends PollingSubscriptionConnectionBase<AbEthConfiguration> { private static final Logger LOGGER = LoggerFactory.getLogger(AbEthConnection.class); diff --git a/plc4j/drivers/ab-eth/src/test/java/org/apache/plc4x/java/abeth/manual/ManualAbEthSubscriptionTest.java b/plc4j/drivers/ab-eth/src/test/java/org/apache/plc4x/java/abeth/manual/ManualAbEthSubscriptionTest.java new file mode 100644 index 0000000000..d17c437802 --- /dev/null +++ b/plc4j/drivers/ab-eth/src/test/java/org/apache/plc4x/java/abeth/manual/ManualAbEthSubscriptionTest.java @@ -0,0 +1,89 @@ +/* + * 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.abeth.manual; + +import org.apache.plc4x.java.api.PlcConnection; +import org.apache.plc4x.java.api.PlcDriverManager; +import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; +import org.apache.plc4x.java.api.model.PlcSubscriptionHandle; +import org.apache.plc4x.java.api.types.PlcResponseCode; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** + * Manual smoke-test for the Allen-Bradley (AB-ETH / DF1-over-Ethernet) driver's <em>emulated</em> + * subscription path. + * <p> + * AB-ETH has no native subscription mechanism; since {@code AbEthConnection} now extends + * {@code PollingSubscriptionConnectionBase}, subscriptions are emulated by polling the tag via the + * regular read path. This test registers two subscriptions against {@code N7:0:WORD}: + * <ul> + * <li>a CYCLIC subscription that fires every second, and</li> + * <li>a CHANGE_OF_STATE subscription that only fires when the value actually changes.</li> + * </ul> + * It listens for 30 seconds, then unsubscribes cleanly. + * <p> + * To produce visible value changes, change {@code N7:0} on the PLC while this is running. + */ +public class ManualAbEthSubscriptionTest { + + private static final String CONNECTION_URL = "ab-eth://192.168.24.40"; + private static final String TAG_ADDRESS = "N7:0:WORD"; + private static final long LISTEN_DURATION_MS = 30_000L; + + public static void main(String[] args) throws Exception { + try (PlcConnection connection = PlcDriverManager.getDefault().getConnectionManager().getConnection(CONNECTION_URL)) { + System.out.printf("Connected. metadata.subscribeSupported=%s%n", + connection.getMetadata().isSubscribeSupported()); + + PlcSubscriptionResponse subResp = connection.subscriptionRequestBuilder() + .addCyclicTagAddress("cyclic-tag", TAG_ADDRESS, Duration.ofSeconds(1)) + .addChangeOfStateTagAddress("change-tag", TAG_ADDRESS) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + PlcResponseCode cyclicCode = subResp.getResponseCode("cyclic-tag"); + PlcResponseCode changeCode = subResp.getResponseCode("change-tag"); + System.out.printf("Subscribe response: cyclic=%s, change-of-state=%s%n", cyclicCode, changeCode); + if (cyclicCode != PlcResponseCode.OK || changeCode != PlcResponseCode.OK) { + System.err.println("Subscription failed — aborting"); + return; + } + + PlcSubscriptionHandle cyclicHandle = subResp.getSubscriptionHandle("cyclic-tag"); + PlcSubscriptionHandle changeHandle = subResp.getSubscriptionHandle("change-tag"); + cyclicHandle.register(event -> System.out.printf("[%s] CYCLIC cyclic-tag = %s%n", + event.getTimestamp(), event.getPlcValue("cyclic-tag"))); + changeHandle.register(event -> System.out.printf("[%s] CHANGE_OF_STATE change-tag = %s%n", + event.getTimestamp(), event.getPlcValue("change-tag"))); + + System.out.printf("Listening for emulated pushes for %ds...%n", LISTEN_DURATION_MS / 1000); + Thread.sleep(LISTEN_DURATION_MS); + + connection.unsubscriptionRequestBuilder() + .addHandles(cyclicHandle, changeHandle) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + System.out.println("Unsubscribed."); + } + } +} diff --git a/plc4j/drivers/eip/pom.xml b/plc4j/drivers/eip/pom.xml index 8e24fdc941..c3003b76bb 100644 --- a/plc4j/drivers/eip/pom.xml +++ b/plc4j/drivers/eip/pom.xml @@ -228,6 +228,11 @@ <artifactId>plc4j-spi-drivers</artifactId> <version>1.0.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.plc4x</groupId> + <artifactId>plc4j-utils-subscription-emulation</artifactId> + <version>1.0.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.plc4x</groupId> <artifactId>plc4j-transports-api</artifactId> diff --git a/plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/EipTcpConnection.java b/plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/EipTcpConnection.java index c3f48d9fb0..a2a920bbbd 100644 --- a/plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/EipTcpConnection.java +++ b/plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/EipTcpConnection.java @@ -34,7 +34,7 @@ import org.apache.plc4x.java.spi.buffers.api.exceptions.BufferException; import org.apache.plc4x.java.spi.buffers.bytebased.ReadBufferByteBased; import org.apache.plc4x.java.spi.buffers.bytebased.WithByteBasedOption; import org.apache.plc4x.java.spi.buffers.bytebased.WriteBufferByteBased; -import org.apache.plc4x.java.spi.drivers.ConnectionBase; +import org.apache.plc4x.java.utils.subscriptionemulation.PollingSubscriptionConnectionBase; import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException; import org.apache.plc4x.java.spi.drivers.messages.*; import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem; @@ -60,7 +60,7 @@ import java.util.regex.Pattern; * EipProtocolLogic to the SPI3 ConnectionBase model. Request/response correlation * uses the 8-byte senderContext echoed back by the device. */ -public class EipTcpConnection extends ConnectionBase<EIPConfiguration> { +public class EipTcpConnection extends PollingSubscriptionConnectionBase<EIPConfiguration> { private static final Logger LOGGER = LoggerFactory.getLogger(EipTcpConnection.class); diff --git a/plc4j/drivers/eip/src/test/java/org/apache/plc4x/java/eip/base/manual/ManualEipSubscriptionTest.java b/plc4j/drivers/eip/src/test/java/org/apache/plc4x/java/eip/base/manual/ManualEipSubscriptionTest.java new file mode 100644 index 0000000000..b3ce9f9a25 --- /dev/null +++ b/plc4j/drivers/eip/src/test/java/org/apache/plc4x/java/eip/base/manual/ManualEipSubscriptionTest.java @@ -0,0 +1,80 @@ +/* + * 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.eip.base.manual; + +import org.apache.plc4x.java.api.PlcConnection; +import org.apache.plc4x.java.api.PlcDriverManager; +import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; +import org.apache.plc4x.java.api.model.PlcSubscriptionHandle; +import org.apache.plc4x.java.api.types.PlcResponseCode; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** + * Manual smoke-test for the EtherNet/IP driver's <em>emulated</em> subscription path. + * <p> + * EtherNet/IP has no native subscription mechanism; since {@code EipTcpConnection} now extends + * {@code PollingSubscriptionConnectionBase}, subscriptions are emulated by polling the tag via the + * regular read path. This test asks for the value of {@code %out01} once per second for 30 seconds + * using a CYCLIC subscription, then unsubscribes cleanly. + * <p> + * To produce visible value changes, change the tag on the PLC while this is running. + */ +public class ManualEipSubscriptionTest { + + private static final String CONNECTION_URL = "eip://192.168.24.32"; + private static final String TAG_ADDRESS = "%out01"; + private static final long LISTEN_DURATION_MS = 30_000L; + + public static void main(String[] args) throws Exception { + try (PlcConnection connection = PlcDriverManager.getDefault().getConnectionManager().getConnection(CONNECTION_URL)) { + System.out.printf("Connected. metadata.subscribeSupported=%s%n", + connection.getMetadata().isSubscribeSupported()); + + PlcSubscriptionResponse subResp = connection.subscriptionRequestBuilder() + .addCyclicTagAddress("subscription-tag", TAG_ADDRESS, Duration.ofSeconds(1)) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + PlcResponseCode code = subResp.getResponseCode("subscription-tag"); + System.out.printf("Subscribe response: %s%n", code); + if (code != PlcResponseCode.OK) { + System.err.println("Subscription failed — aborting"); + return; + } + + PlcSubscriptionHandle handle = subResp.getSubscriptionHandle("subscription-tag"); + handle.register(event -> System.out.printf("[%s] subscription-tag = %s%n", + event.getTimestamp(), event.getPlcValue("subscription-tag"))); + + System.out.printf("Listening for emulated cyclic pushes for %ds...%n", LISTEN_DURATION_MS / 1000); + Thread.sleep(LISTEN_DURATION_MS); + + connection.unsubscriptionRequestBuilder() + .addHandles(handle) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + System.out.println("Unsubscribed."); + } + } +} diff --git a/plc4j/drivers/modbus/pom.xml b/plc4j/drivers/modbus/pom.xml index 3878c2efcb..5e47da183e 100644 --- a/plc4j/drivers/modbus/pom.xml +++ b/plc4j/drivers/modbus/pom.xml @@ -182,6 +182,11 @@ <artifactId>plc4j-spi-drivers</artifactId> <version>1.0.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.plc4x</groupId> + <artifactId>plc4j-utils-subscription-emulation</artifactId> + <version>1.0.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.plc4x</groupId> <artifactId>plc4j-transports-api</artifactId> diff --git a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java index c97e2f641a..41d200abd5 100644 --- a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java +++ b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java @@ -34,7 +34,7 @@ import org.apache.plc4x.java.spi.buffers.api.exceptions.BufferException; import org.apache.plc4x.java.spi.buffers.bytebased.ReadBufferByteBased; import org.apache.plc4x.java.spi.buffers.bytebased.WithByteBasedOption; import org.apache.plc4x.java.spi.buffers.bytebased.WriteBufferByteBased; -import org.apache.plc4x.java.spi.drivers.ConnectionBase; +import org.apache.plc4x.java.utils.subscriptionemulation.PollingSubscriptionConnectionBase; import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException; import org.apache.plc4x.java.spi.drivers.messages.*; import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem; @@ -59,7 +59,7 @@ import java.util.concurrent.*; * Modbus ASCII connection implementation. * Handles read, write, and ping operations over Modbus ASCII protocol. */ -public class ModbusAsciiConnection extends ConnectionBase<ModbusAsciiConfiguration> { +public class ModbusAsciiConnection extends PollingSubscriptionConnectionBase<ModbusAsciiConfiguration> { private static final Logger LOGGER = LoggerFactory.getLogger(ModbusAsciiConnection.class); diff --git a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java index 4a9e7b789a..9f4bf9df13 100644 --- a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java +++ b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java @@ -34,7 +34,7 @@ import org.apache.plc4x.java.spi.buffers.api.exceptions.BufferException; import org.apache.plc4x.java.spi.buffers.bytebased.ReadBufferByteBased; import org.apache.plc4x.java.spi.buffers.bytebased.WithByteBasedOption; import org.apache.plc4x.java.spi.buffers.bytebased.WriteBufferByteBased; -import org.apache.plc4x.java.spi.drivers.ConnectionBase; +import org.apache.plc4x.java.utils.subscriptionemulation.PollingSubscriptionConnectionBase; import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException; import org.apache.plc4x.java.spi.drivers.messages.*; import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem; @@ -59,7 +59,7 @@ import java.util.concurrent.*; * Modbus RTU connection implementation. * Handles read, write, and ping operations over Modbus RTU protocol. */ -public class ModbusRtuConnection extends ConnectionBase<ModbusRtuConfiguration> { +public class ModbusRtuConnection extends PollingSubscriptionConnectionBase<ModbusRtuConfiguration> { private static final Logger LOGGER = LoggerFactory.getLogger(ModbusRtuConnection.class); diff --git a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/ModbusTcpConnection.java b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/ModbusTcpConnection.java index 9801f018e7..01171b93ad 100644 --- a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/ModbusTcpConnection.java +++ b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/ModbusTcpConnection.java @@ -35,7 +35,7 @@ import org.apache.plc4x.java.spi.buffers.api.exceptions.BufferException; import org.apache.plc4x.java.spi.buffers.bytebased.ReadBufferByteBased; import org.apache.plc4x.java.spi.buffers.bytebased.WithByteBasedOption; import org.apache.plc4x.java.spi.buffers.bytebased.WriteBufferByteBased; -import org.apache.plc4x.java.spi.drivers.ConnectionBase; +import org.apache.plc4x.java.utils.subscriptionemulation.PollingSubscriptionConnectionBase; import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException; import org.apache.plc4x.java.spi.drivers.messages.*; import org.apache.plc4x.java.spi.drivers.messages.items.DefaultPlcResponseItem; @@ -60,7 +60,7 @@ import java.util.concurrent.atomic.AtomicInteger; * Modbus TCP connection implementation. * Handles read, write, and ping operations over Modbus TCP protocol. */ -public class ModbusTcpConnection extends ConnectionBase<ModbusTcpConfiguration> { +public class ModbusTcpConnection extends PollingSubscriptionConnectionBase<ModbusTcpConfiguration> { private static final Logger LOGGER = LoggerFactory.getLogger(ModbusTcpConnection.class); diff --git a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/ascii/manual/ManualModbusASCIISubscriptionTest.java b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/ascii/manual/ManualModbusASCIISubscriptionTest.java new file mode 100644 index 0000000000..34a320333f --- /dev/null +++ b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/ascii/manual/ManualModbusASCIISubscriptionTest.java @@ -0,0 +1,81 @@ +/* + * 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.modbus.ascii.manual; + +import org.apache.plc4x.java.api.PlcConnection; +import org.apache.plc4x.java.api.PlcDriverManager; +import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; +import org.apache.plc4x.java.api.model.PlcSubscriptionHandle; +import org.apache.plc4x.java.api.types.PlcResponseCode; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** + * Manual smoke-test for the Modbus/ASCII driver's <em>emulated</em> subscription path. + * <p> + * Modbus has no native subscription mechanism; since {@code ModbusAsciiConnection} now extends + * {@code PollingSubscriptionConnectionBase}, subscriptions are emulated by polling the tag via the + * regular read path. This test asks for the value of {@code 400001:BOOL} once per second for + * 30 seconds using a CYCLIC subscription, then unsubscribes cleanly. + * <p> + * Requires a serial adapter and an attached ASCII device; adjust the connection URL / serial + * parameters to match your setup. + */ +public class ManualModbusASCIISubscriptionTest { + + private static final String CONNECTION_URL = + "modbus-ascii:///dev/tty.usbserial-AR0K3WCE?serial.baud-rate=9600&serial.num-data-bits=8&num-stop-bits=1&serial.parity=EVEN_PARITY"; + private static final String TAG_ADDRESS = "400001:BOOL"; + private static final long LISTEN_DURATION_MS = 30_000L; + + public static void main(String[] args) throws Exception { + try (PlcConnection connection = PlcDriverManager.getDefault().getConnectionManager().getConnection(CONNECTION_URL)) { + System.out.printf("Connected. metadata.subscribeSupported=%s%n", + connection.getMetadata().isSubscribeSupported()); + + PlcSubscriptionResponse subResp = connection.subscriptionRequestBuilder() + .addCyclicTagAddress("subscription-tag", TAG_ADDRESS, Duration.ofSeconds(1)) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + PlcResponseCode code = subResp.getResponseCode("subscription-tag"); + System.out.printf("Subscribe response: %s%n", code); + if (code != PlcResponseCode.OK) { + System.err.println("Subscription failed — aborting"); + return; + } + + PlcSubscriptionHandle handle = subResp.getSubscriptionHandle("subscription-tag"); + handle.register(event -> System.out.printf("[%s] subscription-tag = %s%n", + event.getTimestamp(), event.getPlcValue("subscription-tag"))); + + System.out.printf("Listening for emulated cyclic pushes for %ds...%n", LISTEN_DURATION_MS / 1000); + Thread.sleep(LISTEN_DURATION_MS); + + connection.unsubscriptionRequestBuilder() + .addHandles(handle) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + System.out.println("Unsubscribed."); + } + } +} diff --git a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/manual/ManualModbusRTUSubscriptionTest.java b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/manual/ManualModbusRTUSubscriptionTest.java new file mode 100644 index 0000000000..f212a67f0c --- /dev/null +++ b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/manual/ManualModbusRTUSubscriptionTest.java @@ -0,0 +1,81 @@ +/* + * 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.modbus.rtu.manual; + +import org.apache.plc4x.java.api.PlcConnection; +import org.apache.plc4x.java.api.PlcDriverManager; +import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; +import org.apache.plc4x.java.api.model.PlcSubscriptionHandle; +import org.apache.plc4x.java.api.types.PlcResponseCode; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** + * Manual smoke-test for the Modbus/RTU driver's <em>emulated</em> subscription path. + * <p> + * Modbus has no native subscription mechanism; since {@code ModbusRtuConnection} now extends + * {@code PollingSubscriptionConnectionBase}, subscriptions are emulated by polling the tag via the + * regular read path. This test asks for the value of {@code 400001:BOOL} once per second for + * 30 seconds using a CYCLIC subscription, then unsubscribes cleanly. + * <p> + * Requires a serial adapter and an attached RTU device; adjust the connection URL / serial + * parameters to match your setup. + */ +public class ManualModbusRTUSubscriptionTest { + + private static final String CONNECTION_URL = + "modbus-rtu:///dev/tty.usbserial-AR0K3WCE?serial.baud-rate=9600&serial.num-data-bits=8&num-stop-bits=1&serial.parity=EVEN_PARITY"; + private static final String TAG_ADDRESS = "400001:BOOL"; + private static final long LISTEN_DURATION_MS = 30_000L; + + public static void main(String[] args) throws Exception { + try (PlcConnection connection = PlcDriverManager.getDefault().getConnectionManager().getConnection(CONNECTION_URL)) { + System.out.printf("Connected. metadata.subscribeSupported=%s%n", + connection.getMetadata().isSubscribeSupported()); + + PlcSubscriptionResponse subResp = connection.subscriptionRequestBuilder() + .addCyclicTagAddress("subscription-tag", TAG_ADDRESS, Duration.ofSeconds(1)) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + PlcResponseCode code = subResp.getResponseCode("subscription-tag"); + System.out.printf("Subscribe response: %s%n", code); + if (code != PlcResponseCode.OK) { + System.err.println("Subscription failed — aborting"); + return; + } + + PlcSubscriptionHandle handle = subResp.getSubscriptionHandle("subscription-tag"); + handle.register(event -> System.out.printf("[%s] subscription-tag = %s%n", + event.getTimestamp(), event.getPlcValue("subscription-tag"))); + + System.out.printf("Listening for emulated cyclic pushes for %ds...%n", LISTEN_DURATION_MS / 1000); + Thread.sleep(LISTEN_DURATION_MS); + + connection.unsubscriptionRequestBuilder() + .addHandles(handle) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + System.out.println("Unsubscribed."); + } + } +} diff --git a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/tcp/manual/ManualModbusTCPSubscriptionTest.java b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/tcp/manual/ManualModbusTCPSubscriptionTest.java new file mode 100644 index 0000000000..4ad3a41c63 --- /dev/null +++ b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/tcp/manual/ManualModbusTCPSubscriptionTest.java @@ -0,0 +1,88 @@ +/* + * 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.modbus.tcp.manual; + +import org.apache.plc4x.java.api.PlcConnection; +import org.apache.plc4x.java.api.PlcDriverManager; +import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; +import org.apache.plc4x.java.api.model.PlcSubscriptionHandle; +import org.apache.plc4x.java.api.types.PlcResponseCode; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** + * Manual smoke-test for the Modbus/TCP driver's <em>emulated</em> subscription path. + * <p> + * Modbus has no native subscription mechanism; since {@code ModbusTcpConnection} now extends + * {@code PollingSubscriptionConnectionBase}, subscriptions are emulated by polling the tag via the + * regular read path. This test registers two subscriptions against {@code holding-register:3:WORD}: + * <ul> + * <li>a CYCLIC subscription that fires every second, and</li> + * <li>a CHANGE_OF_STATE subscription that only fires when the value actually changes.</li> + * </ul> + * It listens for 30 seconds, then unsubscribes cleanly. + * <p> + * To produce visible value changes, write to holding register 3 on the PLC while this is running. + */ +public class ManualModbusTCPSubscriptionTest { + + private static final String CONNECTION_URL = "modbus-tcp://192.168.23.30"; + private static final String TAG_ADDRESS = "holding-register:3:WORD"; + private static final long LISTEN_DURATION_MS = 30_000L; + + public static void main(String[] args) throws Exception { + try (PlcConnection connection = PlcDriverManager.getDefault().getConnectionManager().getConnection(CONNECTION_URL)) { + System.out.printf("Connected. metadata.subscribeSupported=%s%n", + connection.getMetadata().isSubscribeSupported()); + + PlcSubscriptionResponse subResp = connection.subscriptionRequestBuilder() + .addCyclicTagAddress("cyclic-tag", TAG_ADDRESS, Duration.ofSeconds(1)) + .addChangeOfStateTagAddress("change-tag", TAG_ADDRESS) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + PlcResponseCode cyclicCode = subResp.getResponseCode("cyclic-tag"); + PlcResponseCode changeCode = subResp.getResponseCode("change-tag"); + System.out.printf("Subscribe response: cyclic=%s, change-of-state=%s%n", cyclicCode, changeCode); + if (cyclicCode != PlcResponseCode.OK || changeCode != PlcResponseCode.OK) { + System.err.println("Subscription failed — aborting"); + return; + } + + PlcSubscriptionHandle cyclicHandle = subResp.getSubscriptionHandle("cyclic-tag"); + PlcSubscriptionHandle changeHandle = subResp.getSubscriptionHandle("change-tag"); + cyclicHandle.register(event -> System.out.printf("[%s] CYCLIC cyclic-tag = %s%n", + event.getTimestamp(), event.getPlcValue("cyclic-tag"))); + changeHandle.register(event -> System.out.printf("[%s] CHANGE_OF_STATE change-tag = %s%n", + event.getTimestamp(), event.getPlcValue("change-tag"))); + + System.out.printf("Listening for emulated pushes for %ds...%n", LISTEN_DURATION_MS / 1000); + Thread.sleep(LISTEN_DURATION_MS); + + connection.unsubscriptionRequestBuilder() + .addHandles(cyclicHandle, changeHandle) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + System.out.println("Unsubscribed."); + } + } +} diff --git a/plc4j/drivers/slmp/pom.xml b/plc4j/drivers/slmp/pom.xml index d07abc4f3b..5e66bf544f 100644 --- a/plc4j/drivers/slmp/pom.xml +++ b/plc4j/drivers/slmp/pom.xml @@ -181,6 +181,11 @@ <artifactId>plc4j-spi-drivers</artifactId> <version>1.0.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.plc4x</groupId> + <artifactId>plc4j-utils-subscription-emulation</artifactId> + <version>1.0.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.plc4x</groupId> <artifactId>plc4j-transports-api</artifactId> diff --git a/plc4j/drivers/slmp/src/main/java/org/apache/plc4x/java/slmp/SlmpConnection.java b/plc4j/drivers/slmp/src/main/java/org/apache/plc4x/java/slmp/SlmpConnection.java index ef0a65b298..65df346fa3 100644 --- a/plc4j/drivers/slmp/src/main/java/org/apache/plc4x/java/slmp/SlmpConnection.java +++ b/plc4j/drivers/slmp/src/main/java/org/apache/plc4x/java/slmp/SlmpConnection.java @@ -35,7 +35,7 @@ import org.apache.plc4x.java.slmp.readwrite.SlmpResponseFrame3E; import org.apache.plc4x.java.slmp.readwrite.SlmpWriteRequest; import org.apache.plc4x.java.slmp.tag.SlmpTag; import org.apache.plc4x.java.slmp.tag.SlmpTagHandler; -import org.apache.plc4x.java.spi.drivers.ConnectionBase; +import org.apache.plc4x.java.utils.subscriptionemulation.PollingSubscriptionConnectionBase; import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException; import org.apache.plc4x.java.spi.drivers.messages.DefaultPlcReadRequest; import org.apache.plc4x.java.spi.drivers.messages.DefaultPlcReadResponse; @@ -59,7 +59,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; -public class SlmpConnection extends ConnectionBase<SlmpConfiguration> { +public class SlmpConnection extends PollingSubscriptionConnectionBase<SlmpConfiguration> { private static final Logger LOGGER = LoggerFactory.getLogger(SlmpConnection.class); diff --git a/plc4j/drivers/slmp/src/test/java/org/apache/plc4x/java/slmp/manual/ManualSlmpSubscriptionTest.java b/plc4j/drivers/slmp/src/test/java/org/apache/plc4x/java/slmp/manual/ManualSlmpSubscriptionTest.java new file mode 100644 index 0000000000..8541ee51cd --- /dev/null +++ b/plc4j/drivers/slmp/src/test/java/org/apache/plc4x/java/slmp/manual/ManualSlmpSubscriptionTest.java @@ -0,0 +1,89 @@ +/* + * 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. + */ +package org.apache.plc4x.java.slmp.manual; + +import org.apache.plc4x.java.api.PlcConnection; +import org.apache.plc4x.java.api.PlcDriverManager; +import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; +import org.apache.plc4x.java.api.model.PlcSubscriptionHandle; +import org.apache.plc4x.java.api.types.PlcResponseCode; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** + * Manual smoke-test for the SLMP / MELSEC Communication 3E driver's <em>emulated</em> + * subscription path. + * <p> + * SLMP has no native subscription mechanism; since {@code SlmpConnection} now extends + * {@code PollingSubscriptionConnectionBase}, subscriptions are emulated by polling the tag via the + * regular read path. This test registers two subscriptions against data register {@code D350:WORD}: + * <ul> + * <li>a CYCLIC subscription that fires every second, and</li> + * <li>a CHANGE_OF_STATE subscription that only fires when the value actually changes.</li> + * </ul> + * It listens for 30 seconds, then unsubscribes cleanly. + * <p> + * To produce visible value changes, change data register {@code D350} on the PLC while this is running. + */ +public class ManualSlmpSubscriptionTest { + + private static final String CONNECTION_URL = "slmp://192.168.24.41"; + private static final String TAG_ADDRESS = "D350:WORD"; + private static final long LISTEN_DURATION_MS = 30_000L; + + public static void main(String[] args) throws Exception { + try (PlcConnection connection = PlcDriverManager.getDefault().getConnectionManager().getConnection(CONNECTION_URL)) { + System.out.printf("Connected. metadata.subscribeSupported=%s%n", + connection.getMetadata().isSubscribeSupported()); + + PlcSubscriptionResponse subResp = connection.subscriptionRequestBuilder() + .addCyclicTagAddress("cyclic-tag", TAG_ADDRESS, Duration.ofSeconds(1)) + .addChangeOfStateTagAddress("change-tag", TAG_ADDRESS) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + PlcResponseCode cyclicCode = subResp.getResponseCode("cyclic-tag"); + PlcResponseCode changeCode = subResp.getResponseCode("change-tag"); + System.out.printf("Subscribe response: cyclic=%s, change-of-state=%s%n", cyclicCode, changeCode); + if (cyclicCode != PlcResponseCode.OK || changeCode != PlcResponseCode.OK) { + System.err.println("Subscription failed — aborting"); + return; + } + + PlcSubscriptionHandle cyclicHandle = subResp.getSubscriptionHandle("cyclic-tag"); + PlcSubscriptionHandle changeHandle = subResp.getSubscriptionHandle("change-tag"); + cyclicHandle.register(event -> System.out.printf("[%s] CYCLIC cyclic-tag = %s%n", + event.getTimestamp(), event.getPlcValue("cyclic-tag"))); + changeHandle.register(event -> System.out.printf("[%s] CHANGE_OF_STATE change-tag = %s%n", + event.getTimestamp(), event.getPlcValue("change-tag"))); + + System.out.printf("Listening for emulated pushes for %ds...%n", LISTEN_DURATION_MS / 1000); + Thread.sleep(LISTEN_DURATION_MS); + + connection.unsubscriptionRequestBuilder() + .addHandles(cyclicHandle, changeHandle) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + System.out.println("Unsubscribed."); + } + } +} diff --git a/plc4j/drivers/umas/pom.xml b/plc4j/drivers/umas/pom.xml index 9314ea43ac..248a2c1be7 100644 --- a/plc4j/drivers/umas/pom.xml +++ b/plc4j/drivers/umas/pom.xml @@ -157,6 +157,11 @@ <artifactId>plc4j-spi-drivers</artifactId> <version>1.0.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.apache.plc4x</groupId> + <artifactId>plc4j-utils-subscription-emulation</artifactId> + <version>1.0.0-SNAPSHOT</version> + </dependency> <dependency> <groupId>org.apache.plc4x</groupId> <artifactId>plc4j-transports-api</artifactId> diff --git a/plc4j/drivers/umas/src/main/java/org/apache/plc4x/java/umas/UmasConnection.java b/plc4j/drivers/umas/src/main/java/org/apache/plc4x/java/umas/UmasConnection.java index 4cd972d6c0..029f4ed517 100644 --- a/plc4j/drivers/umas/src/main/java/org/apache/plc4x/java/umas/UmasConnection.java +++ b/plc4j/drivers/umas/src/main/java/org/apache/plc4x/java/umas/UmasConnection.java @@ -42,7 +42,7 @@ import org.apache.plc4x.java.spi.buffers.api.WithOption; import org.apache.plc4x.java.spi.buffers.api.exceptions.BufferException; import org.apache.plc4x.java.spi.buffers.bytebased.ReadBufferByteBased; import org.apache.plc4x.java.spi.buffers.bytebased.WithByteBasedOption; -import org.apache.plc4x.java.spi.drivers.ConnectionBase; +import org.apache.plc4x.java.utils.subscriptionemulation.PollingSubscriptionConnectionBase; import org.apache.plc4x.java.spi.drivers.exceptions.MessageCodecException; import org.apache.plc4x.java.spi.drivers.messages.DefaultPlcBrowseItem; import org.apache.plc4x.java.spi.drivers.messages.DefaultPlcBrowseResponse; @@ -135,7 +135,7 @@ import java.util.concurrent.atomic.AtomicInteger; * UMAS PDU is recorded in {@link UmasFunctionKeyTracker} so the codec can * resolve the response subtype (UMAS responses share function key 0xFE).</p> */ -public class UmasConnection extends ConnectionBase<UmasConfiguration> { +public class UmasConnection extends PollingSubscriptionConnectionBase<UmasConfiguration> { private static final Logger LOGGER = LoggerFactory.getLogger(UmasConnection.class); diff --git a/plc4j/drivers/umas/src/test/java/org/apache/plc4x/java/umas/manual/ManualUmasSubscriptionTest.java b/plc4j/drivers/umas/src/test/java/org/apache/plc4x/java/umas/manual/ManualUmasSubscriptionTest.java new file mode 100644 index 0000000000..3f2e8abc73 --- /dev/null +++ b/plc4j/drivers/umas/src/test/java/org/apache/plc4x/java/umas/manual/ManualUmasSubscriptionTest.java @@ -0,0 +1,90 @@ +/* + * 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. + */ +package org.apache.plc4x.java.umas.manual; + +import org.apache.plc4x.java.DefaultPlcDriverManager; +import org.apache.plc4x.java.api.PlcConnection; +import org.apache.plc4x.java.api.authentication.PlcNullAuthentication; +import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse; +import org.apache.plc4x.java.api.model.PlcSubscriptionHandle; +import org.apache.plc4x.java.api.types.PlcResponseCode; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; + +/** + * Manual smoke-test for the UMAS driver's <em>emulated</em> subscription path. + * <p> + * UMAS has no native subscription mechanism; since {@code UmasConnection} now extends + * {@code PollingSubscriptionConnectionBase}, subscriptions are emulated by polling the tag via the + * regular read path. This test registers two subscriptions against {@code g_b16}: + * <ul> + * <li>a CYCLIC subscription that fires every second, and</li> + * <li>a CHANGE_OF_STATE subscription that only fires when the value actually changes.</li> + * </ul> + * It listens for 30 seconds, then unsubscribes cleanly. + * <p> + * To produce visible value changes, change {@code g_b16} on the PLC while this is running. + */ +public class ManualUmasSubscriptionTest { + + private static final String CONNECTION_URL = "umas://192.168.24.31"; + private static final String TAG_ADDRESS = "g_b16"; + private static final long LISTEN_DURATION_MS = 30_000L; + + public static void main(String[] args) throws Exception { + try (PlcConnection connection = new DefaultPlcDriverManager() + .getConnection(CONNECTION_URL, new PlcNullAuthentication())) { + System.out.printf("Connected. metadata.subscribeSupported=%s%n", + connection.getMetadata().isSubscribeSupported()); + + PlcSubscriptionResponse subResp = connection.subscriptionRequestBuilder() + .addCyclicTagAddress("cyclic-tag", TAG_ADDRESS, Duration.ofSeconds(1)) + .addChangeOfStateTagAddress("change-tag", TAG_ADDRESS) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + + PlcResponseCode cyclicCode = subResp.getResponseCode("cyclic-tag"); + PlcResponseCode changeCode = subResp.getResponseCode("change-tag"); + System.out.printf("Subscribe response: cyclic=%s, change-of-state=%s%n", cyclicCode, changeCode); + if (cyclicCode != PlcResponseCode.OK || changeCode != PlcResponseCode.OK) { + System.err.println("Subscription failed — aborting"); + return; + } + + PlcSubscriptionHandle cyclicHandle = subResp.getSubscriptionHandle("cyclic-tag"); + PlcSubscriptionHandle changeHandle = subResp.getSubscriptionHandle("change-tag"); + cyclicHandle.register(event -> System.out.printf("[%s] CYCLIC cyclic-tag = %s%n", + event.getTimestamp(), event.getPlcValue("cyclic-tag"))); + changeHandle.register(event -> System.out.printf("[%s] CHANGE_OF_STATE change-tag = %s%n", + event.getTimestamp(), event.getPlcValue("change-tag"))); + + System.out.printf("Listening for emulated pushes for %ds...%n", LISTEN_DURATION_MS / 1000); + Thread.sleep(LISTEN_DURATION_MS); + + connection.unsubscriptionRequestBuilder() + .addHandles(cyclicHandle, changeHandle) + .build() + .execute() + .get(5, TimeUnit.SECONDS); + System.out.println("Unsubscribed."); + } + } +}
