diff --git a/.asf.yaml b/.asf.yaml
index 4a1d4e77..322a356e 100644
--- a/.asf.yaml
+++ b/.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:
diff --git a/plc4j/integrations/apache-calcite/pom.xml b/plc4j/integrations/apache-calcite/pom.xml
index 9b2b79ac..c794faee 100644
--- a/plc4j/integrations/apache-calcite/pom.xml
+++ b/plc4j/integrations/apache-calcite/pom.xml
@@ -57,7 +57,7 @@
org.apache.plc4x
- plc4j-scraper
+ plc4j-tools-event-pump
${plc4x.version}
diff --git a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xBaseTable.java b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xBaseTable.java
index 625e4d39..445c185e 100644
--- a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xBaseTable.java
+++ b/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 queue;
- private final JobConfiguration conf;
+ private final BatchConfiguration conf;
private final long tableCutoff;
private Plc4xSchema.Record current;
private final List names;
- public Plc4xBaseTable(BlockingQueue queue, JobConfiguration conf, long tableCutoff) {
+ public Plc4xBaseTable(BlockingQueue 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 types = names.stream()
.map(n -> {
diff --git a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchema.java b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchema.java
index 9f1b3ee3..6950a505 100644
--- a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchema.java
+++ b/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> queues;
protected final Map tableMap;
+ /** batch id -> connection id, so a record can be attributed to the PLC it came from. */
+ protected final Map 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 queue, JobConfiguration configuration, Long limit) {
+ Table defineTable(BlockingQueue 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 values) {
}
}
- class QueueHandler implements ResultHandler {
+ class QueueHandler implements TagBatch.TagBatchListener {
@Override
- public void handle(String job, String alias, Map results) {
+ public void onTagsFetched(TagBatch batch, PlcReadResponse response) {
+ String batchId = batch.getBatchId();
+ Map 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);
diff --git a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchemaFactory.java b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchemaFactory.java
index b2876107..92e4d2e5 100644
--- a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchemaFactory.java
+++ b/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 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 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;
}
diff --git a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xStreamTable.java b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xStreamTable.java
index 7c9668d9..840c8cac 100644
--- a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xStreamTable.java
+++ b/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 queue, JobConfiguration conf) {
+ public Plc4xStreamTable(BlockingQueue queue, BatchConfiguration conf) {
super(queue, conf, -1L);
}
diff --git a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xTable.java b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xTable.java
index 9c479a03..2b64b682 100644
--- a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xTable.java
+++ b/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 queue, JobConfiguration conf, long tableCutoff) {
+ public Plc4xTable(BlockingQueue queue, BatchConfiguration conf, long tableCutoff) {
super(queue, conf, tableCutoff);
}
diff --git a/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/DriverManagerTest.java b/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/DriverManagerTest.java
index c9f93233..8116acd0 100644
--- a/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/DriverManagerTest.java
+++ b/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);
diff --git a/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/Plc4XBaseTableTest.java b/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/Plc4XBaseTableTest.java
index 19c719a1..932e7ae0 100644
--- a/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/Plc4XBaseTableTest.java
+++ b/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 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 objects = Collections.singletonMap("key", "value");
queue.add(new Plc4xSchema.Record(Instant.now(), "", objects));
diff --git a/plc4j/integrations/apache-calcite/src/test/resources/example.yml b/plc4j/integrations/apache-calcite/src/test/resources/example.yml
index d90dd3c1..b0809a17 100644
--- a/plc4j/integrations/apache-calcite/src/test/resources/example.yml
+++ b/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
diff --git a/plc4j/integrations/apache-kafka/README.md b/plc4j/integrations/apache-kafka/README.md
index b8c5be77..13dbdbf5 100644
--- a/plc4j/integrations/apache-kafka/README.md
+++ b/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.
diff --git a/plc4j/integrations/apache-kafka/config/plc4x-source.properties b/plc4j/integrations/apache-kafka/config/plc4x-source.properties
index 6c024cb7..13205b99 100644
--- a/plc4j/integrations/apache-kafka/config/plc4x-source.properties
+++ b/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.
diff --git a/plc4j/integrations/apache-kafka/pom.xml b/plc4j/integrations/apache-kafka/pom.xml
index 18bc4bdd..8d9ecc00 100644
--- a/plc4j/integrations/apache-kafka/pom.xml
+++ b/plc4j/integrations/apache-kafka/pom.xml
@@ -157,7 +157,7 @@
org.apache.plc4x
- plc4j-scraper
+ plc4j-tools-event-pump
${plc4x.version}
diff --git a/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java b/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java
index 55270756..86731a24 100644
--- a/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java
+++ b/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java
@@ -57,7 +57,7 @@ public Class taskClass() {
@Override
public List> 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> 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> configs = new LinkedList<>();
for (Source source : sourceConfig.getSources()) {
// Build a list of job configurations only containing the ones referenced from
diff --git a/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceTask.java b/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceTask.java
index 9a4a3538..3bc3f7a8 100644
--- a/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceTask.java
+++ b/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceTask.java
@@ -28,13 +28,10 @@
import org.apache.plc4x.java.DefaultPlcDriverManager;
import org.apache.plc4x.java.api.PlcConnectionManager;
import org.apache.plc4x.java.api.value.PlcValue;
-import org.apache.plc4x.java.scraper.config.triggeredscraper.JobConfigurationTriggeredImplBuilder;
-import org.apache.plc4x.java.scraper.config.triggeredscraper.ScraperConfigurationTriggeredImpl;
-import org.apache.plc4x.java.scraper.config.triggeredscraper.ScraperConfigurationTriggeredImplBuilder;
-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.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.triggers.TimerTrigger;
import org.apache.plc4x.java.utils.cache.CachedPlcConnectionManager;
import org.apache.plc4x.kafka.config.Constants;
import org.apache.plc4x.kafka.util.VersionUtil;
@@ -93,10 +90,10 @@ public class Plc4xSourceTask extends SourceTask {
.field(Constants.JOB_NAME_FIELD, Schema.STRING_SCHEMA)
.build();
- // Internal buffer into which all incoming scraper responses are written to.
+ // Internal buffer into which all incoming responses are written to.
private ArrayBlockingQueue buffer;
private Integer pollReturnInterval;
- private TriggeredScraperImpl scraper;
+ private EventPump eventPump;
private final SecureRandom random = new SecureRandom();
@Override
@@ -112,12 +109,14 @@ public void start(Map props) {
pollReturnInterval = config.getInt(Constants.KAFKA_POLL_RETURN_CONFIG);
Integer bufferSize = config.getInt(Constants.BUFFER_SIZE_CONFIG);
- Map topics = new HashMap<>();
// Create a buffer with a capacity of BUFFER_SIZE_CONFIG elements which schedules access in a fair way.
buffer = new ArrayBlockingQueue<>(bufferSize, true);
- ScraperConfigurationTriggeredImplBuilder builder = new ScraperConfigurationTriggeredImplBuilder();
- builder.addSource(connectionName, plc4xConnectionString);
+ PlcConnectionManager connectionManager = CachedPlcConnectionManager.getBuilder()
+ .withConnectionManager(new DefaultPlcDriverManager())
+ .build();
+
+ eventPump = new EventPump();
List jobConfigs = config.getList(Constants.QUERIES_CONFIG);
for (String jobConfig : jobConfigs) {
@@ -129,11 +128,11 @@ public void start(Map props) {
continue;
}
- String jobName = jobConfigSegments[0];
- String topic = jobConfigSegments[1];
- Integer rate = Integer.valueOf(jobConfigSegments[2]);
- JobConfigurationTriggeredImplBuilder jobBuilder = builder.job(
- jobName, String.format("(SCHEDULED,%s)", rate)).source(connectionName);
+ final String jobName = jobConfigSegments[0];
+ final String topic = jobConfigSegments[1];
+ final int rate = Integer.parseInt(jobConfigSegments[2]);
+
+ Map tags = new LinkedHashMap<>();
for (int i = 3; i < jobConfigSegments.length; i++) {
String[] tagSegments = jobConfigSegments[i].split("#");
if (tagSegments.length != 2) {
@@ -142,108 +141,131 @@ public void start(Map props) {
jobName, jobConfigSegments[i]);
continue;
}
- String tagAlias = tagSegments[0];
- String tagAddress = tagSegments[1];
- jobBuilder.tag(tagAlias, tagAddress);
- topics.put(jobName, topic);
+ tags.put(tagSegments[0], tagSegments[1]);
+ }
+ if (tags.isEmpty()) {
+ log.warn("Job configuration '{}' doesn't contain any valid tags ... skipping it.", jobName);
+ continue;
}
- jobBuilder.build();
- }
- ScraperConfigurationTriggeredImpl scraperConfig = builder.build();
+ // One batch per job: all tags of a job are read together, at the job's rate.
+ TagBatch batch = TagBatch.builder()
+ .withBatchId(jobName)
+ .withConnectionManager(connectionManager)
+ .withConnectionString(plc4xConnectionString)
+ .addTagAddresses(tags)
+ .withTrigger(new TimerTrigger(rate, TimeUnit.MILLISECONDS))
+ .withListener(new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch tagBatch, PlcReadResponse response) {
+ handleResponse(tagBatch.getBatchId(), connectionName, topic, response);
+ }
- try {
- PlcConnectionManager connectionManager = CachedPlcConnectionManager.getBuilder()
- .withConnectionManager(new DefaultPlcDriverManager())
+ @Override
+ public void onError(TagBatch tagBatch, Throwable error) {
+ log.error("Error reading tags for job '{}': {}", tagBatch.getBatchId(), error.getMessage());
+ }
+
+ @Override
+ public void onFetchSkipped(TagBatch tagBatch, long lastFetchDurationMs, long consecutiveSkips) {
+ log.warn("Job '{}' is configured to be read every {}ms, but the last read took {}ms " +
+ "({} consecutive reads skipped).",
+ tagBatch.getBatchId(), rate, lastFetchDurationMs, consecutiveSkips);
+ }
+ })
.build();
- TriggerCollector triggerCollector = new TriggerCollectorImpl(connectionManager);
- scraper = new TriggeredScraperImpl(scraperConfig, connectionManager, (jobName, sourceName, results) -> {
- try {
- Long timestamp = System.currentTimeMillis();
+ eventPump.addBatch(batch);
+ }
- Map sourcePartition = new HashMap<>();
- sourcePartition.put("sourceName", sourceName);
- sourcePartition.put("jobName", jobName);
+ eventPump.startAll();
+ }
- Map sourceOffset = Collections.singletonMap("offset", timestamp);
+ /**
+ * Turns one response into a Kafka {@link SourceRecord} and adds it to the buffer that
+ * {@link #poll()} drains.
+ */
+ private void handleResponse(String jobName, String sourceName, String topic, PlcReadResponse response) {
+ Map results = response.getTagNames().stream()
+ .collect(HashMap::new, (map, name) -> map.put(name, response.getObject(name)), HashMap::putAll);
+ try {
+ Long timestamp = System.currentTimeMillis();
- String topic = topics.get(jobName);
+ Map sourcePartition = new HashMap<>();
+ sourcePartition.put("sourceName", sourceName);
+ sourcePartition.put("jobName", jobName);
- // Prepare the key structure.
- Struct key = new Struct(KEY_SCHEMA)
- .put(Constants.SOURCE_NAME_FIELD, sourceName)
- .put(Constants.JOB_NAME_FIELD, jobName);
+ Map sourceOffset = Collections.singletonMap("offset", timestamp);
- // Build the Schema for the result struct.
- SchemaBuilder tagSchemaBuilder = SchemaBuilder.struct()
- .name("org.apache.plc4x.kafka.schema.Tag");
+ // Prepare the key structure.
+ Struct key = new Struct(KEY_SCHEMA)
+ .put(Constants.SOURCE_NAME_FIELD, sourceName)
+ .put(Constants.JOB_NAME_FIELD, jobName);
+ // Build the Schema for the result struct.
+ SchemaBuilder tagSchemaBuilder = SchemaBuilder.struct()
+ .name("org.apache.plc4x.kafka.schema.Tag");
- for (Map.Entry result : results.entrySet()) {
- // Get tag-name and -value from the results.
- String tagName = result.getKey();
- Object tagValue = result.getValue();
- // Get the schema for the given value type.
- Schema valueSchema = getSchema(tagValue);
+ for (Map.Entry result : results.entrySet()) {
+ // Get tag-name and -value from the results.
+ String tagName = result.getKey();
+ Object tagValue = result.getValue();
- // Add the schema description for the current tag.
- tagSchemaBuilder.field(tagName, valueSchema);
- }
- Schema tagSchema = tagSchemaBuilder.build();
-
- Schema recordSchema = SchemaBuilder.struct()
- .name("org.apache.plc4x.kafka.schema.JobResult")
- .doc("PLC Job result. This contains all of the received PLCValues as well as a received timestamp")
- .field(Constants.TAGS_CONFIG, tagSchema)
- .field(Constants.TIMESTAMP_CONFIG, Schema.INT64_SCHEMA)
- .field(Constants.EXPIRES_CONFIG, Schema.OPTIONAL_INT64_SCHEMA)
- .build();
-
- // Build the struct itself.
- Struct tagStruct = new Struct(tagSchema);
- for (Map.Entry result : results.entrySet()) {
- // Get tag-name and -value from the results.
- String tagName = result.getKey();
- Object tagValue = result.getValue();
-
- if (tagSchema.field(tagName).schema().type() == Schema.Type.ARRAY) {
- tagStruct.put(tagName, ((List) tagValue).stream().map(p -> ((PlcValue) p).getObject()).collect(Collectors.toList()));
- } else {
- tagStruct.put(tagName, tagValue);
- }
- }
+ // Get the schema for the given value type.
+ Schema valueSchema = getSchema(tagValue);
- Struct recordStruct = new Struct(recordSchema)
- .put(Constants.TAGS_CONFIG, tagStruct)
- .put(Constants.TIMESTAMP_CONFIG, timestamp);
-
- // Prepare the source-record element.
- SourceRecord sourceRecord = new SourceRecord(
- sourcePartition, sourceOffset,
- topic,
- KEY_SCHEMA, key,
- recordSchema, recordStruct
- );
-
- // Add the new source-record to the buffer.
- buffer.add(sourceRecord);
- } catch (Exception e) {
- log.error("Error while parsing returned values", e);
+ // Add the schema description for the current tag.
+ tagSchemaBuilder.field(tagName, valueSchema);
+ }
+ Schema tagSchema = tagSchemaBuilder.build();
+
+ Schema recordSchema = SchemaBuilder.struct()
+ .name("org.apache.plc4x.kafka.schema.JobResult")
+ .doc("PLC Job result. This contains all of the received PLCValues as well as a received timestamp")
+ .field(Constants.TAGS_CONFIG, tagSchema)
+ .field(Constants.TIMESTAMP_CONFIG, Schema.INT64_SCHEMA)
+ .field(Constants.EXPIRES_CONFIG, Schema.OPTIONAL_INT64_SCHEMA)
+ .build();
+
+ // Build the struct itself.
+ Struct tagStruct = new Struct(tagSchema);
+ for (Map.Entry result : results.entrySet()) {
+ // Get tag-name and -value from the results.
+ String tagName = result.getKey();
+ Object tagValue = result.getValue();
+
+ if (tagSchema.field(tagName).schema().type() == Schema.Type.ARRAY) {
+ tagStruct.put(tagName, ((List) tagValue).stream().map(p -> ((PlcValue) p).getObject()).collect(Collectors.toList()));
+ } else {
+ tagStruct.put(tagName, tagValue);
}
- }, triggerCollector);
- scraper.start();
- triggerCollector.start();
- } catch (ScraperException e) {
- log.error("Error starting the scraper", e);
+ }
+ Struct recordStruct = new Struct(recordSchema)
+ .put(Constants.TAGS_CONFIG, tagStruct)
+ .put(Constants.TIMESTAMP_CONFIG, timestamp);
+
+ // Prepare the source-record element.
+ SourceRecord sourceRecord = new SourceRecord(
+ sourcePartition, sourceOffset,
+ topic,
+ KEY_SCHEMA, key,
+ recordSchema, recordStruct
+ );
+
+ // Add the new source-record to the buffer.
+ buffer.add(sourceRecord);
+ } catch (Exception e) {
+ log.error("Error while parsing returned values", e);
}
}
@Override
public void stop() {
synchronized (this) {
- scraper.stop();
+ if (eventPump != null) {
+ eventPump.close();
+ }
notifyAll(); // wake up thread waiting in awaitFetch
}
}
diff --git a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SinkTaskTest.java b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SinkTaskTest.java
index 27b20a34..e1c80d7d 100644
--- a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SinkTaskTest.java
+++ b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SinkTaskTest.java
@@ -79,7 +79,7 @@ public void parseConfig() throws Exception {
assertEquals("machineSinkA", config.get(0).get(Constants.TOPIC_CONFIG));
assertEquals("5", config.get(0).get(Constants.RETRIES_CONFIG));
assertEquals("5000", config.get(0).get(Constants.TIMEOUT_CONFIG));
- assertEquals("running#RANDOM/Temporary:Boolean|conveyorEntry#RANDOM/Temporary:Boolean|load#RANDOM/Temporary:Boolean|unload#RANDOM/Temporary:Boolean|transferLeft#RANDOM/Temporary:Boolean|transferRight#RANDOM/Temporary:Boolean|conveyorLeft#RANDOM/Temporary:Boolean|conveyorRight#RANDOM/Temporary:Boolean|numLargeBoxes#STATE/Temporary:Integer|numSmallBoxes#RANDOM/Temporary:Integer",
+ assertEquals("running#RANDOM/Temporary:BOOL|conveyorEntry#RANDOM/Temporary:BOOL|load#RANDOM/Temporary:BOOL|unload#RANDOM/Temporary:BOOL|transferLeft#RANDOM/Temporary:BOOL|transferRight#RANDOM/Temporary:BOOL|conveyorLeft#RANDOM/Temporary:BOOL|conveyorRight#RANDOM/Temporary:BOOL|numLargeBoxes#STATE/Temporary:Integer|numSmallBoxes#RANDOM/Temporary:DINT",
config.get(0).get(Constants.QUERIES_CONFIG));
}
diff --git a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SourceTaskTest.java b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SourceTaskTest.java
index 2fdfe23b..dae0111c 100644
--- a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SourceTaskTest.java
+++ b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SourceTaskTest.java
@@ -20,6 +20,7 @@
import static org.junit.jupiter.api.Assertions.*;
+import org.apache.kafka.connect.source.SourceRecord;
import org.apache.plc4x.kafka.Plc4xSourceConnector;
import org.apache.plc4x.kafka.Plc4xSourceTask;
@@ -69,7 +70,7 @@ public void parseConfig() throws Exception {
assertEquals("simulated://127.0.0.1", config.get(0).get(Constants.CONNECTION_STRING_CONFIG));
assertEquals("1000", config.get(0).get(Constants.BUFFER_SIZE_CONFIG));
assertEquals("5000", config.get(0).get(Constants.KAFKA_POLL_RETURN_CONFIG));
- assertEquals("simulateddashboard|machineData|1000|running#RANDOM/Temporary:Boolean|conveyorEntry#RANDOM/Temporary:Boolean|load#RANDOM/Temporary:Boolean|unload#RANDOM/Temporary:Boolean|transferLeft#RANDOM/Temporary:Boolean|transferRight#RANDOM/Temporary:Boolean|conveyorLeft#RANDOM/Temporary:Boolean|conveyorRight#RANDOM/Temporary:Boolean|numLargeBoxes#RANDOM/Temporary:Integer|numSmallBoxes#RANDOM/Temporary:Integer[2],simulatedheartbeat|simulatedheartbeat|500|active#RANDOM/Temporary:Integer", config.get(0).get(Constants.QUERIES_CONFIG));
+ assertEquals("simulateddashboard|machineData|1000|running#RANDOM/Temporary:BOOL|conveyorEntry#RANDOM/Temporary:BOOL|load#RANDOM/Temporary:BOOL|unload#RANDOM/Temporary:BOOL|transferLeft#RANDOM/Temporary:BOOL|transferRight#RANDOM/Temporary:BOOL|conveyorLeft#RANDOM/Temporary:BOOL|conveyorRight#RANDOM/Temporary:BOOL|numLargeBoxes#RANDOM/Temporary:DINT|numSmallBoxes#RANDOM/Temporary:DINT[2],simulatedheartbeat|simulatedheartbeat|500|active#RANDOM/Temporary:DINT", config.get(0).get(Constants.QUERIES_CONFIG));
}
@Test
@@ -85,8 +86,18 @@ public void startTasks() throws Exception {
sourceTask.start(taskConfig);
}
Thread.sleep(5000);
- for (Plc4xSourceTask sourceTask : sourceList) {
- assertNotNull(sourceTask.poll());
+ try {
+ for (Plc4xSourceTask sourceTask : sourceList) {
+ List records = sourceTask.poll();
+ assertNotNull(records);
+ // The jobs are configured to be read every 1000ms/500ms, so after 5s there
+ // has to be data in the buffer - an empty result means nothing was ever read.
+ assertFalse(records.isEmpty(), "expected the source task to have collected records");
+ }
+ } finally {
+ for (Plc4xSourceTask sourceTask : sourceList) {
+ sourceTask.stop();
+ }
}
}
diff --git a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/sink_task_no_error.properties b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/sink_task_no_error.properties
index 68b31f37..ea26c4f1 100644
--- a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/sink_task_no_error.properties
+++ b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/sink_task_no_error.properties
@@ -26,24 +26,24 @@ sinks.machineA.timeout=5000
sinks.machineA.retries=5
sinks.machineA.tags=running,conveyorEntry,load,unload,transferLeft,transferRight,conveyorLeft,conveyorRight,numLargeBoxes,numSmallBoxes
-sinks.machineA.tags.running=RANDOM/Temporary:Boolean
-sinks.machineA.tags.conveyorEntry=RANDOM/Temporary:Boolean
-sinks.machineA.tags.load=RANDOM/Temporary:Boolean
-sinks.machineA.tags.unload=RANDOM/Temporary:Boolean
-sinks.machineA.tags.transferLeft=RANDOM/Temporary:Boolean
-sinks.machineA.tags.transferRight=RANDOM/Temporary:Boolean
-sinks.machineA.tags.conveyorLeft=RANDOM/Temporary:Boolean
-sinks.machineA.tags.conveyorRight=RANDOM/Temporary:Boolean
+sinks.machineA.tags.running=RANDOM/Temporary:BOOL
+sinks.machineA.tags.conveyorEntry=RANDOM/Temporary:BOOL
+sinks.machineA.tags.load=RANDOM/Temporary:BOOL
+sinks.machineA.tags.unload=RANDOM/Temporary:BOOL
+sinks.machineA.tags.transferLeft=RANDOM/Temporary:BOOL
+sinks.machineA.tags.transferRight=RANDOM/Temporary:BOOL
+sinks.machineA.tags.conveyorLeft=RANDOM/Temporary:BOOL
+sinks.machineA.tags.conveyorRight=RANDOM/Temporary:BOOL
sinks.machineA.tags.numLargeBoxes=STATE/Temporary:Integer
-sinks.machineA.tags.numSmallBoxes=RANDOM/Temporary:Integer
+sinks.machineA.tags.numSmallBoxes=RANDOM/Temporary:DINT
sinks.machineB.connectionString=simulated://127.0.0.1
sinks.machineB.topic=machineSinkB
sinks.machineB.timeout=5000
sinks.machineB.retries=5
sinks.machineB.tags=running,conveyorEntry
-sinks.machineB.tags.running=RANDOM/Temporary:Boolean
-sinks.machineB.tags.conveyorEntry=RANDOM/Temporary:Boolean
+sinks.machineB.tags.running=RANDOM/Temporary:BOOL
+sinks.machineB.tags.conveyorEntry=RANDOM/Temporary:BOOL
bootstrap.servers=127.0.0.1:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
diff --git a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/source_task_no_error.properties b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/source_task_no_error.properties
index adf250e1..a8f0a793 100644
--- a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/source_task_no_error.properties
+++ b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/source_task_no_error.properties
@@ -31,20 +31,20 @@ jobs=simulateddashboard,simulatedheartbeat
jobs.simulateddashboard.interval=1000
jobs.simulateddashboard.tags=running,conveyorEntry,load,unload,transferLeft,transferRight,conveyorLeft,conveyorRight,numLargeBoxes,numSmallBoxes
-jobs.simulateddashboard.tags.running=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.conveyorEntry=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.load=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.unload=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.transferLeft=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.transferRight=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.conveyorLeft=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.conveyorRight=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.numLargeBoxes=RANDOM/Temporary:Integer
-jobs.simulateddashboard.tags.numSmallBoxes=RANDOM/Temporary:Integer[2]
+jobs.simulateddashboard.tags.running=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.conveyorEntry=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.load=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.unload=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.transferLeft=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.transferRight=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.conveyorLeft=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.conveyorRight=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.numLargeBoxes=RANDOM/Temporary:DINT
+jobs.simulateddashboard.tags.numSmallBoxes=RANDOM/Temporary:DINT[2]
jobs.simulatedheartbeat.interval=500
jobs.simulatedheartbeat.tags=active
-jobs.simulatedheartbeat.tags.active=RANDOM/Temporary:Integer
+jobs.simulatedheartbeat.tags.active=RANDOM/Temporary:DINT
bootstrap.servers=localhost:9092
#key.converter=org.apache.kafka.connect.json.JsonConverter