CAMEL-25617: camel-plc4x - Upgrade to PLC4X 1.0.0

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.

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>

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.

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!");

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<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 @@ 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,51 +87,14 @@ private void startUnTriggered() {
         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);
             }
         }), 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 @@ 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();
     }