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 294a7b6998822..37f10570ac510 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` 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 32f826e824f05..0bea7af6f9ebe 100644 --- a/components/camel-plc4x/pom.xml +++ b/components/camel-plc4x/pom.xml @@ -44,11 +44,6 @@ plc4j-api ${plc4x-version} - - org.apache.plc4x - plc4j-scraper - ${plc4x-version} - diff --git a/components/camel-plc4x/src/main/docs/plc4x-component.adoc b/components/camel-plc4x/src/main/docs/plc4x-component.adoc index 294a7b6998822..37f10570ac510 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` 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 d973e4e37ddd7..6f995dde53fa0 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 581191a7929d8..bb4d8dbe76861 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.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 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 @@ protected void doStart() throws Exception { } 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 @@ private void startUnTriggered() { future = executorService.schedule(() -> request.execute().thenAccept(response -> { try { Exchange exchange = plc4XEndpoint.createExchange(); - Map 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 @@ private void startUnTriggered() { }), 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 tagList) { - String config = "(TRIGGER_VAR," + plc4XEndpoint.getPeriod() + ",(" + plc4XEndpoint.getTrigger() + ")==(true))"; - List job = Collections.singletonList( - new JobConfigurationImpl("PLC4X-Camel", config, 0, Collections.singletonList(Constants.PLC_NAME), tagList)); - Map 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 @@ protected void doStop() throws Exception { 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 55fd8e9238816..c541fb8998018 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.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 @@ 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 boolean isAutoReconnect() { */ 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 void setupConnection() throws PlcConnectionException { * @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 void reconnectIfNeeded() throws PlcConnectionException { * @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 Producer createProducer() { @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 PollingConsumer createPollingConsumer() { /** * 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 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 tag : tags.entrySet()) { try { @@ -228,15 +275,32 @@ public PlcReadRequest buildPlcReadRequest() { return builder.build(); } + /** + * @return the values of a read response by tag name + */ + static Map tagValues(PlcReadResponse response) { + Map values = new HashMap<>(); + for (String tag : response.getTagNames()) { + values.put(tag, response.getObject(tag)); + } + return values; + } + /** * Build a {@link PlcWriteRequest}. *

* - * @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> tags) { - PlcWriteRequest.Builder builder = connection.writeRequestBuilder(); + public PlcWriteRequest buildPlcWriteRequest(Map> 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> entry : tags.entrySet()) { //Tags are stored like this --> Map> for writing @@ -271,16 +335,34 @@ public void setTags(Map tags) { @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 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 6fcf7242244f1..b35c2d306e6ea 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 @@ protected Exchange doReceive(long timeout) { Exchange exchange = plc4XEndpoint.createExchange(); try { plc4XEndpoint.reconnectIfNeeded(); - if (plc4XEndpoint.connection == null) { - throw new PlcConnectionException("Cannot establish connection"); - } - PlcReadRequest request = plc4XEndpoint.buildPlcReadRequest(); CompletableFuture future = request.execute().whenComplete((plcReadResponse, throwable) -> { @@ -101,11 +95,7 @@ protected Exchange doReceive(long timeout) { response = future.get(); } - Map 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 6b82e831ead0e..13719137f77e1 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 void process(Exchange exchange) throws Exception { 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 0000000000000..a742ac2a878b2 --- /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 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 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 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 71815047e3a75..a27f979dc6f32 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 PROGRAMMED_TAGS = new ConcurrentHashMap<>(); + private static volatile CompletableFuture heldReads = CompletableFuture.completedFuture(null); + private static volatile CompletableFuture heldRead = new CompletableFuture<>(); + private static final AtomicInteger REFUSED_CONNECTIONS = new AtomicInteger(); + private static final List 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 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 String getProtocolName() { @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 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 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 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 read(Map tags) { + heldRead.complete(null); + // join() does not return early on an interrupt + heldReads.join(); + Map 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. getArgument(0))); + case "getResponseCode" -> values.get(invocation. 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 values; + private int next; + + ProgrammedTag(List 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 0000000000000..6a94252023fda --- /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 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 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 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 f2362156a1479..0000000000000 --- 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 scrapers = mockConstruction(TriggeredScraperImpl.class); - MockedConstruction 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 fa595463a25ee..0000000000000 --- 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 24cdd40c11d43..28df776b2c83f 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 @@ public void isSingleton() { } @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 54ac03fe2253a..e1fd611d8f821 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.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 @@ void testPollingConsumerFailsWhenReadFails() throws Exception { 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 @@ void testPollingConsumerFailsWhenReadFails() throws Exception { 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 @@ void testPollingConsumerReturnsNullWithoutResponse() throws Exception { @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 9af133837d120..ff0a3a7a7a923 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 @@ -3582,3 +3582,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 853bcdb2b7665..98ba4df630507 100644 --- a/parent/pom.xml +++ b/parent/pom.xml @@ -448,7 +448,7 @@ 6.2.2 4.7.7 3.1.0 - 0.13.1 + 1.0.0 4.14.0 1.2.0 2.1.1