This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 089c1994e634 CAMEL-25617: camel-plc4x - Upgrade to PLC4X 1.0.0 (#28387)
089c1994e634 is described below
commit 089c1994e634e8ec1d1352733b3266d2c0f8ae27
Author: Serdar Gökay <[email protected]>
AuthorDate: Tue Oct 27 09:54:07 2026 +0300
CAMEL-25617: camel-plc4x - Upgrade to PLC4X 1.0.0 (#28387)
Upgrades camel-plc4x from PLC4X 0.13.1 to 1.0.0, which requires Java 21 and
removed the plc4j-scraper module the triggered consumer was built on.
The triggered consumer is now a Camel ScheduledPollConsumer on the
endpoint's connection. It reads the trigger and the data tags in one request
and keeps the rising-edge semantics, recovers after a failed read, and leaves
no scraper thread pools behind. A non-boolean trigger or a read without an
answer within 5 s fails the poll.
A lost connection is closed and replaced instead of reconnected, since the
1.0.0 S7 and EtherNet/IP connections cannot connect again. The connection is
replaced under the endpoint lock, a live connection is kept on restart, a lost
one is closed on stop, and a request without a connection fails with
PlcConnectionException instead of an NPE. The driver no longer receives Camel's
own endpoint options in its connection string.
The upgrade guide covers the PLC4X 1.0.0 changes: Java 21, the array
notation of tag addresses, the removed s7-light driver, renamed connection
parameters and the polled trigger.
Closes #28387
Co-authored-by: Claude Fable 5.1 <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/plc4x-component.adoc | 35 ++
components/camel-plc4x/pom.xml | 5 -
.../camel-plc4x/src/main/docs/plc4x-component.adoc | 35 ++
.../apache/camel/component/plc4x/Constants.java | 1 -
.../camel/component/plc4x/Plc4XConsumer.java | 74 +---
.../camel/component/plc4x/Plc4XEndpoint.java | 142 ++++++--
.../component/plc4x/Plc4XPollingConsumer.java | 12 +-
.../camel/component/plc4x/Plc4XProducer.java | 2 +-
.../component/plc4x/Plc4XTriggeredConsumer.java | 150 ++++++++
.../apache/camel/component/plc4x/MockDriver.java | 258 ++++++++++---
.../component/plc4x/Plc4XConsumerRouteTest.java | 397 +++++++++++++++++++++
.../component/plc4x/Plc4XConsumerStopTest.java | 67 ----
.../camel/component/plc4x/Plc4XConsumerTest.java | 90 -----
.../camel/component/plc4x/Plc4XEndpointTest.java | 25 +-
.../camel/component/plc4x/Plc4XFailureTest.java | 4 -
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 44 +++
parent/pom.xml | 2 +-
17 files changed, 1016 insertions(+), 327 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc
index 294a7b699882..37f10570ac51 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/plc4x-component.adoc
@@ -91,6 +91,41 @@ no PLC. With a real PLC, the driver in the URI and the
addresses of the tags fol
message: "Tags read: ${body}"
----
+With the `trigger` option, the consumer reads the tags each time a boolean
trigger tag rises. It polls the trigger tag
+every `period` milliseconds on the connection of the endpoint, starting when
the route starts; `period` is required
+with `trigger` and must be positive. On a rising edge of the trigger, from
`false` (or from no value read yet) to
+`true`, the consumer delivers one exchange whose body is a
`Map<String,Object>` of the tags, read in the same request
+as the trigger. The trigger tag is not part of the body; it is read under the
reserved name `TRIGGER_VAR`, which no tag
+may use. While the trigger stays `true`, no further exchange is delivered.
+
+A poll that fails, for example because a tag cannot be read, the PLC cannot be
reached or does not answer within five
+seconds, or the trigger reads as a value that is not a boolean (such as a
number or an array), is reported to the
+exception handler of the consumer (or to the error handler of the route with
`bridgeErrorHandler=true`), and polling
+continues.
+The trigger value read before the failure is forgotten, so a trigger that is
`true` at the first successful poll after
+a failure counts as a rising edge. Automatic recovery after a lost connection
needs `autoReconnect=true`: a poll then
+replaces a lost connection with a new one. With the default
`autoReconnect=false`, the endpoint stops on the first lost
+connection and does not recover. Likewise, a connection that cannot be set up
when the route starts is retried by the
+polls with `autoReconnect=true`, and fails the start of the route without it.
+
+The following route reads a register of a Modbus TCP device each time coil 1
goes from `false` to `true`, checking the
+coil every 100 milliseconds:
+
+[source,yaml]
+----
+- route:
+ from:
+ uri: plc4x:modbus-tcp://192.168.0.10:502
+ parameters:
+ trigger: "coil:1"
+ period: 100
+ autoReconnect: true
+ tag.counter: "holding-register:2:INT"
+ steps:
+ - log:
+ message: "Counter read on trigger: ${body}"
+----
+
=== Polling Consumer
The polling consumer supports consecutive reading. The input and output are
the same as for the regular consumer.
diff --git a/components/camel-plc4x/pom.xml b/components/camel-plc4x/pom.xml
index 32f826e824f0..0bea7af6f9eb 100644
--- a/components/camel-plc4x/pom.xml
+++ b/components/camel-plc4x/pom.xml
@@ -44,11 +44,6 @@
<artifactId>plc4j-api</artifactId>
<version>${plc4x-version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.plc4x</groupId>
- <artifactId>plc4j-scraper</artifactId>
- <version>${plc4x-version}</version>
- </dependency>
<!-- Include all drivers -->
<dependency>
diff --git a/components/camel-plc4x/src/main/docs/plc4x-component.adoc
b/components/camel-plc4x/src/main/docs/plc4x-component.adoc
index 294a7b699882..37f10570ac51 100644
--- a/components/camel-plc4x/src/main/docs/plc4x-component.adoc
+++ b/components/camel-plc4x/src/main/docs/plc4x-component.adoc
@@ -91,6 +91,41 @@ no PLC. With a real PLC, the driver in the URI and the
addresses of the tags fol
message: "Tags read: ${body}"
----
+With the `trigger` option, the consumer reads the tags each time a boolean
trigger tag rises. It polls the trigger tag
+every `period` milliseconds on the connection of the endpoint, starting when
the route starts; `period` is required
+with `trigger` and must be positive. On a rising edge of the trigger, from
`false` (or from no value read yet) to
+`true`, the consumer delivers one exchange whose body is a
`Map<String,Object>` of the tags, read in the same request
+as the trigger. The trigger tag is not part of the body; it is read under the
reserved name `TRIGGER_VAR`, which no tag
+may use. While the trigger stays `true`, no further exchange is delivered.
+
+A poll that fails, for example because a tag cannot be read, the PLC cannot be
reached or does not answer within five
+seconds, or the trigger reads as a value that is not a boolean (such as a
number or an array), is reported to the
+exception handler of the consumer (or to the error handler of the route with
`bridgeErrorHandler=true`), and polling
+continues.
+The trigger value read before the failure is forgotten, so a trigger that is
`true` at the first successful poll after
+a failure counts as a rising edge. Automatic recovery after a lost connection
needs `autoReconnect=true`: a poll then
+replaces a lost connection with a new one. With the default
`autoReconnect=false`, the endpoint stops on the first lost
+connection and does not recover. Likewise, a connection that cannot be set up
when the route starts is retried by the
+polls with `autoReconnect=true`, and fails the start of the route without it.
+
+The following route reads a register of a Modbus TCP device each time coil 1
goes from `false` to `true`, checking the
+coil every 100 milliseconds:
+
+[source,yaml]
+----
+- route:
+ from:
+ uri: plc4x:modbus-tcp://192.168.0.10:502
+ parameters:
+ trigger: "coil:1"
+ period: 100
+ autoReconnect: true
+ tag.counter: "holding-register:2:INT"
+ steps:
+ - log:
+ message: "Counter read on trigger: ${body}"
+----
+
=== Polling Consumer
The polling consumer supports consecutive reading. The input and output are
the same as for the regular consumer.
diff --git
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Constants.java
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Constants.java
index d973e4e37ddd..6f995dde53fa 100644
---
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Constants.java
+++
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Constants.java
@@ -21,7 +21,6 @@ public final class Constants {
public static final String FIELD_NAME_HEADER = "fieldName";
public static final String FIELD_QUERY_HEADER = "fieldQuery";
public static final String TRIGGER = "TRIGGER_VAR";
- public static final String PLC_NAME = "PLC";
private Constants() {
throw new IllegalStateException("Utility class!");
diff --git
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java
index 581191a7929d..bb4d8dbe7686 100644
---
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java
+++
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XConsumer.java
@@ -16,10 +16,6 @@
*/
package org.apache.camel.component.plc4x;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -30,33 +26,20 @@ import org.apache.camel.Processor;
import org.apache.camel.support.DefaultConsumer;
import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
import org.apache.plc4x.java.api.messages.PlcReadRequest;
-import org.apache.plc4x.java.scraper.config.JobConfigurationImpl;
-import org.apache.plc4x.java.scraper.config.ScraperConfiguration;
-import
org.apache.plc4x.java.scraper.config.triggeredscraper.ScraperConfigurationTriggeredImpl;
-import org.apache.plc4x.java.scraper.exception.ScraperException;
-import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScraperImpl;
-import
org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollector;
-import
org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollectorImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class Plc4XConsumer extends DefaultConsumer {
private static final Logger LOGGER =
LoggerFactory.getLogger(Plc4XConsumer.class);
- private final Map<String, String> tags;
- private final String trigger;
private final Plc4XEndpoint plc4XEndpoint;
private ScheduledExecutorService executorService;
private ScheduledFuture<?> future;
- private TriggeredScraperImpl scraper;
- private TriggerCollector collector;
public Plc4XConsumer(Plc4XEndpoint endpoint, Processor processor) {
super(endpoint, processor);
this.plc4XEndpoint = endpoint;
- this.tags = endpoint.getTags();
- this.trigger = endpoint.getTrigger();
}
@Override
@@ -82,20 +65,13 @@ public class Plc4XConsumer extends DefaultConsumer {
}
doStop();
}
- if (trigger == null) {
- startUnTriggered();
- } else {
- startTriggered();
- }
+ startUnTriggered();
}
private void startUnTriggered() {
PlcReadRequest request;
try {
plc4XEndpoint.reconnectIfNeeded();
- if (plc4XEndpoint.connection == null) {
- throw new PlcConnectionException("Cannot establish
connection");
- }
request = plc4XEndpoint.buildPlcReadRequest();
} catch (Exception e) {
if (LOGGER.isTraceEnabled()) {
@@ -111,11 +87,7 @@ public class Plc4XConsumer extends DefaultConsumer {
future = executorService.schedule(() ->
request.execute().thenAccept(response -> {
try {
Exchange exchange = plc4XEndpoint.createExchange();
- Map<String, Object> rsp = new HashMap<>();
- for (String field : response.getTagNames()) {
- rsp.put(field, response.getObject(field));
- }
- exchange.getIn().setBody(rsp);
+ exchange.getIn().setBody(Plc4XEndpoint.tagValues(response));
getProcessor().process(exchange);
} catch (Exception e) {
getExceptionHandler().handleException(e);
@@ -123,39 +95,6 @@ public class Plc4XConsumer extends DefaultConsumer {
}), 500, TimeUnit.MILLISECONDS);
}
- private void startTriggered() throws ScraperException {
- ScraperConfiguration configuration = getScraperConfig(tags);
- collector = new
TriggerCollectorImpl(plc4XEndpoint.getPlcDriverManager());
-
- scraper = new TriggeredScraperImpl(configuration, (job, alias,
response) -> {
- try {
- plc4XEndpoint.reconnectIfNeeded();
-
- Exchange exchange = plc4XEndpoint.createExchange();
- exchange.getIn().setBody(response);
- getProcessor().process(exchange);
- } catch (PlcConnectionException e) {
- if (LOGGER.isTraceEnabled()) {
- LOGGER.warn("Unable to reconnect, skipping request", e);
- } else {
- LOGGER.warn("Unable to reconnect, skipping request");
- }
- } catch (Exception e) {
- getExceptionHandler().handleException(e);
- }
- }, collector);
- scraper.start();
- collector.start();
- }
-
- private ScraperConfigurationTriggeredImpl getScraperConfig(Map<String,
String> tagList) {
- String config = "(TRIGGER_VAR," + plc4XEndpoint.getPeriod() + ",(" +
plc4XEndpoint.getTrigger() + ")==(true))";
- List<JobConfigurationImpl> job = Collections.singletonList(
- new JobConfigurationImpl("PLC4X-Camel", config, 0,
Collections.singletonList(Constants.PLC_NAME), tagList));
- Map<String, String> source =
Collections.singletonMap(Constants.PLC_NAME, plc4XEndpoint.getUri());
- return new ScraperConfigurationTriggeredImpl(source, job);
- }
-
@Override
protected void doStop() throws Exception {
// First stop the polling process
@@ -167,15 +106,6 @@ public class Plc4XConsumer extends DefaultConsumer {
getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executorService);
executorService = null;
}
- // a scraper that is not stopped keeps reading the PLC and sending
exchanges to the route
- if (scraper != null) {
- scraper.stop();
- scraper = null;
- }
- if (collector != null) {
- collector.stop();
- collector = null;
- }
super.doStop();
}
diff --git
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java
index 55fd8e923881..c541fb899801 100644
---
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java
+++
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XEndpoint.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.plc4x;
+import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.camel.Category;
@@ -28,17 +31,20 @@ import org.apache.camel.spi.BodyType;
import org.apache.camel.spi.BodyType.Carrier;
import org.apache.camel.spi.BodyTypes;
import org.apache.camel.spi.EndpointServiceLocation;
+import org.apache.camel.spi.EndpointUriFactory;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.support.DefaultEndpoint;
import org.apache.camel.util.StringHelper;
+import org.apache.camel.util.URISupport;
import org.apache.plc4x.java.DefaultPlcDriverManager;
import org.apache.plc4x.java.api.PlcConnection;
import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
import org.apache.plc4x.java.api.exceptions.PlcIncompatibleDatatypeException;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,8 +67,11 @@ import org.slf4j.LoggerFactory;
public class Plc4XEndpoint extends DefaultEndpoint implements
EndpointServiceLocation {
private static final Logger LOGGER =
LoggerFactory.getLogger(Plc4XEndpoint.class);
+ // how long the producer and the triggered consumer wait for the PLC to
answer a request
+ static final long REQUEST_TIMEOUT_MILLIS = 5000;
+
protected DefaultPlcDriverManager plcDriverManager;
- protected PlcConnection connection;
+ protected volatile PlcConnection connection;
@UriPath
@Metadata(required = true, description = "PLC4X connection string for the
connection to the target")
@@ -82,6 +91,8 @@ public class Plc4XEndpoint extends DefaultEndpoint implements
EndpointServiceLoc
private boolean autoReconnect;
private String uri;
+ // the connection string handed to the driver: the uri without the options
of this endpoint
+ private String connectionString;
public Plc4XEndpoint(String endpointUri, Component component) {
super(endpointUri, component);
@@ -133,7 +144,13 @@ public class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
*/
public void setupConnection() throws PlcConnectionException {
try {
- connection = plcDriverManager.getConnection(this.uri);
+ synchronized (this) {
+ // a restarted route or another consumer or producer of the
endpoint keeps a live connection
+ if (!isConnected()) {
+ closeConnection();
+ connection =
plcDriverManager.getConnection(connectionString);
+ }
+ }
if (!isConnected()) {
reconnectIfNeeded();
}
@@ -158,24 +175,34 @@ public class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
* @throws PlcConnectionException If reconnect failed and auto-reconnect
is turned on
*/
public void reconnectIfNeeded() throws PlcConnectionException {
- if (isConnected()) {
- LOGGER.trace("No reconnect needed, already connected");
- } else if (autoReconnect && connection == null) {
- connection = plcDriverManager.getConnection(uri);
- LOGGER.debug("Successfully reconnected");
- } else if (autoReconnect && !isConnected()) {
- if (connection != null) {
- connection.connect();
+ // the producer and the consumers of the endpoint share the
connection: one of them replaces it at a time
+ synchronized (this) {
+ if (isConnected()) {
+ LOGGER.trace("No reconnect needed, already connected");
+ return;
}
- // If reconnection fails without Exception, reset connection
- if (!isConnected()) {
- LOGGER.debug("No connection established after connect,
resetting connection");
- connection = plcDriverManager.getConnection(uri);
+ if (autoReconnect) {
+ // a lost PLC4X connection cannot always connect again (the S7
and EtherNet/IP drivers fail on a closed
+ // transport), so it is replaced by a new one
+ closeConnection();
+ connection = plcDriverManager.getConnection(connectionString);
+ LOGGER.debug("Successfully reconnected");
+ return;
+ }
+ }
+ // stop() is called outside the lock, as a concurrent stop() holds the
service lock while doStop() waits for it
+ LOGGER.warn("Connection lost and auto-reconnect is turned off,
shutting down Plc4XEndpoint");
+ stop();
+ }
+
+ private void closeConnection() {
+ if (connection != null) {
+ try {
+ connection.close();
+ } catch (Exception e) {
+ LOGGER.debug("Closing the connection failed: {}",
e.getMessage());
}
- LOGGER.debug("Successfully reconnected");
- } else {
- LOGGER.warn("Connection lost and auto-reconnect is turned off,
shutting down Plc4XEndpoint");
- stop();
+ connection = null;
}
}
@@ -183,14 +210,16 @@ public class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
* @return true if connection supports writing, else false
*/
public boolean canWrite() {
- return connection != null &&
connection.getMetadata().isWriteSupported();
+ PlcConnection current = connection;
+ return current != null && current.getMetadata().isWriteSupported();
}
/**
* @return true if connection is established and still connected
*/
public boolean isConnected() {
- return connection != null && connection.isConnected();
+ PlcConnection current = connection;
+ return current != null && current.isConnected();
}
@Override
@@ -200,7 +229,8 @@ public class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- Plc4XConsumer consumer = new Plc4XConsumer(this, processor);
+ Consumer consumer = trigger == null
+ ? new Plc4XConsumer(this, processor) : new
Plc4XTriggeredConsumer(this, processor);
configureConsumer(consumer);
return consumer;
}
@@ -213,8 +243,25 @@ public class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
/**
* Build a {@link PlcReadRequest} using the tags specified in the endpoint.
*/
- public PlcReadRequest buildPlcReadRequest() {
- PlcReadRequest.Builder builder = connection.readRequestBuilder();
+ public PlcReadRequest buildPlcReadRequest() throws PlcConnectionException {
+ return buildPlcReadRequest(Map.of());
+ }
+
+ /**
+ * Build a {@link PlcReadRequest} using the given tags and the tags
specified in the endpoint.
+ *
+ * @param additionalTags tag addresses by name to read in
addition to the tags of the endpoint
+ * @return {@link PlcReadRequest}
+ * @throws PlcConnectionException if there is no connection
+ */
+ PlcReadRequest buildPlcReadRequest(Map<String, String> additionalTags)
throws PlcConnectionException {
+ // read the field once: a concurrent stop or reconnect can clear it at
any time
+ PlcConnection current = connection;
+ if (current == null) {
+ throw new PlcConnectionException("Cannot establish connection");
+ }
+ PlcReadRequest.Builder builder = current.readRequestBuilder();
+ additionalTags.forEach(builder::addTagAddress);
if (tags != null) {
for (Map.Entry<String, String> tag : tags.entrySet()) {
try {
@@ -228,15 +275,32 @@ public class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
return builder.build();
}
+ /**
+ * @return the values of a read response by tag name
+ */
+ static Map<String, Object> tagValues(PlcReadResponse response) {
+ Map<String, Object> values = new HashMap<>();
+ for (String tag : response.getTagNames()) {
+ values.put(tag, response.getObject(tag));
+ }
+ return values;
+ }
+
/**
* Build a {@link PlcWriteRequest}.
* <p>
*
- * @param tags tags to add to write request
- * @return {@link PlcWriteRequest}
+ * @param tags tags to add to write request
+ * @return {@link PlcWriteRequest}
+ * @throws PlcConnectionException if there is no connection
*/
- public PlcWriteRequest buildPlcWriteRequest(Map<String, Map<String,
Object>> tags) {
- PlcWriteRequest.Builder builder = connection.writeRequestBuilder();
+ public PlcWriteRequest buildPlcWriteRequest(Map<String, Map<String,
Object>> tags) throws PlcConnectionException {
+ // read the field once: a concurrent stop or reconnect can clear it at
any time
+ PlcConnection current = connection;
+ if (current == null) {
+ throw new PlcConnectionException("Cannot establish connection");
+ }
+ PlcWriteRequest.Builder builder = current.writeRequestBuilder();
for (Map.Entry<String, Map<String, Object>> entry : tags.entrySet()) {
//Tags are stored like this --> Map<Tagname,Map<Query,Value>> for
writing
@@ -271,16 +335,34 @@ public class Plc4XEndpoint extends DefaultEndpoint
implements EndpointServiceLoc
@Override
protected void doStart() throws Exception {
super.doStart();
+ connectionString = withoutEndpointOptions(uri);
ClassLoader cl = getCamelContext().getApplicationContextClassLoader();
this.plcDriverManager = new DefaultPlcDriverManager(cl);
}
+ /**
+ * Removes the options of this endpoint from a connection string, so the
PLC4X driver only gets the parameters it
+ * knows.
+ */
+ private String withoutEndpointOptions(String endpointUri) throws
URISyntaxException {
+ EndpointUriFactory factory =
getCamelContext().getCamelContextExtension().getEndpointUriFactory("plc4x");
+ if (factory == null) {
+ throw new IllegalStateException("Cannot find the
EndpointUriFactory of the plc4x component");
+ }
+ Map<String, Object> parameters
+ = new
LinkedHashMap<>(URISupport.parseQuery(URISupport.extractQuery(endpointUri),
true));
+ parameters.keySet().removeIf(name ->
factory.propertyNames().contains(name)
+ ||
factory.multiValuePrefixes().values().stream().anyMatch(name::startsWith));
+ String query = URISupport.createQueryString(parameters, false);
+ String path = URISupport.stripQuery(endpointUri);
+ return query.isEmpty() ? path : path + "?" + query;
+ }
+
@Override
public void doStop() throws Exception {
super.doStop();
- if (isConnected() && connection != null) {
- connection.close();
- connection = null;
+ synchronized (this) {
+ closeConnection();
}
}
diff --git
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java
index 6fcf7242244f..b35c2d306e6e 100644
---
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java
+++
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XPollingConsumer.java
@@ -16,8 +16,6 @@
*/
package org.apache.camel.component.plc4x;
-import java.util.HashMap;
-import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -86,10 +84,6 @@ public class Plc4XPollingConsumer extends
EventDrivenPollingConsumer {
Exchange exchange = plc4XEndpoint.createExchange();
try {
plc4XEndpoint.reconnectIfNeeded();
- if (plc4XEndpoint.connection == null) {
- throw new PlcConnectionException("Cannot establish
connection");
- }
-
PlcReadRequest request = plc4XEndpoint.buildPlcReadRequest();
CompletableFuture<? extends PlcReadResponse> future
= request.execute().whenComplete((plcReadResponse,
throwable) -> {
@@ -101,11 +95,7 @@ public class Plc4XPollingConsumer extends
EventDrivenPollingConsumer {
response = future.get();
}
- Map<String, Object> rsp = new HashMap<>();
- for (String field : response.getTagNames()) {
- rsp.put(field, response.getObject(field));
- }
- exchange.getIn().setBody(rsp);
+ exchange.getIn().setBody(Plc4XEndpoint.tagValues(response));
} catch (TimeoutException e) {
// nothing received within the timeout
LOGGER.debug("No response from the PLC within {} millis", timeout);
diff --git
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XProducer.java
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XProducer.java
index 6b82e831ead0..13719137f77e 100644
---
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XProducer.java
+++
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XProducer.java
@@ -83,7 +83,7 @@ public class Plc4XProducer extends DefaultAsyncProducer {
int currentlyOpenRequests = openRequests.incrementAndGet();
try {
log.debug("Currently open requests including {}:{}", exchange,
currentlyOpenRequests);
- Object plcWriteResponse = completableFuture.get(5000,
TimeUnit.MILLISECONDS);
+ Object plcWriteResponse =
completableFuture.get(Plc4XEndpoint.REQUEST_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS);
if (exchange.getPattern().isOutCapable()) {
Message out = exchange.getMessage();
out.copyFrom(exchange.getIn());
diff --git
a/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java
new file mode 100644
index 000000000000..a742ac2a878b
--- /dev/null
+++
b/components/camel-plc4x/src/main/java/org/apache/camel/component/plc4x/Plc4XTriggeredConsumer.java
@@ -0,0 +1,150 @@
+/*
+ * 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.camel.component.plc4x;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.support.ScheduledPollConsumer;
+import org.apache.plc4x.java.api.exceptions.PlcException;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
+
+/**
+ * Reads the trigger tag together with the tags every {@code period}
milliseconds and emits the tags once on each rising
+ * edge of the trigger.
+ */
+public class Plc4XTriggeredConsumer extends ScheduledPollConsumer {
+ private final Plc4XEndpoint plc4XEndpoint;
+ private final String trigger;
+ private final int period;
+
+ // whether the trigger was true on the last poll; null when unknown,
before the first poll or after a failed one
+ private Boolean lastTrigger;
+
+ public Plc4XTriggeredConsumer(Plc4XEndpoint endpoint, Processor processor)
{
+ super(endpoint, processor);
+ this.plc4XEndpoint = endpoint;
+ this.trigger = endpoint.getTrigger();
+ this.period = endpoint.getPeriod();
+ setInitialDelay(0);
+ setDelay(period);
+ setTimeUnit(TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public Plc4XEndpoint getEndpoint() {
+ return plc4XEndpoint;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (period <= 0) {
+ throw new IllegalArgumentException(
+ "The period must be a positive number of milliseconds when
a trigger is set, but was " + period);
+ }
+ Map<String, String> tags = plc4XEndpoint.getTags();
+ if (tags != null && tags.containsKey(Constants.TRIGGER)) {
+ throw new IllegalArgumentException(
+ "The tag name " + Constants.TRIGGER + " is reserved for
the trigger when a trigger is set");
+ }
+ lastTrigger = null;
+ // without autoReconnect a connection that cannot be set up fails the
start, as no poll would reconnect
+ plc4XEndpoint.setupConnection();
+ super.doStart();
+ }
+
+ @Override
+ protected int poll() throws Exception {
+ PlcReadResponse response;
+ try {
+ response = read();
+ } catch (Exception e) {
+ lastTrigger = null;
+ throw e;
+ }
+ if (!isRunAllowed()) {
+ // the consumer stopped while the read was in progress
+ return 0;
+ }
+ boolean triggered = (Boolean) response.getObject(Constants.TRIGGER);
+ boolean risingEdge = triggered && !Boolean.TRUE.equals(lastTrigger);
+ lastTrigger = triggered;
+ if (!risingEdge) {
+ return 0;
+ }
+
+ Map<String, Object> tags = Plc4XEndpoint.tagValues(response);
+ tags.remove(Constants.TRIGGER);
+ Exchange exchange = createExchange(false);
+ exchange.getIn().setBody(tags);
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ } finally {
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing
exchange", exchange, exchange.getException());
+ }
+ releaseExchange(exchange, false);
+ }
+ return 1;
+ }
+
+ /**
+ * Reads the trigger and the tags in one request; fails unless every tag
was read and the trigger is a boolean.
+ */
+ private PlcReadResponse read() throws Exception {
+ plc4XEndpoint.reconnectIfNeeded();
+ CompletableFuture<? extends PlcReadResponse> future
+ = plc4XEndpoint.buildPlcReadRequest(Map.of(Constants.TRIGGER,
trigger)).execute();
+ PlcReadResponse response;
+ try {
+ response = future.get(Plc4XEndpoint.REQUEST_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS);
+ } catch (ExecutionException e) {
+ throw e.getCause() instanceof Exception cause ? cause : e;
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ throw new TimeoutException(
+ "No response to the read of the trigger " + trigger + "
within "
+ + Plc4XEndpoint.REQUEST_TIMEOUT_MILLIS
+ " millis");
+ }
+ for (String tag : response.getTagNames()) {
+ PlcResponseCode code = response.getResponseCode(tag);
+ if (code != PlcResponseCode.OK) {
+ throw new PlcException("Reading the tag " + tag + " failed
with response code " + code);
+ }
+ }
+ Object value = response.getObject(Constants.TRIGGER);
+ if (!(value instanceof Boolean)) {
+ throw new PlcException(
+ "The trigger " + trigger + " must read as a Boolean, but
read "
+ + (value == null ? "no value" : "a " +
value.getClass().getName()));
+ }
+ return response;
+ }
+
+ @Override
+ public String toString() {
+ return "Plc4XTriggeredConsumer[" + plc4XEndpoint + "]";
+ }
+}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java
index 71815047e3a7..a27f979dc6f3 100644
---
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java
+++
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/MockDriver.java
@@ -16,32 +16,150 @@
*/
package org.apache.camel.component.plc4x;
-import java.util.HashMap;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.plc4x.java.api.PlcConnection;
import org.apache.plc4x.java.api.PlcDriver;
import org.apache.plc4x.java.api.authentication.PlcAuthentication;
import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
+import org.apache.plc4x.java.api.exceptions.PlcException;
+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.PlcSubscriptionRequest;
-import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse;
import org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest;
import org.apache.plc4x.java.api.messages.PlcWriteRequest;
-import org.apache.plc4x.java.spi.messages.DefaultPlcSubscriptionResponse;
-import org.apache.plc4x.java.spi.messages.PlcSubscriber;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
import static org.mockito.Mockito.*;
+/**
+ * Test driver for the {@code mock} protocol. A read returns the values a test
programmed with
+ * {@link #programRead(String, Object...)} when every tag in the request is
programmed, and Mockito deep stubs
+ * otherwise. A programmed {@link #failedRead()} fails the whole read, a
programmed
+ * {@link #responseCode(PlcResponseCode)} answers that tag with an error code.
A connection reports that it is connected
+ * from {@code connect()} until {@link #dropConnections()} drops it; a dropped
connection cannot connect again, as after
+ * a PLC restart. The programming is static because every driver manager loads
its own driver instance; tests call
+ * {@link #reset()}.
+ */
public class MockDriver implements PlcDriver {
- public static final Logger LOGGER =
LoggerFactory.getLogger(MockDriver.class);
+ private static final Map<String, ProgrammedTag> PROGRAMMED_TAGS = new
ConcurrentHashMap<>();
+ private static volatile CompletableFuture<Void> heldReads =
CompletableFuture.completedFuture(null);
+ private static volatile CompletableFuture<Void> heldRead = new
CompletableFuture<>();
+ private static final AtomicInteger REFUSED_CONNECTIONS = new
AtomicInteger();
+ private static final List<String> CONNECTION_STRINGS = new
CopyOnWriteArrayList<>();
+ private static final Object FAILED_READ = new Object();
+ private static volatile boolean refuseConnections;
+ private static final AtomicInteger CREATED_CONNECTIONS = new
AtomicInteger();
+ private static volatile int droppedConnections;
+
+ /**
+ * Programs the values that successive reads of a tag address return; the
last value repeats once they are used up.
+ */
+ public static void programRead(String tagAddress, Object... values) {
+ if (values.length == 0) {
+ throw new IllegalArgumentException("At least one value is required
for " + tagAddress);
+ }
+ PROGRAMMED_TAGS.put(tagAddress, new
ProgrammedTag(Arrays.asList(values)));
+ }
+
+ /**
+ * A programmed value that makes the whole read in which it is due fail.
+ */
+ public static Object failedRead() {
+ return FAILED_READ;
+ }
+
+ /**
+ * A programmed value that makes the read in which it is due answer the
tag with the given response code and no
+ * value.
+ */
+ public static Object responseCode(PlcResponseCode responseCode) {
+ return new FailedTag(responseCode);
+ }
+
+ /**
+ * Holds the following reads until {@link #releaseReads()}; a held read
ignores interrupts, as a read that a PLC has
+ * already answered does.
+ */
+ public static void holdReads() {
+ heldRead = new CompletableFuture<>();
+ heldReads = new CompletableFuture<>();
+ }
+
+ /**
+ * Waits until a read is held by {@link #holdReads()}.
+ */
+ public static void awaitHeldRead() throws Exception {
+ heldRead.get(10, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Lets the reads held by {@link #holdReads()} complete.
+ */
+ public static void releaseReads() {
+ heldReads.complete(null);
+ }
+
+ /**
+ * Makes every following connection attempt fail.
+ */
+ public static void refuseConnections() {
+ refuseConnections = true;
+ }
+
+ /**
+ * Makes the following connection attempts succeed again.
+ */
+ public static void acceptConnections() {
+ refuseConnections = false;
+ }
- ExecutorService executorService = Executors.newFixedThreadPool(10);
+ /**
+ * Drops the connections made so far, as a restarted PLC does: they report
that they are not connected from now on
+ * and fail to connect again, while a new connection succeeds.
+ */
+ public static void dropConnections() {
+ droppedConnections = CREATED_CONNECTIONS.get();
+ }
+
+ /**
+ * @return the number of connection attempts refused since the last {@link
#reset()}
+ */
+ public static int refusedConnections() {
+ return REFUSED_CONNECTIONS.get();
+ }
+
+ /**
+ * @return the connection strings the driver was asked to connect with
since the last {@link #reset()}
+ */
+ public static List<String> connectionStrings() {
+ return CONNECTION_STRINGS;
+ }
+
+ /**
+ * Forgets all programmed tag values and accepts connections again.
+ */
+ public static void reset() {
+ releaseReads();
+ PROGRAMMED_TAGS.clear();
+ refuseConnections = false;
+ REFUSED_CONNECTIONS.set(0);
+ CREATED_CONNECTIONS.set(0);
+ droppedConnections = 0;
+ CONNECTION_STRINGS.clear();
+ }
@Override
public String getProtocolCode() {
@@ -55,52 +173,108 @@ public class MockDriver implements PlcDriver {
@Override
public PlcConnection getConnection(String url) throws
PlcConnectionException {
+ CONNECTION_STRINGS.add(url);
+ if (refuseConnections) {
+ REFUSED_CONNECTIONS.incrementAndGet();
+ throw new PlcConnectionException("Connection refused by the mock
driver");
+ }
// Mock a connection.
PlcConnection plcConnectionMock = mock(PlcConnection.class,
RETURNS_DEEP_STUBS);
+ int connectionNumber = CREATED_CONNECTIONS.getAndIncrement();
+ AtomicBoolean connected = new AtomicBoolean();
+ doAnswer(invocation -> {
+ if (connectionNumber < droppedConnections) {
+ throw new PlcConnectionException("Transport is not open");
+ }
+ connected.set(true);
+ return null;
+ }).when(plcConnectionMock).connect();
+ when(plcConnectionMock.isConnected())
+ .thenAnswer(invocation -> connected.get() && connectionNumber
>= droppedConnections);
when(plcConnectionMock.getMetadata().isReadSupported()).thenReturn(true);
when(plcConnectionMock.getMetadata().isWriteSupported()).thenReturn(true);
-
when(plcConnectionMock.readRequestBuilder()).thenReturn(mock(PlcReadRequest.Builder.class,
RETURNS_DEEP_STUBS));
+ when(plcConnectionMock.readRequestBuilder()).thenAnswer(invocation ->
readRequestBuilder());
when(plcConnectionMock.writeRequestBuilder()).thenReturn(mock(PlcWriteRequest.Builder.class,
RETURNS_DEEP_STUBS));
when(plcConnectionMock.subscriptionRequestBuilder())
.thenReturn(mock(PlcSubscriptionRequest.Builder.class,
RETURNS_DEEP_STUBS));
when(plcConnectionMock.unsubscriptionRequestBuilder())
.thenReturn(mock(PlcUnsubscriptionRequest.Builder.class,
RETURNS_DEEP_STUBS));
-
- // Mock a typical subscriber.
- PlcSubscriber plcSubscriber = mock(PlcSubscriber.class,
RETURNS_DEEP_STUBS);
-
when(plcSubscriber.subscribe(any(PlcSubscriptionRequest.class))).thenAnswer(invocation
-> {
- LOGGER.info("Received {}", invocation);
- // TODO: Translate this so it actually does something ...
- /*PlcSubscriptionRequest subscriptionRequest =
invocation.getArgument(0);
- List<PlcSubscriptionResponse> responseItems =
- subscriptionRequest.getFieldNames().stream().map(
- fieldName ->
subscriptionRequest.getField(fieldName)).map(field -> {
- Consumer consumer = subscriptionRequestItem.getConsumer();
- executorService.submit(() -> {
- while (!Thread.currentThread().isInterrupted()) {
- consumer.accept(new SubscriptionEventItem<>(null,
Calendar.getInstance(), Collections.singletonList("HelloWorld")));
- try {
- TimeUnit.MILLISECONDS.sleep(100);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- }
- }
- });
- return new
SubscriptionResponseItem<>(subscriptionRequestItem,
- mock(PlcSubscriptionHandle.class, RETURNS_DEEP_STUBS),
PlcResponseCode.OK);
- }).collect(Collectors.toList());
- PlcSubscriptionResponse response = new
PlcSubscriptionResponse(subscriptionRequest, responseItems);*/
- PlcSubscriptionResponse response
- = new
DefaultPlcSubscriptionResponse(mock(PlcSubscriptionRequest.class), new
HashMap<>());
- return CompletableFuture.completedFuture(response);
- });
return plcConnectionMock;
}
@Override
public PlcConnection getConnection(String url, PlcAuthentication
authentication) throws PlcConnectionException {
- return getConnection(null);
+ return getConnection(url);
+ }
+
+ private static PlcReadRequest.Builder readRequestBuilder() {
+ PlcReadRequest.Builder unprogrammed =
mock(PlcReadRequest.Builder.class, RETURNS_DEEP_STUBS);
+ Map<String, String> tags = new LinkedHashMap<>();
+ return mock(PlcReadRequest.Builder.class, invocation -> switch
(invocation.getMethod().getName()) {
+ case "addTagAddress" -> {
+ tags.put(invocation.getArgument(0), invocation.getArgument(1));
+ unprogrammed.addTagAddress(invocation.getArgument(0),
invocation.getArgument(1));
+ yield invocation.getMock();
+ }
+ case "build" -> !tags.isEmpty() &&
PROGRAMMED_TAGS.keySet().containsAll(tags.values())
+ ? programmedRequest(tags) : unprogrammed.build();
+ default -> RETURNS_DEFAULTS.answer(invocation);
+ });
+ }
+
+ private static PlcReadRequest programmedRequest(Map<String, String> tags) {
+ return mock(PlcReadRequest.class, invocation -> switch
(invocation.getMethod().getName()) {
+ case "execute" -> read(tags);
+ case "getTagNames" -> new LinkedHashSet<>(tags.keySet());
+ default -> RETURNS_DEFAULTS.answer(invocation);
+ });
+ }
+
+ private static CompletableFuture<PlcReadResponse> read(Map<String, String>
tags) {
+ heldRead.complete(null);
+ // join() does not return early on an interrupt
+ heldReads.join();
+ Map<String, Object> values = new LinkedHashMap<>();
+ tags.forEach((name, address) -> values.put(name,
PROGRAMMED_TAGS.get(address).next()));
+ if (values.containsValue(FAILED_READ)) {
+ return CompletableFuture.failedFuture(new PlcException("Read
failed in the mock driver"));
+ }
+ return CompletableFuture.completedFuture(mock(PlcReadResponse.class,
+ invocation -> switch (invocation.getMethod().getName()) {
+ case "getTagNames" -> List.copyOf(values.keySet());
+ case "getObject" -> value(values.get(invocation.<String>
getArgument(0)));
+ case "getResponseCode" -> values.get(invocation.<String>
getArgument(0)) instanceof FailedTag failed
+ ? failed.responseCode() : PlcResponseCode.OK;
+ default -> RETURNS_DEFAULTS.answer(invocation);
+ }));
+ }
+
+ private static Object value(Object programmed) {
+ if (programmed instanceof FailedTag failed) {
+ throw new PlcRuntimeException("Tag read failed with response code
" + failed.responseCode());
+ }
+ return programmed;
+ }
+
+ private record FailedTag(PlcResponseCode responseCode) {
+ }
+
+ private static final class ProgrammedTag {
+
+ private final List<Object> values;
+ private int next;
+
+ ProgrammedTag(List<Object> values) {
+ this.values = values;
+ }
+
+ synchronized Object next() {
+ Object value = values.get(next);
+ if (next < values.size() - 1) {
+ next++;
+ }
+ return value;
+ }
}
}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java
new file mode 100644
index 000000000000..6a94252023fd
--- /dev/null
+++
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerRouteTest.java
@@ -0,0 +1,397 @@
+/*
+ * 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.camel.component.plc4x;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The consumer's contract at the route seam: a route reads a {@code
plc4x:mock:} endpoint whose tag values are
+ * programmed in {@link MockDriver}, and only what arrives at the mock
endpoint is asserted.
+ */
+public class Plc4XConsumerRouteTest extends CamelTestSupport {
+
+ private static final String PLC =
"plc4x:mock:10.10.10.1/1/1?autoReconnect=true";
+ private static final String TRIGGERED = PLC +
"&trigger=TRIG&period=100&tag.temperature=TEMP&tag.pressure=PRES";
+ private static final String UNTRIGGERED = PLC +
"&tag.temperature=TEMP&tag.pressure=PRES";
+ private static final String REPORTED = PLC +
"&trigger=TRIG&period=50&tag.temperature=TEMP&tag.pressure=PRES"
+ + "&exceptionHandler=#pollFailures";
+ private static final String WITHOUT_AUTO_RECONNECT
+ =
"plc4x:mock:10.10.10.1/1/1?trigger=TRIG&period=50&tag.temperature=TEMP";
+ private static final String RESERVED_TAG_NAME = PLC +
"&trigger=TRIG&period=50&tag.TRIGGER_VAR=TEMP";
+ private static final String WITHOUT_PERIOD = PLC +
"&trigger=TRIG&tag.temperature=TEMP";
+ private static final String DRIVER_PARAMETERS = PLC +
"&trigger=TRIG&period=100&tag.temperature=TEMP"
+ +
"&exceptionHandler=#pollFailures&exchangePattern=InOnly"
+ +
"&remote-rack=1&request-timeout-ms=500&percent-encoded=%25DB1";
+
+ private final PollFailures pollFailures = new PollFailures();
+
+ @BeforeEach
+ void resetMockDriver() {
+ MockDriver.reset();
+ }
+
+ @AfterAll
+ static void leaveMockDriverUnprogrammed() {
+ MockDriver.reset();
+ }
+
+ @Test
+ public void emitsOncePerRisingEdgeOfTheTrigger() throws Exception {
+ MockDriver.programRead("TRIG", false, false, true, true, false, true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:triggered");
+ result.expectedMessageCount(2);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("triggered");
+
+ result.assertIsSatisfied(10000);
+ }
+
+ @Test
+ public void deliversTheDataTagsWithoutTheTrigger() throws Exception {
+ MockDriver.programRead("TRIG", false, true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:triggered");
+ result.expectedMessageCount(1);
+
+ context.getRouteController().startRoute("triggered");
+
+ result.assertIsSatisfied(10000);
+ assertEquals(Map.of("temperature", 21, "pressure", 1013),
result.getExchanges().get(0).getIn().getBody(),
+ "the body holds the data tags, not the trigger");
+ }
+
+ @Test
+ public void emitsOnceForATriggerThatIsAlreadyTrueOnTheFirstRead() throws
Exception {
+ MockDriver.programRead("TRIG", true, true, false);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:triggered");
+ result.expectedMessageCount(1);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("triggered");
+
+ result.assertIsSatisfied(10000);
+ }
+
+ @Test
+ public void
stopsReadingWhenTheRouteStopsAndResumesWithoutDuplicatesAfterARestart() throws
Exception {
+ MockDriver.programRead("TRIG", false, true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:triggered");
+ result.expectedMessageCount(3);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("triggered");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
result.getReceivedCounter() == 1);
+ context.getRouteController().stopRoute("triggered");
+
+ MockDriver.programRead("TRIG", false, true, false, true, false);
+ await().during(1, TimeUnit.SECONDS).atMost(2,
TimeUnit.SECONDS).until(() -> result.getReceivedCounter() == 1);
+
+ context.getRouteController().startRoute("triggered");
+
+ result.assertIsSatisfied(10000);
+ assertEquals(1, MockDriver.connectionStrings().size(), "the restarted
route uses the connection that is still open");
+ }
+
+ @Test
+ public void readsTheTagsOnceWithoutATrigger() throws Exception {
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:untriggered");
+ result.expectedMessageCount(1);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("untriggered");
+
+ result.assertIsSatisfied(10000);
+ assertEquals(Map.of("temperature", 21, "pressure", 1013),
result.getExchanges().get(0).getIn().getBody());
+ }
+
+ @Test
+ public void deliversTheDataTagsOfTheReadThatSawTheEdge() throws Exception {
+ MockDriver.programRead("TRIG", false, true, false, true);
+ MockDriver.programRead("TEMP", 21, 22, 23, 24);
+ MockDriver.programRead("PRES", 1011, 1012, 1013, 1014);
+ MockEndpoint result = getMockEndpoint("mock:triggered");
+ result.expectedMessageCount(2);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("triggered");
+
+ result.assertIsSatisfied(10000);
+ assertEquals(List.of(Map.of("temperature", 22, "pressure", 1012),
Map.of("temperature", 24, "pressure", 1014)),
+ result.getExchanges().stream().map(exchange ->
exchange.getIn().getBody()).toList());
+ }
+
+ @Test
+ public void reportsEachFailedPollAndResumesWhenThePlcReturns() throws
Exception {
+ MockDriver.programRead("TRIG", false, true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockDriver.refuseConnections();
+ MockEndpoint result = getMockEndpoint("mock:reported");
+ result.expectedMessageCount(1);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("reported");
+
assertTrue(context.getRouteController().getRouteStatus("reported").isStarted());
+ await().atMost(10, TimeUnit.SECONDS).until(() -> pollFailures.count()
>= 3);
+ assertEquals(0, result.getReceivedCounter(), "nothing is delivered
while the PLC refuses connections");
+
+ MockDriver.acceptConnections();
+
+ result.assertIsSatisfied(10000);
+
assertTrue(pollFailures.causes().stream().allMatch(PlcConnectionException.class::isInstance),
+ () -> "Unexpected causes: " + pollFailures.causes());
+ }
+
+ @Test
+ public void pollsOnTheOneConnectionOfTheEndpoint() throws Exception {
+ MockDriver.programRead("TRIG", false, false, false, false, false,
true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:reported");
+ result.expectedMessageCount(1);
+
+ context.getRouteController().startRoute("reported");
+
+ result.assertIsSatisfied(10000);
+ assertEquals(1, MockDriver.connectionStrings().size(), "six polls use
the connection set up on start");
+ }
+
+ @Test
+ public void resumesOnANewConnectionWhenTheDroppedOneCannotConnectAgain()
throws Exception {
+ MockDriver.programRead("TRIG", false, true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:reported");
+ result.expectedMessageCount(2);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("reported");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
result.getReceivedCounter() == 1);
+ MockDriver.dropConnections();
+ MockDriver.programRead("TRIG", false, true);
+
+ result.assertIsSatisfied(10000);
+ }
+
+ @Test
+ public void aFailedPollForgetsTheTriggerSoAStillTrueTriggerEmitsAgain()
throws Exception {
+ MockDriver.programRead("TRIG", true, true, MockDriver.failedRead(),
true, true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:reported");
+ result.expectedMessageCount(2);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("reported");
+
+ result.assertIsSatisfied(10000);
+ assertEquals(1, pollFailures.count());
+ }
+
+ @Test
+ public void aTagReadWithAnErrorCodeFailsThePoll() throws Exception {
+ MockDriver.programRead("TRIG", false, true, true);
+ MockDriver.programRead("TEMP", 21,
MockDriver.responseCode(PlcResponseCode.NOT_FOUND), 23);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:reported");
+ result.expectedMessageCount(1);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("reported");
+
+ result.assertIsSatisfied(10000);
+ assertEquals(Map.of("temperature", 23, "pressure", 1013),
result.getExchanges().get(0).getIn().getBody());
+ assertEquals(1, pollFailures.count());
+
assertTrue(pollFailures.causes().get(0).getMessage().contains("NOT_FOUND"),
+ () -> "Unexpected cause: " + pollFailures.causes().get(0));
+ }
+
+ @Test
+ public void aTriggerThatIsNotABooleanFailsThePoll() throws Exception {
+ MockDriver.programRead("TRIG", 1, true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockEndpoint result = getMockEndpoint("mock:reported");
+ result.expectedMessageCount(1);
+ result.setAssertPeriod(1000);
+
+ context.getRouteController().startRoute("reported");
+
+ result.assertIsSatisfied(10000);
+ assertEquals(1, pollFailures.count(), () -> "Unexpected failures: " +
pollFailures.causes());
+ String message = pollFailures.causes().get(0).getMessage();
+ assertTrue(message.contains("TRIG") &&
message.contains(Integer.class.getName()), message);
+ }
+
+ @Test
+ public void aReadThatCompletesWhileTheRouteStopsIsNotDelivered() throws
Exception {
+ MockDriver.programRead("TRIG", true);
+ MockDriver.programRead("TEMP", 21);
+ MockDriver.programRead("PRES", 1013);
+ MockDriver.holdReads();
+ MockEndpoint result = getMockEndpoint("mock:reported");
+ result.expectedMessageCount(0);
+
+ context.getRouteController().startRoute("reported");
+ MockDriver.awaitHeldRead();
+ context.getRouteController().stopRoute("reported");
+ MockDriver.releaseReads();
+
+ // with no message expected, the mock endpoint waits the whole time
before it is satisfied
+ result.assertIsSatisfied(1000);
+ assertEquals(0, pollFailures.count(), () -> "Unexpected failures: " +
pollFailures.causes());
+ }
+
+ @Test
+ public void aRefusedConnectionFailsTheRouteStartWithoutAutoReconnect() {
+ MockDriver.refuseConnections();
+
+ Exception e = assertThrows(Exception.class,
+ () ->
context.getRouteController().startRoute("withoutAutoReconnect"));
+
+ assertNotNull(causeOfType(e, PlcConnectionException.class), () -> "No
PlcConnectionException in " + e);
+ }
+
+ @Test
+ public void aTagNamedLikeTheTriggerFailsTheRouteStart() {
+ Exception e = assertThrows(Exception.class, () ->
context.getRouteController().startRoute("reservedTagName"));
+
+ Throwable cause = causeOfType(e, IllegalArgumentException.class);
+ assertNotNull(cause, () -> "No IllegalArgumentException in " + e);
+ assertTrue(cause.getMessage().contains(Constants.TRIGGER),
cause::getMessage);
+ }
+
+ @Test
+ public void aTriggerWithoutAPeriodFailsTheRouteStart() {
+ Exception e = assertThrows(Exception.class, () ->
context.getRouteController().startRoute("withoutPeriod"));
+
+ Throwable cause = causeOfType(e, IllegalArgumentException.class);
+ assertNotNull(cause, () -> "No IllegalArgumentException in " + e);
+ assertTrue(cause.getMessage().contains("period"), cause::getMessage);
+ }
+
+ @Test
+ public void handsTheDriverOnlyItsOwnConnectionParameters() throws
Exception {
+ MockDriver.programRead("TRIG", true);
+ MockDriver.programRead("TEMP", 21);
+ MockEndpoint result = getMockEndpoint("mock:driverParameters");
+ result.expectedMessageCount(1);
+
+ context.getRouteController().startRoute("driverParameters");
+
+ result.assertIsSatisfied(10000);
+
assertEquals(Set.of("mock:10.10.10.1/1/1?percent-encoded=%25DB1&remote-rack=1&request-timeout-ms=500"),
+ Set.copyOf(MockDriver.connectionStrings()));
+ String uri = context.getEndpoint(DRIVER_PARAMETERS,
Plc4XEndpoint.class).getUri();
+ assertTrue(uri.contains("trigger=TRIG") &&
uri.contains("autoReconnect=true"),
+ () -> "The endpoint no longer reports its configured uri: " +
uri);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(TRIGGERED).routeId("triggered").autoStartup(false).to("mock:triggered");
+
from(UNTRIGGERED).routeId("untriggered").autoStartup(false).to("mock:untriggered");
+
from(REPORTED).routeId("reported").autoStartup(false).to("mock:reported");
+
from(WITHOUT_AUTO_RECONNECT).routeId("withoutAutoReconnect").autoStartup(false)
+ .to("mock:withoutAutoReconnect");
+
from(RESERVED_TAG_NAME).routeId("reservedTagName").autoStartup(false).to("mock:reservedTagName");
+
from(WITHOUT_PERIOD).routeId("withoutPeriod").autoStartup(false).to("mock:withoutPeriod");
+
from(DRIVER_PARAMETERS).routeId("driverParameters").autoStartup(false).to("mock:driverParameters");
+ }
+ };
+ }
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("pollFailures", pollFailures);
+ }
+
+ private static Throwable causeOfType(Throwable exception, Class<? extends
Throwable> type) {
+ Throwable cause = exception;
+ while (cause != null && !type.isInstance(cause)) {
+ cause = cause.getCause();
+ }
+ return cause;
+ }
+
+ /**
+ * Collects the failures the consumer reports to its exception handler.
+ */
+ private static final class PollFailures implements ExceptionHandler {
+
+ private final List<Throwable> causes = new CopyOnWriteArrayList<>();
+
+ @Override
+ public void handleException(Throwable exception) {
+ causes.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ causes.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ causes.add(exception);
+ }
+
+ int count() {
+ return causes.size();
+ }
+
+ List<Throwable> causes() {
+ return causes;
+ }
+ }
+
+}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerStopTest.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerStopTest.java
deleted file mode 100644
index f2362156a147..000000000000
---
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerStopTest.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.camel.component.plc4x;
-
-import java.util.Map;
-
-import org.apache.camel.Processor;
-import org.apache.camel.impl.DefaultCamelContext;
-import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScraperImpl;
-import
org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollectorImpl;
-import org.junit.jupiter.api.Test;
-import org.mockito.MockedConstruction;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockConstruction;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-/**
- * A stopped triggered consumer must stop its scraper, or the scraper keeps
reading the PLC and sending exchanges to the
- * route, also after the route is started again with a new consumer
(duplicates).
- */
-class Plc4XConsumerStopTest {
-
- @Test
- void testStopStopsTheScraper() throws Exception {
- Plc4XEndpoint endpoint = mock(Plc4XEndpoint.class);
- when(endpoint.getCamelContext()).thenReturn(new DefaultCamelContext());
-
when(endpoint.getTrigger()).thenReturn("(PLC4X_TRIGGER_VAR,10,(%DB1:DBX0.0:BOOL)==(true))");
- when(endpoint.getTags()).thenReturn(Map.of("tag1", "%DB1:DBW2:INT"));
- when(endpoint.getPeriod()).thenReturn(1000);
- when(endpoint.getUri()).thenReturn("mock:plc");
-
- try (MockedConstruction<TriggeredScraperImpl> scrapers =
mockConstruction(TriggeredScraperImpl.class);
- MockedConstruction<TriggerCollectorImpl> collectors =
mockConstruction(TriggerCollectorImpl.class)) {
- Plc4XConsumer consumer = new Plc4XConsumer(endpoint,
mock(Processor.class));
- consumer.start();
-
- assertEquals(1, scrapers.constructed().size());
- TriggeredScraperImpl scraper = scrapers.constructed().get(0);
- TriggerCollectorImpl collector = collectors.constructed().get(0);
- verify(scraper).start();
- verify(scraper, never()).stop();
-
- consumer.stop();
-
- verify(scraper).stop();
- verify(collector).stop();
- }
- }
-}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerTest.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerTest.java
deleted file mode 100644
index fa595463a25e..000000000000
---
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XConsumerTest.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.camel.component.plc4x;
-
-import java.util.Collections;
-import java.util.Map;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.impl.DefaultCamelContext;
-import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.*;
-import static org.mockito.Mockito.*;
-
-class Plc4XConsumerTest {
-
- private Plc4XEndpoint endpoint;
- private Processor processor;
- private Plc4XConsumer consumer;
-
- @BeforeEach
- void setUp() {
- endpoint = mock(Plc4XEndpoint.class);
- processor = mock(Processor.class);
-
- when(endpoint.getTags()).thenReturn(Collections.emptyMap());
- when(endpoint.getTrigger()).thenReturn(null); // untriggered
- when(endpoint.getCamelContext()).thenReturn(new DefaultCamelContext());
-
- consumer = new Plc4XConsumer(endpoint, processor);
- }
-
- @Test
- void doStart() throws Exception {
-
- doNothing().when(endpoint).setupConnection();
- consumer.doStart();
- verify(endpoint, times(1)).setupConnection();
- }
-
- @Test
- void doStartBadStart() throws Exception {
-
- doThrow(new PlcConnectionException("fail"))
- .when(endpoint).setupConnection();
- consumer.doStart();
- verify(endpoint).setupConnection();
- assertFalse(consumer.isStarted());
- }
-
- @Test
- void shouldStartTriggeredScraperWhenTriggerPresent() throws Exception {
-
- when(endpoint.getTrigger()).thenReturn("someTrigger");
- when(endpoint.getTags()).thenReturn(Map.of("tag1", "value1"));
- when(endpoint.getPeriod()).thenReturn(1000);
- when(endpoint.getUri()).thenReturn("mock:plc");
-
- // Avoid real connection
- doNothing().when(endpoint).setupConnection();
- doNothing().when(endpoint).reconnectIfNeeded();
- when(endpoint.createExchange()).thenReturn(mock(Exchange.class));
- consumer.doStart();
- verify(endpoint, atLeastOnce()).getTrigger();
- }
-
- @Test
- void doStop() throws Exception {
- consumer.doStop();
- assertFalse(consumer.isStarted(), "Consumer should not be started
after doStop");
- }
-
-}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java
index 24cdd40c11d4..28df776b2c83 100644
---
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java
+++
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XEndpointTest.java
@@ -16,16 +16,21 @@
*/
package org.apache.camel.component.plc4x;
+import java.util.Map;
+
import org.apache.camel.Component;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.plc4x.java.api.PlcConnection;
+import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.Is.is;
import static org.hamcrest.core.IsNull.notNullValue;
+import static org.hamcrest.core.IsNull.nullValue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.*;
class Plc4XEndpointTest {
@@ -56,12 +61,26 @@ class Plc4XEndpointTest {
}
@Test
- void doStopDisconnectedConnection() throws Exception {
+ void doStopClosesALostConnection() throws Exception {
PlcConnection plcConnectionMock = mock(PlcConnection.class);
+ // isConnected() returns false on the mock, and close() fails as on a
dead transport
+ doThrow(new PlcConnectionException("Transport is not
open")).when(plcConnectionMock).close();
sut.connection = plcConnectionMock;
- // isConnected() returns false by default on the mock, so doStop()
should skip close()
+
sut.doStop();
- verify(plcConnectionMock, never()).close();
+
+ verify(plcConnectionMock).close();
+ assertThat(sut.connection, nullValue());
+ }
+
+ @Test
+ void buildPlcReadRequestWithoutAConnection() {
+ assertThrows(PlcConnectionException.class, () ->
sut.buildPlcReadRequest());
+ }
+
+ @Test
+ void buildPlcWriteRequestWithoutAConnection() {
+ assertThrows(PlcConnectionException.class, () ->
sut.buildPlcWriteRequest(Map.of()));
}
}
diff --git
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XFailureTest.java
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XFailureTest.java
index 54ac03fe2253..e1fd611d8f82 100644
---
a/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XFailureTest.java
+++
b/components/camel-plc4x/src/test/java/org/apache/camel/component/plc4x/Plc4XFailureTest.java
@@ -24,7 +24,6 @@ import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.support.DefaultExchange;
-import org.apache.plc4x.java.api.PlcConnection;
import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
import org.apache.plc4x.java.api.messages.PlcReadRequest;
@@ -111,7 +110,6 @@ class Plc4XFailureTest {
PlcRuntimeException cause = new PlcRuntimeException("read failed");
PlcReadRequest request = mock(PlcReadRequest.class);
when(request.execute()).thenAnswer(invocation ->
CompletableFuture.failedFuture(cause));
- endpoint.connection = mock(PlcConnection.class);
when(endpoint.buildPlcReadRequest()).thenReturn(request);
Exchange exchange = new Plc4XPollingConsumer(endpoint).receive(1000);
@@ -124,7 +122,6 @@ class Plc4XFailureTest {
void testPollingConsumerReturnsNullWithoutResponse() throws Exception {
PlcReadRequest request = mock(PlcReadRequest.class);
when(request.execute()).thenAnswer(invocation -> new
CompletableFuture<>());
- endpoint.connection = mock(PlcConnection.class);
when(endpoint.buildPlcReadRequest()).thenReturn(request);
// receiveNoWait and receive(timeout) return null when nothing is
received in time
@@ -134,7 +131,6 @@ class Plc4XFailureTest {
@Test
void testPollingConsumerReadsTags() throws Exception {
- endpoint.connection = mock(PlcConnection.class);
PlcReadRequest request = mock(PlcReadRequest.class,
RETURNS_DEEP_STUBS);
when(endpoint.buildPlcReadRequest()).thenReturn(request);
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index dac0f26b258c..0be69c592ec1 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -3600,3 +3600,47 @@ Use `0.0.0.0` as the host to accept connections on all
network interfaces.
`AvroRpcHttpServerFactory` has a new method `create(SpecificResponder, String,
int)` that receives the host. A custom factory should implement it.
+=== camel-plc4x - PLC4X 1.0.0 and the triggered consumer
+
+`camel-plc4x` now uses PLC4X 1.0.0, which requires Java 21, as Camel 4.24 does.
+
+PLC4X 1.0.0 changes the connection strings of its drivers. Several parameters
are renamed, for example
+`request-timeout` to `request-timeout-ms`, and the S7 driver reads the rack
and slot as `cotp.remote-rack` and
+`cotp.remote-slot` instead of `remote-rack` and `remote-slot`. A parameter
that a driver does not know, such as an old
+name, is logged as a warning and does not apply. The OPC UA driver now uses
the security policy `Basic256Sha256` by
+default instead of `NONE`, `insecure-certificate-verification=true` becomes
`tls.verify=false`, and a user name and
+password are refused over a channel that is neither signed nor encrypted
unless `allow-insecure-credentials` is set.
+See the incompatible changes in the
https://github.com/apache/plc4x/releases/tag/v1.0.0[PLC4X 1.0.0 release notes]
+for the full list.
+
+PLC4X 1.0.0 writes the array selection of a tag address before the data type,
and as a range instead of a count. An
+address written as `holding-register:1:INT[4]` or `%DB42:28.0:BYTE[4]` for
four elements is now written
+`holding-register:1[0..3]:INT` or `%DB42:28.0[0..3]:BYTE`; `[4]` before the
data type selects the single element at
+index 4. Addresses in the old form no longer parse. This applies to the
`tag.*` options, to `trigger` and to the
+addresses in the body of the producer.
+
+The S7 driver of PLC4X 1.0.0 no longer includes the `s7-light` driver, so
`plc4x:s7-light://` endpoints no longer find
+a driver. Use `plc4x:s7://` instead.
+
+A consumer with `trigger` now reads the trigger tag and the tags in one
request on the connection of the endpoint,
+every `period` milliseconds from the start of the route. Before, it sampled
the trigger first after about one second,
+and read the tags in a later, separate request on another connection. A failed
read is now reported to the exception
+handler of the consumer (or to the error handler of the route with
`bridgeErrorHandler=true`), and polling continues;
+before, the consumer stopped triggering after the first failed read of the
trigger. After a failed poll the remembered
+trigger value is unknown, so a trigger that is still `true` when the PLC
returns emits once. Since the consumer now uses
+the connection of the endpoint, `autoReconnect` governs its reconnects. A
connection that cannot be set up when the
+route starts fails the start of the route, unless `autoReconnect=true`, in
which case the polls retry it. A route that
+starts with the CamelContext then fails the start of the CamelContext too,
unless the route is supervised by the
+xref:route-controller.adoc[supervising route controller] or has
`autoStartup=false`. As before, `period` must be
+positive when `trigger` is set. A trigger that reads as a value that is not a
boolean, such as a number or an array,
+fails the poll. See the xref:components::plc4x-component.adoc[PLC4X component]
for how triggered reading works.
+
+With `autoReconnect=true`, a lost connection is now closed and replaced with a
new one from the driver instead of being
+reconnected, because the PLC4X 1.0.0 S7 and EtherNet/IP drivers cannot connect
a connection again once it is lost. This
+applies to the producer and to the consumers alike.
+
+The component now removes its own options (`trigger`, `period`, `tag.*`,
`autoReconnect` and the generic consumer
+options such as `exceptionHandler`) from the connection string it hands to the
driver. Before, the driver received them
+as connection string parameters too.
+
+The constant `Constants.PLC_NAME`, which was only used with the PLC4X scraper,
was removed.
diff --git a/parent/pom.xml b/parent/pom.xml
index 2646cc902415..2e25b597926c 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -448,7 +448,7 @@
<phase2-lib-version>6.2.2</phase2-lib-version>
<picocli-version>4.7.7</picocli-version>
<pinecone-client-version>3.1.0</pinecone-client-version>
- <plc4x-version>0.13.1</plc4x-version>
+ <plc4x-version>1.0.0</plc4x-version>
<plexus-archiver-version>4.14.0</plexus-archiver-version>
<plexus-build-api-version>1.2.0</plexus-build-api-version>
<plexus-container-default-version>2.1.1</plexus-container-default-version>
