Chore/update asf yml

.asf.yaml

@@ -76,14 +76,15 @@ github:
         excludes: []
       bypass_teams:
         - root
+        - plc4x-committers
       restrict_deletion: true
       restrict_force_push: true
       required_pull_request_reviews:
         dismiss_stale_reviews: true
-        require_last_push_approval: false
+        require_last_push_approval: true
         required_approving_review_count: 1
       required_linear_history: false
-      required_signatures: true
+      required_signatures: false
       required_conversation_resolution: true
 
 notifications:

plc4j/integrations/apache-calcite/pom.xml

@@ -57,7 +57,7 @@
     </dependency>
     <dependency>
       <groupId>org.apache.plc4x</groupId>
-      <artifactId>plc4j-scraper</artifactId>
+      <artifactId>plc4j-tools-event-pump</artifactId>
       <version>${plc4x.version}</version>
     </dependency>
     <dependency>

plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xBaseTable.java

@@ -29,7 +29,7 @@
 import org.apache.calcite.schema.impl.AbstractTable;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
-import org.apache.plc4x.java.scraper.config.JobConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,12 +53,12 @@ public abstract class Plc4xBaseTable extends AbstractTable {
     private static final Logger logger = LoggerFactory.getLogger(Plc4xBaseTable.class);
 
     private final BlockingQueue<Plc4xSchema.Record> queue;
-    private final JobConfiguration conf;
+    private final BatchConfiguration conf;
     private final long tableCutoff;
     private Plc4xSchema.Record current;
     private final List<String> names;
 
-    public Plc4xBaseTable(BlockingQueue<Plc4xSchema.Record> queue, JobConfiguration conf, long tableCutoff) {
+    public Plc4xBaseTable(BlockingQueue<Plc4xSchema.Record> queue, BatchConfiguration conf, long tableCutoff) {
         this.tableCutoff = tableCutoff;
         logger.info("Instantiating new PLC4X Table with configuration: {}", conf);
         this.queue = queue;
@@ -107,7 +107,7 @@ public RelDataType getRowType(RelDataTypeFactory typeFactory) {
         } catch (ExecutionException | TimeoutException e) {
             throw new PlcRuntimeException("Unable to fetch first record and infer arguments!", e);
         }
-        logger.info("Inferring types for Table '{}' based on values: {}", conf.getName(), first.values);
+        logger.info("Inferring types for Table '{}' based on values: {}", conf.getId(), first.values);
         // Extract types
         List<RelDataType> types = names.stream()
             .map(n -> {

plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchema.java

@@ -22,12 +22,12 @@
 import org.apache.calcite.schema.impl.AbstractSchema;
 import org.apache.plc4x.java.DefaultPlcDriverManager;
 import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
-import org.apache.plc4x.java.scraper.ResultHandler;
-import org.apache.plc4x.java.scraper.Scraper;
-import org.apache.plc4x.java.scraper.ScraperImpl;
-import org.apache.plc4x.java.scraper.config.JobConfiguration;
-import org.apache.plc4x.java.scraper.config.ScraperConfiguration;
-import org.apache.plc4x.java.scraper.exception.ScraperException;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.tools.eventpump.EventPump;
+import org.apache.plc4x.java.tools.eventpump.TagBatch;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.EventPumpConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.EventPumpFactory;
 import org.apache.plc4x.java.utils.cache.CachedPlcConnectionManager;
 
 import java.time.Instant;
@@ -38,36 +38,43 @@
 
 public class Plc4xSchema extends AbstractSchema {
 
-    protected final ScraperConfiguration configuration;
-    protected final Scraper scraper;
+    protected final EventPumpConfiguration configuration;
+    protected final EventPump eventPump;
     protected final QueueHandler handler;
     protected final Map<String, BlockingQueue<Record>> queues;
     protected final Map<String, Table> tableMap;
+    /** batch id -&gt; connection id, so a record can be attributed to the PLC it came from. */
+    protected final Map<String, String> connectionIds;
 
-    public Plc4xSchema(ScraperConfiguration configuration, long tableCutoff) throws ScraperException {
+    public Plc4xSchema(EventPumpConfiguration configuration, long tableCutoff) throws Exception {
         this.configuration = configuration;
         this.handler = new QueueHandler();
-        this.scraper = new ScraperImpl(handler,
-            CachedPlcConnectionManager.getBuilder()
-                .withConnectionManager(new DefaultPlcDriverManager())
-                .build(),
-            configuration.getJobs());
-        this.queues = configuration.getJobConfigurations().stream()
+        this.connectionIds = configuration.getBatches().stream()
+            .collect(Collectors.toMap(
+                BatchConfiguration::getId,
+                BatchConfiguration::getConnectionId
+            ));
+        this.queues = configuration.getBatches().stream()
             .collect(Collectors.toMap(
-                JobConfiguration::getName,
+                BatchConfiguration::getId,
                 conf -> new ArrayBlockingQueue<>(1000)
             ));
-        // Create the tables
-        this.tableMap = configuration.getJobConfigurations().stream()
+        // Create the tables - one per batch
+        this.tableMap = configuration.getBatches().stream()
             .collect(Collectors.toMap(
-                JobConfiguration::getName,
-                conf -> defineTable(queues.get(conf.getName()), conf, tableCutoff)
+                BatchConfiguration::getId,
+                conf -> defineTable(queues.get(conf.getId()), conf, tableCutoff)
             ));
-        // Start the scraper
-        this.scraper.start();
+        // Every batch reports to the same handler, which routes by batch id
+        this.eventPump = EventPumpFactory.create(configuration,
+            CachedPlcConnectionManager.getBuilder()
+                .withConnectionManager(new DefaultPlcDriverManager())
+                .build(),
+            handler);
+        this.eventPump.startAll();
     }
 
-    Table defineTable(BlockingQueue<Record> queue, JobConfiguration configuration, Long limit) {
+    Table defineTable(BlockingQueue<Record> queue, BatchConfiguration configuration, Long limit) {
         if (limit <= 0) {
             return new Plc4xStreamTable(queue, configuration);
         } else {
@@ -94,13 +101,16 @@ public Record(Instant timestamp, String source, Map<String, Object> values) {
         }
     }
 
-    class QueueHandler implements ResultHandler {
+    class QueueHandler implements TagBatch.TagBatchListener {
 
         @Override
-        public void handle(String job, String alias, Map<String, Object> results) {
+        public void onTagsFetched(TagBatch batch, PlcReadResponse response) {
+            String batchId = batch.getBatchId();
+            Map<String, Object> results = response.getTagNames().stream()
+                .collect(Collectors.toMap(name -> name, response::getObject));
             try {
-                Record record = new Record(Instant.now(), alias, results);
-                queues.get(job).put(record);
+                Record record = new Record(Instant.now(), connectionIds.get(batchId), results);
+                queues.get(batchId).put(record);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new PlcRuntimeException("Handling got interrupted", e);

plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchemaFactory.java

@@ -22,12 +22,11 @@
 import org.apache.calcite.schema.SchemaFactory;
 import org.apache.calcite.schema.SchemaPlus;
 import org.apache.commons.lang3.Validate;
-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.tools.eventpump.config.EventPumpConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.Map;
 
@@ -40,9 +39,17 @@ public Schema create(SchemaPlus parentSchema, String name, Map<String, Object> o
         Object config = operand.get("config");
         Validate.notNull(config, "No configuration file given. Please specify operand 'config'...'");
         // Load configuration from file
-        ScraperConfiguration configuration;
+        EventPumpConfiguration configuration;
+        String configPath = config.toString();
         try {
-            configuration = ScraperConfiguration.fromFile(config.toString(), ScraperConfigurationTriggeredImpl.class);
+            File configFile = new File(configPath);
+            if (configPath.endsWith(".json")) {
+                configuration = EventPumpConfiguration.fromJson(configFile);
+            } else if (configPath.endsWith(".xml")) {
+                configuration = EventPumpConfiguration.fromXml(configFile);
+            } else {
+                configuration = EventPumpConfiguration.fromYaml(configFile);
+            }
         } catch (IOException e) {
             throw new IllegalArgumentException("Unable to load configuration file!", e);
         }
@@ -59,8 +66,8 @@ public Schema create(SchemaPlus parentSchema, String name, Map<String, Object> o
         // Pass the configuration to the Schema
         try {
             return new Plc4xSchema(configuration, parsedLimit);
-        } catch (ScraperException e) {
-            LOGGER.warn("Could not evaluate Plc4xSchema",e);
+        } catch (Exception e) {
+            LOGGER.warn("Could not evaluate Plc4xSchema", e);
             //ToDo Exception, but interface does not accept ... null is fishy
             return null;
         }

plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xStreamTable.java

@@ -23,13 +23,13 @@
 import org.apache.calcite.schema.ScannableTable;
 import org.apache.calcite.schema.StreamableTable;
 import org.apache.calcite.schema.Table;
-import org.apache.plc4x.java.scraper.config.JobConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
 
 import java.util.concurrent.BlockingQueue;
 
 public class Plc4xStreamTable extends Plc4xBaseTable implements ScannableTable, StreamableTable {
 
-    public Plc4xStreamTable(BlockingQueue<Plc4xSchema.Record> queue, JobConfiguration conf) {
+    public Plc4xStreamTable(BlockingQueue<Plc4xSchema.Record> queue, BatchConfiguration conf) {
         super(queue, conf, -1L);
     }
 

plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xTable.java

@@ -21,13 +21,13 @@
 import org.apache.calcite.DataContext;
 import org.apache.calcite.linq4j.Enumerable;
 import org.apache.calcite.schema.ScannableTable;
-import org.apache.plc4x.java.scraper.config.JobConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
 
 import java.util.concurrent.BlockingQueue;
 
 public class Plc4xTable extends Plc4xBaseTable implements ScannableTable {
 
-    public Plc4xTable(BlockingQueue<Plc4xSchema.Record> queue, JobConfiguration conf, long tableCutoff) {
+    public Plc4xTable(BlockingQueue<Plc4xSchema.Record> queue, BatchConfiguration conf, long tableCutoff) {
         super(queue, conf, tableCutoff);
     }
 

plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/DriverManagerTest.java

@@ -20,12 +20,11 @@
 
 import org.apache.calcite.jdbc.CalciteConnection;
 import org.apache.calcite.jdbc.Driver;
-import org.apache.plc4x.java.scraper.config.ScraperConfiguration;
-import org.apache.plc4x.java.scraper.config.ScraperConfigurationClassicImpl;
-import org.apache.plc4x.java.scraper.exception.ScraperException;
+import org.apache.plc4x.java.tools.eventpump.config.EventPumpConfiguration;
 import org.assertj.core.api.WithAssertions;
 import org.junit.jupiter.api.Test;
 
+import java.io.File;
 import java.io.IOException;
 import java.sql.Connection;
 import java.sql.ResultSet;
@@ -36,12 +35,12 @@
 public class DriverManagerTest implements WithAssertions {
 
     @Test
-    void query() throws SQLException, IOException, ScraperException {
+    void query() throws Exception {
         Driver driver = new Driver();
         Connection connection = driver.connect("jdbc:calcite:asdf;lex=MYSQL_ANSI", new Properties());
 
         CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);
-        calciteConnection.getRootSchema().add("plc4x", new Plc4xSchema(ScraperConfiguration.fromFile("src/test/resources/example.yml", ScraperConfigurationClassicImpl.class), 100));
+        calciteConnection.getRootSchema().add("plc4x", new Plc4xSchema(EventPumpConfiguration.fromYaml(new File("src/test/resources/example.yml")), 100));
 
         ResultSet rs = connection.prepareStatement("SELECT * FROM \"plc4x\".\"job1\"").executeQuery();
         validateResult(rs);

plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/Plc4XBaseTableTest.java

@@ -19,7 +19,7 @@
 package org.apache.plc4x;
 
 import org.apache.calcite.linq4j.Enumerator;
-import org.apache.plc4x.java.scraper.config.JobConfigurationImpl;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
 import org.assertj.core.api.WithAssertions;
 import org.junit.jupiter.api.Test;
 
@@ -33,12 +33,10 @@ class Plc4XBaseTableTest implements WithAssertions {
     @Test
     void testOnBlockingQueue() {
         ArrayBlockingQueue<Plc4xSchema.Record> queue = new ArrayBlockingQueue<>(100);
-        Plc4xStreamTable table = new Plc4xStreamTable(queue, new JobConfigurationImpl(
-            "job1",
-            null,
-            100,
-            Collections.emptyList(),
-            Collections.singletonMap("key", "address")));
+        BatchConfiguration conf = new BatchConfiguration();
+        conf.setId("job1");
+        conf.setSimpleTags(Collections.singletonMap("key", "address"));
+        Plc4xStreamTable table = new Plc4xStreamTable(queue, conf);
 
         Map<String, Object> objects = Collections.singletonMap("key", "value");
         queue.add(new Plc4xSchema.Record(Instant.now(), "", objects));

plc4j/integrations/apache-calcite/src/test/resources/example.yml

@@ -17,16 +17,28 @@
 # under the License.
 # ----------------------------------------------------------------------------
 ---
-sources:
-  test: simulated:test
-  test2: simulated:test2
+connections:
+  - id: test
+    url: simulated:test
+  - id: test2
+    url: simulated:test2
 
-jobs:
-  - name: job1
-    scrapeRate: 10
-    sources:
-    - test
-    - test2
+# Every batch becomes one table, named after the batch id.
+batches:
+  - id: job1
+    connectionId: test
     tags:
       test: 'RANDOM/test:DINT'
       test2: 'RANDOM/test:STRING'
+    trigger:
+      type: timer
+      intervalMillis: 10
+
+  - id: job2
+    connectionId: test2
+    tags:
+      test: 'RANDOM/test:DINT'
+      test2: 'RANDOM/test:STRING'
+    trigger:
+      type: timer
+      intervalMillis: 10

plc4j/integrations/apache-kafka/README.md

@@ -203,7 +203,8 @@ If an error occurs when reading or writing PLC addresses a graceful backoff has
 bombarded with requests. However as the number of connectors for each PLC should be limited to reduce the load on the PLC,
 the graceful backoff shouldn't have a major impact.
 
-For the source connector the PLC4X scraper logic is able to handle randomized polling rates on failures, this is buffered within the
+For the source connector the PLC4X event-pump backs off exponentially on failures (1s, doubling up to 60s) and skips a poll
+if the previous read is still running, this is buffered within the
 connector, the poll rate of the connector has no affect on the PLC poll rate.
 
 For the sink connector, if a write fails it is retried a configurable number of times with a timeout between each time.

plc4j/integrations/apache-kafka/config/plc4x-source.properties

@@ -32,7 +32,7 @@ sources.machineA.connectionString=simulated://127.0.0.1
 #This value controls how often it returns when no messages are received.
 sources.machineA.pollReturnInterval=5000
 
-#There is an internal buffer between the PLC4X scraper and Kafka Connect.
+#There is an internal buffer between the PLC4X event-pump and Kafka Connect.
 #This is the size of that buffer.
 sources.machineA.bufferSize=1000
 
@@ -45,7 +45,7 @@ sources.machineA.jobReferences.simulated-heartbeat.topic=simulated-heartbeat-top
 #A list of jobs specified in the following section.
 jobs=simulated-dashboard,simulated-heartbeat
 
-#The poll rate for this job. the PLC4X scraper will request data every interval (ms).
+#The poll rate for this job. The PLC4X event-pump will request data every interval (ms).
 jobs.simulated-dashboard.interval=1000
 
 #A list of tags. Each tag is a map between an alias and a PLC4X address.

plc4j/integrations/apache-kafka/pom.xml

@@ -157,7 +157,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.plc4x</groupId>
-            <artifactId>plc4j-scraper</artifactId>
+            <artifactId>plc4j-tools-event-pump</artifactId>
             <version>${plc4x.version}</version>
         </dependency>
         <dependency>

plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java

@@ -57,7 +57,7 @@ public Class<? extends Task> taskClass() {
     @Override
     public List<Map<String, String>> taskConfigs(int maxTasks) {
         // Initially we planned to have the simple assumption that one task maps to one PLC connection.
-        // But we could easily say that one scraper instance maps to a task and one scraper task can
+        // But we could easily say that one event-pump instance maps to a task and one batch can
         // process multiple PLC connections. But I guess this would be an optimization as we have to
         // balance the load manually.
         if(sourceConfig.getJobs().size() > maxTasks) {
@@ -66,8 +66,8 @@ public List<Map<String, String>> taskConfigs(int maxTasks) {
             return Collections.emptyList();
         }
 
-        // For each configured source we'll start a dedicated scraper instance collecting
-        // all the scraper jobs enabled for this source.
+        // For each configured source we'll start a dedicated event-pump instance collecting
+        // all the jobs enabled for this source.
         List<Map<String, String>> configs = new LinkedList<>();
         for (Source source : sourceConfig.getSources()) {
             // Build a list of job configurations only containing the ones referenced from