You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by tm...@apache.org on 2019/05/05 18:44:28 UTC
[plc4x] 01/01: PLC4X-109 implemented comparison to previous value
for numeric values, extracted trigger into a separate class,
updated java-docs, divers improvements
This is an automated email from the ASF dual-hosted git repository.
tmitsch pushed a commit to branch scraper_refactoring_and_improvement
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 7ed2946b6dee93d8dba5994ac15f08c927c7c8d7
Author: Tim Mitsch <t....@pragmaticindustries.de>
AuthorDate: Tue Apr 9 18:13:12 2019 +0200
PLC4X-109 implemented comparison to previous value for numeric values, extracted trigger into a separate class, updated java-docs, divers improvements
---
.../org/apache/plc4x/java/s7/netty/S7Protocol.java | 26 +-
plc4j/utils/scraper/README.md | 21 +
.../triggeredscraper/TriggeredScraperImpl.java | 170 +++--
.../triggeredscraper/TriggeredScraperTask.java | 60 +-
.../triggerhandler/TriggerConfiguration.java | 692 +++++++++++++++------
.../triggerhandler/TriggerHandler.java | 112 +---
.../triggerhandler/TriggerHandlerImpl.java | 112 ++++
.../triggerhandler/collector/TriggerCollector.java | 61 ++
.../collector/TriggerCollectorImpl.java | 316 ++++++++++
.../plc4x/java/scraper/TriggeredScraperRunner.java | 10 +-
.../java/scraper/TriggeredScraperRunnerModbus.java | 12 +-
.../triggerhandler/TriggerConfigurationTest.java | 60 +-
.../test/resources/example_triggered_scraper.yml | 24 +-
13 files changed, 1265 insertions(+), 411 deletions(-)
diff --git a/plc4j/protocols/s7/src/main/java/org/apache/plc4x/java/s7/netty/S7Protocol.java b/plc4j/protocols/s7/src/main/java/org/apache/plc4x/java/s7/netty/S7Protocol.java
index 17214ca..02eb677 100644
--- a/plc4j/protocols/s7/src/main/java/org/apache/plc4x/java/s7/netty/S7Protocol.java
+++ b/plc4j/protocols/s7/src/main/java/org/apache/plc4x/java/s7/netty/S7Protocol.java
@@ -528,8 +528,10 @@ public class S7Protocol extends ChannelDuplexHandler {
maxAmqCallee = setupCommunicationParameter.getMaxAmqCallee();
pduSize = setupCommunicationParameter.getPduLength();
- logger.info("S7Connection established pdu-size {}, max-amq-caller {}, " +
+ if(logger.isInfoEnabled()) {
+ logger.info("S7Connection established pdu-size {}, max-amq-caller {}, " +
"max-amq-callee {}", pduSize, maxAmqCaller, maxAmqCallee);
+ }
// Only if the controller type is set to "ANY", then try to identify the PLC type.
if(controllerType == S7ControllerType.ANY) {
@@ -544,11 +546,14 @@ public class S7Protocol extends ChannelDuplexHandler {
}
// If a concrete type was specified, then we're done here.
else {
+ if(logger.isDebugEnabled()) {
+ logger.debug("Successfully connected to S7: {}", controllerType.name());
+ logger.debug("- max amq caller: {}", maxAmqCaller);
+ logger.debug("- max amq callee: {}", maxAmqCallee);
+ logger.debug("- pdu size: {}", pduSize);
+ }
if(logger.isInfoEnabled()) {
- logger.info(String.format("Successfully connected to S7: %s", controllerType.name()));
- logger.info(String.format("- max amq caller: %s", maxAmqCaller));
- logger.info(String.format("- max amq callee: %s", maxAmqCallee));
- logger.info(String.format("- pdu size: %s", pduSize));
+ logger.info("Successfully connected to S7: {} wit PDU {}", controllerType.name(),pduSize);
}
// Send an event that connection setup is complete.
@@ -567,11 +572,14 @@ public class S7Protocol extends ChannelDuplexHandler {
}
}
}
+ if(logger.isDebugEnabled()) {
+ logger.debug("Successfully connected to S7: {}", controllerType.name());
+ logger.debug("- max amq caller: {}", maxAmqCaller);
+ logger.debug("- max amq callee: {}", maxAmqCallee);
+ logger.debug("- pdu size: {}", pduSize);
+ }
if(logger.isInfoEnabled()) {
- logger.info(String.format("Successfully connected to S7: %s", controllerType.name()));
- logger.info(String.format("- max amq caller: %s", maxAmqCaller));
- logger.info(String.format("- max amq callee: %s", maxAmqCallee));
- logger.info(String.format("- pdu size: %s", pduSize));
+ logger.info("Successfully connected to S7: {} wit PDU {}", controllerType.name(),pduSize);
}
// Send an event that connection setup is complete.
diff --git a/plc4j/utils/scraper/README.md b/plc4j/utils/scraper/README.md
new file mode 100644
index 0000000..1c75f33
--- /dev/null
+++ b/plc4j/utils/scraper/README.md
@@ -0,0 +1,21 @@
+<!--
+
+ 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.
+
+-->
+# Usage of (Triggered) Scraper
+
+ToDo: Write this!
\ No newline at end of file
diff --git a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/TriggeredScraperImpl.java b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/TriggeredScraperImpl.java
index 40ad5d5..7de9e56 100644
--- a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/TriggeredScraperImpl.java
+++ b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/TriggeredScraperImpl.java
@@ -29,10 +29,15 @@ import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
import org.apache.plc4x.java.PlcDriverManager;
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.PlcReadResponse;
import org.apache.plc4x.java.scraper.*;
import org.apache.plc4x.java.scraper.exception.ScraperException;
import org.apache.plc4x.java.scraper.config.triggeredscraper.TriggeredScraperConfiguration;
+import org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollector;
import org.apache.plc4x.java.scraper.util.PercentageAboveThreshold;
+import org.apache.plc4x.java.spi.PlcDriver;
import org.apache.plc4x.java.utils.connectionpool.PooledPlcDriverManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,6 +46,7 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.*;
+import java.util.stream.Collectors;
/**
* replaces the old Scraper that only could do scheduled scraping jobs
@@ -57,6 +63,7 @@ import java.util.concurrent.*;
public class TriggeredScraperImpl implements Scraper {
private static final Logger LOGGER = LoggerFactory.getLogger(TriggeredScraperImpl.class);
+ private static final int DEFAULT_FUTURE_TIME_OUT = 2000;
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10,
new BasicThreadFactory.Builder()
@@ -74,18 +81,49 @@ public class TriggeredScraperImpl implements Scraper {
private final ResultHandler resultHandler;
private final MultiValuedMap<ScrapeJob, ScraperTask> tasks = new ArrayListValuedHashMap<>();
- private final MultiValuedMap<ScraperTask, ScheduledFuture<?>> futures = new ArrayListValuedHashMap<>();
+ private final MultiValuedMap<ScraperTask, ScheduledFuture<?>> scraperTaskMap = new ArrayListValuedHashMap<>();
private final PlcDriverManager driverManager;
private final List<ScrapeJob> jobs;
+ private long futureTimeOut;
+
+ private final TriggerCollector triggerCollector;
+
/**
* Creates a Scraper instance from a configuration.
* By default a {@link PooledPlcDriverManager} is used.
* @param config Configuration to use.
- * @param resultHandler
+ * @param resultHandler handler the defines the processing of acquired data
+ */
+ public TriggeredScraperImpl(TriggeredScraperConfiguration config, ResultHandler resultHandler,TriggerCollector triggerCollector) throws ScraperException {
+ this(resultHandler, createPooledDriverManager(), config.getJobs(),triggerCollector,DEFAULT_FUTURE_TIME_OUT);
+ }
+
+ /**
+ * Creates a Scraper instance from a configuration.
+ * @param config Configuration to use.
+ * @param plcDriverManager external DriverManager
+ * @param resultHandler handler the defines the processing of acquired data
*/
- public TriggeredScraperImpl(TriggeredScraperConfiguration config, ResultHandler resultHandler) throws ScraperException {
- this(resultHandler, createPooledDriverManager(), config.getJobs());
+ public TriggeredScraperImpl(TriggeredScraperConfiguration config, PlcDriverManager plcDriverManager, ResultHandler resultHandler,TriggerCollector triggerCollector) throws ScraperException {
+ this(resultHandler, plcDriverManager, config.getJobs(),triggerCollector,DEFAULT_FUTURE_TIME_OUT);
+ }
+
+ /**
+ * Creates a Scraper instance from a configuration.
+ * @param plcDriverManager external DriverManager
+ * @param resultHandler handler the defines the processing of acquired data
+ * @param jobs list of jobs that scraper shall handle
+ * @param triggerCollector a collection that centralizes the trigger requests and joins them to grouped plc requests
+ * @param futureTimeOut max duration of future to return a result
+ */
+ public TriggeredScraperImpl(ResultHandler resultHandler, PlcDriverManager plcDriverManager, List<ScrapeJob> jobs,TriggerCollector triggerCollector, long futureTimeOut) {
+ this.resultHandler = resultHandler;
+ Validate.notEmpty(jobs);
+ this.driverManager = plcDriverManager;
+ this.jobs = jobs;
+ this.triggerCollector = triggerCollector;
+ this.futureTimeOut = futureTimeOut;
}
/**
@@ -96,21 +134,13 @@ public class TriggeredScraperImpl implements Scraper {
private static PooledPlcDriverManager createPooledDriverManager() {
return new PooledPlcDriverManager(pooledPlcConnectionFactory -> {
GenericKeyedObjectPoolConfig<PlcConnection> poolConfig = new GenericKeyedObjectPoolConfig<>();
- poolConfig.setMinIdlePerKey(1); // This should avoid problems with long running connect attempts??
+ poolConfig.setMinIdlePerKey(1);
poolConfig.setTestOnBorrow(true);
poolConfig.setTestOnReturn(true);
return new GenericKeyedObjectPool<>(pooledPlcConnectionFactory, poolConfig);
});
}
-
- public TriggeredScraperImpl(ResultHandler resultHandler, PlcDriverManager driverManager, List<ScrapeJob> jobs) {
- this.resultHandler = resultHandler;
- Validate.notEmpty(jobs);
- this.driverManager = driverManager;
- this.jobs = jobs;
- }
-
/**
* Start the scraping.
*/
@@ -119,39 +149,44 @@ public class TriggeredScraperImpl implements Scraper {
public void start() {
// Schedule all jobs
LOGGER.info("Starting jobs...");
- jobs.stream()
- .flatMap(job -> job.getSourceConnections().entrySet().stream()
- .map(entry -> Triple.of(job, entry.getKey(), entry.getValue()))
- )
- .forEach(
- tuple -> {
- LOGGER.debug("Register task for job {} for conn {} ({}) at rate {} ms",
- tuple.getLeft().getJobName(), tuple.getMiddle(), tuple.getRight(), tuple.getLeft().getScrapeRate());
- TriggeredScraperTask task =
- null;
- try {
- task = new TriggeredScraperTask(driverManager,
- tuple.getLeft().getJobName(),
- tuple.getMiddle(),
- tuple.getRight(),
- tuple.getLeft().getFields(),
- 1_000,
- executorService,
- resultHandler,
- (TriggeredScrapeJobImpl) tuple.getLeft());
- // Add task to internal list
- tasks.put(tuple.getLeft(), task);
- ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(task,
- 0, tuple.getLeft().getScrapeRate(), TimeUnit.MILLISECONDS);
-
- // Store the handle for stopping, etc.
- futures.put(task, future);
- } catch (ScraperException e) {
- LOGGER.warn("Error executing the job {} for conn {} ({}) at rate {} ms",tuple.getLeft().getJobName(), tuple.getMiddle(), tuple.getRight(), tuple.getLeft().getScrapeRate(),e);
- }
+ //start iterating over all available jobs
+ for(ScrapeJob job:jobs){
+ //iterate over all source the jobs shall performed on
+ for(Map.Entry<String,String> sourceEntry:job.getSourceConnections().entrySet()){
+ LOGGER.debug("Register task for job {} for conn {} ({}) at rate {} ms",
+ job.getJobName(),
+ sourceEntry.getKey(),
+ sourceEntry.getValue(),
+ job.getScrapeRate());
+
+ //create the regarding triggered scraper task
+ TriggeredScraperTask triggeredScraperTask;
+ try {
+ triggeredScraperTask = new TriggeredScraperTask(
+ driverManager,
+ job.getJobName(),
+ sourceEntry.getKey(),
+ sourceEntry.getValue(),
+ job.getFields(),
+ futureTimeOut,
+ executorService,
+ resultHandler,
+ (TriggeredScrapeJobImpl) job,
+ triggerCollector);
+
+ // Add task to internal list
+ LOGGER.info("Task {} added to scheduling", triggeredScraperTask);
+ tasks.put(job, triggeredScraperTask);
+ ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(triggeredScraperTask, 0, job.getScrapeRate(), TimeUnit.MILLISECONDS);
+ // Store the handle for stopping, etc.
+ scraperTaskMap.put(triggeredScraperTask, future);
+ } catch (ScraperException e) {
+ LOGGER.warn("Error executing the job {} for conn {} ({}) at rate {} ms",job.getJobName(), sourceEntry.getKey(), sourceEntry.getValue(), job.getScrapeRate(),e);
}
- );
+ }
+
+ }
// Add statistics tracker
scheduler.scheduleAtFixedRate(() -> {
@@ -177,11 +212,54 @@ public class TriggeredScraperImpl implements Scraper {
public void stop() {
// Stop all futures
LOGGER.info("Stopping scraper...");
- for (Map.Entry<ScraperTask, ScheduledFuture<?>> entry : futures.entries()) {
+ for (Map.Entry<ScraperTask, ScheduledFuture<?>> entry : scraperTaskMap.entries()) {
LOGGER.debug("Stopping task {}...", entry.getKey());
entry.getValue().cancel(true);
}
// Clear the map
- futures.clear();
+ scraperTaskMap.clear();
+ }
+
+ /**
+ * acquires a plc connection from connection pool
+ * @param plcDriverManager Driver manager handling connection and pools
+ * @param connectionString Connection string as defined in the regarding implementation of {@link PlcDriver}
+ * @param executorService ExecuterService holding a pool as threads handling requests and stuff
+ * @param requestTimeoutMs maximum awaiting for the the future to return a result
+ * @return the {@link PlcConnection} used for acquiring data from PLC endpoint
+ * @throws InterruptedException
+ * @throws ExecutionException
+ * @throws TimeoutException
+ */
+ public static PlcConnection getPlcConnection(PlcDriverManager plcDriverManager,
+ String connectionString,
+ ExecutorService executorService,
+ long requestTimeoutMs) throws InterruptedException, ExecutionException, TimeoutException {
+ CompletableFuture<PlcConnection> future = CompletableFuture.supplyAsync(() -> {
+ try {
+ return plcDriverManager.getConnection(connectionString);
+ } catch (PlcConnectionException e) {
+ LOGGER.warn("Unable to instantiate connection to " + connectionString, e);
+ throw new PlcRuntimeException(e);
+ }
+ catch (Exception e){
+ LOGGER.warn("Unable to instantiate connection to " + connectionString, e);
+ throw new PlcRuntimeException(e);
+ }
+ }, executorService);
+ return future.get(requestTimeoutMs, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * transforms the results from a {@link PlcReadResponse} into a map
+ * @param plcReadResponse response that shall be converted to map for further processing
+ * @return the converted map
+ */
+ public static Map<String, Object> convertPlcResponseToMap(PlcReadResponse plcReadResponse) {
+ return plcReadResponse.getFieldNames().stream()
+ .collect(Collectors.toMap(
+ name -> name,
+ plcReadResponse::getObject
+ ));
}
}
diff --git a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/TriggeredScraperTask.java b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/TriggeredScraperTask.java
index 56549a5..cb037a6 100644
--- a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/TriggeredScraperTask.java
+++ b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/TriggeredScraperTask.java
@@ -23,8 +23,6 @@ import org.apache.commons.lang3.time.StopWatch;
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
import org.apache.plc4x.java.PlcDriverManager;
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;
import org.apache.plc4x.java.api.messages.PlcReadResponse;
import org.apache.plc4x.java.api.types.PlcResponseCode;
@@ -32,6 +30,8 @@ import org.apache.plc4x.java.scraper.ResultHandler;
import org.apache.plc4x.java.scraper.ScraperTask;
import org.apache.plc4x.java.scraper.exception.ScraperException;
import org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.TriggerHandler;
+import org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.TriggerHandlerImpl;
+import org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,6 +49,8 @@ import java.util.stream.Collectors;
* ToDo Implement the monitoring as well: PLC4X-90
*/
public class TriggeredScraperTask implements ScraperTask {
+
+
private static final Logger LOGGER = LoggerFactory.getLogger(TriggeredScraperTask.class);
private final PlcDriverManager driverManager;
@@ -66,6 +68,7 @@ public class TriggeredScraperTask implements ScraperTask {
private final DescriptiveStatistics latencyStatistics = new DescriptiveStatistics(1000);
private final DescriptiveStatistics failedStatistics = new DescriptiveStatistics(1000);
+
public TriggeredScraperTask(PlcDriverManager driverManager,
String jobName,
String connectionAlias,
@@ -74,7 +77,8 @@ public class TriggeredScraperTask implements ScraperTask {
long requestTimeoutMs,
ExecutorService executorService,
ResultHandler resultHandler,
- TriggeredScrapeJobImpl triggeredScrapeJob) throws ScraperException {
+ TriggeredScrapeJobImpl triggeredScrapeJob,
+ TriggerCollector triggerCollector) throws ScraperException {
this.driverManager = driverManager;
this.jobName = jobName;
this.connectionAlias = connectionAlias;
@@ -83,7 +87,7 @@ public class TriggeredScraperTask implements ScraperTask {
this.requestTimeoutMs = requestTimeoutMs;
this.executorService = executorService;
this.resultHandler = resultHandler;
- this.triggerHandler = new TriggerHandler(triggeredScrapeJob.getTriggerConfig(),triggeredScrapeJob,this);
+ this.triggerHandler = new TriggerHandlerImpl(triggeredScrapeJob.getTriggerConfig(),triggeredScrapeJob,this,triggerCollector);
}
@Override
@@ -91,22 +95,18 @@ public class TriggeredScraperTask implements ScraperTask {
public void run() {
if(this.triggerHandler.checkTrigger()) {
// Does a single fetch only when trigger is valid
- LOGGER.trace("Start new scrape of task of job {} for connection {}", jobName, connectionAlias);
+ if(LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Start new scrape of task of job {} for connection {}", jobName, connectionAlias);
+ }
requestCounter.incrementAndGet();
StopWatch stopWatch = new StopWatch();
stopWatch.start();
PlcConnection connection = null;
try {
- CompletableFuture<PlcConnection> future = CompletableFuture.supplyAsync(() -> {
- try {
- return driverManager.getConnection(connectionString);
- } catch (PlcConnectionException e) {
- LOGGER.warn("Unable to instantiate connection to " + connectionString, e);
- throw new PlcRuntimeException(e);
- }
- }, executorService);
- connection = future.get(10 * requestTimeoutMs, TimeUnit.MILLISECONDS);
- LOGGER.trace("Connection to {} established: {}", connectionString, connection);
+ connection = TriggeredScraperImpl.getPlcConnection(driverManager,connectionString,executorService,requestTimeoutMs);
+ if(LOGGER.isTraceEnabled()) {
+ LOGGER.trace("Connection to {} established: {}", connectionString, connection);
+ }
PlcReadResponse response;
try {
PlcReadRequest.Builder builder = connection.readRequestBuilder();
@@ -131,7 +131,7 @@ public class TriggeredScraperTask implements ScraperTask {
// Validate response
validateResponse(response);
// Handle response (Async)
- CompletableFuture.runAsync(() -> resultHandler.handle(jobName, connectionAlias, transformResponseToMap(response)), executorService);
+ CompletableFuture.runAsync(() -> resultHandler.handle(jobName, connectionAlias, TriggeredScraperImpl.convertPlcResponseToMap(response)), executorService);
} catch (Exception e) {
LOGGER.debug("Exception during scrape", e);
handleException(e);
@@ -147,6 +147,11 @@ public class TriggeredScraperTask implements ScraperTask {
}
}
+ /**
+ * detects if {@link PlcReadResponse} is valid
+ * //ToDo CHECK: is this thing S7 specific? if not this comment can be removed
+ * @param response the {@link PlcReadResponse} that should be validated
+ */
private void validateResponse(PlcReadResponse response) {
Map<String, PlcResponseCode> failedFields = response.getFieldNames().stream()
.filter(name -> !PlcResponseCode.OK.equals(response.getResponseCode(name)))
@@ -159,13 +164,7 @@ public class TriggeredScraperTask implements ScraperTask {
}
}
- private Map<String, Object> transformResponseToMap(PlcReadResponse response) {
- return response.getFieldNames().stream()
- .collect(Collectors.toMap(
- name -> name,
- response::getObject
- ));
- }
+
@Override
public String getJobName() {
@@ -222,4 +221,19 @@ public class TriggeredScraperTask implements ScraperTask {
public long getRequestTimeoutMs() {
return requestTimeoutMs;
}
+
+ @Override
+ public String toString() {
+ return "TriggeredScraperTask{" +
+ "driverManager=" + driverManager +
+ ", jobName='" + jobName + '\'' +
+ ", connectionAlias='" + connectionAlias + '\'' +
+ ", connectionString='" + connectionString + '\'' +
+ ", fields=" + fields +
+ ", requestTimeoutMs=" + requestTimeoutMs +
+ ", executorService=" + executorService +
+ ", resultHandler=" + resultHandler +
+ ", triggerHandler=" + triggerHandler +
+ '}';
+ }
}
diff --git a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerConfiguration.java b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerConfiguration.java
index 8d7ba26..9394e8b 100644
--- a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerConfiguration.java
+++ b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerConfiguration.java
@@ -27,6 +27,8 @@ import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScrapeJobImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -43,57 +45,54 @@ public class TriggerConfiguration{
private static final double TOLERANCE_FLOATING_EQUALITY = 1e-6;
private static final Pattern TRIGGER_STRATEGY_PATTERN =
- Pattern.compile("\\((?<strategy>[A-Z_0-9]+),(?<scheduledInterval>\\d+)(,(\\((?<triggerVar>\\S+)\\))((?<comp>[!=<>]{1,2}))(\\((?<compVar>[a-z0-9.\\-]+)\\)))?\\)");
+ Pattern.compile("\\((?<strategy>[A-Z_0-9]+),(?<scheduledInterval>\\d+)(,(\\((?<triggerVar>[^!=<>()]+)\\))((?<comp>[!=<>]{1,2}))(\\((?<compVar>[PREVa-z0-9.\\-]+)\\))((?<concatConn>[ANDOR]{2,3})(\\((?<triggerVar2>[^!=<>()]+)\\))((?<comp2>[!=<>]{1,2}))(\\((?<compVar2>[PREVa-z0-9.\\-]+)\\)))?)?\\)");
+ private static final String PREVIOUS_DEF = "PREV";
private final TriggerType triggerType;
private final Long scrapeInterval;
- private final String triggerVariable;
- private final String comparator;
- private Comparators comparatorType;
private TriggeredScrapeJobImpl triggeredScrapeJobImpl;
- private final Object compareValue;
- private final PlcField plcField;
+
+ private List<TriggerElement> triggerElementList;
+
+
/**
* default constructor when an S7Field should be used for triggering
* @param triggerType type of trigger from enum
* @param scrapeInterval scrape interval of triggered variable
- * @param triggerVariable field that is conditional for trigger comparison
- * @param comparator selected comparator
- * @param compareValue selected ref-value that is comapred against
+ * @param triggerElementList list of triggerElemts with concat that combined is used as triger
* @param triggeredScrapeJobImpl the job which is valid for the configuration
* @throws ScraperException when something goes wrong with configuration
*/
- public TriggerConfiguration(TriggerType triggerType, String scrapeInterval, String triggerVariable, String comparator, String compareValue, TriggeredScrapeJobImpl triggeredScrapeJobImpl) throws ScraperException {
+ public TriggerConfiguration(TriggerType triggerType,
+ String scrapeInterval,
+ List<TriggerElement> triggerElementList,
+ TriggeredScrapeJobImpl triggeredScrapeJobImpl)
+ throws ScraperException {
+ this.triggerElementList = triggerElementList;
this.triggerType = triggerType;
this.triggeredScrapeJobImpl = triggeredScrapeJobImpl;
this.scrapeInterval = parseScrapeInterval(scrapeInterval);
- this.triggerVariable = triggerVariable;
- this.comparator = comparator;
+
+ String exceptionMessage;
if(this.triggerType.equals(TriggerType.S7_TRIGGER_VAR)) {
//test for valid field-connection string, on exception quit job and return message to user
- try {
- // TODO: PLC4X-106 - Make the Scraper not depend on S7 directly
- this.plcField = S7Field.of(triggerVariable);
- } catch (PlcInvalidFieldException e) {
- logger.debug(e.getMessage(), e);
- String exceptionMessage = String.format("Invalid trigger Field for Job %s: %s", triggeredScrapeJobImpl.getJobName(), triggerVariable);
- throw new ScraperException(exceptionMessage);
- }
+
+ if(this.triggerElementList.isEmpty()){
+ exceptionMessage = String.format("No items in trigger List for trigger-type S7_TRIGGER_VAR for Job !", triggeredScrapeJobImpl.getJobName());
+ throw new ScraperException(exceptionMessage);
+ }
+ checkTriggerVarList();
+
//ToDo add more and other trigger
}
else{
- String exceptionMessage = String.format("TriggerType %s is not yet implemented", this.triggerType);
+ exceptionMessage = String.format("TriggerType %s is not yet implemented", this.triggerType);
throw new ScraperException(exceptionMessage);
}
-
- this.compareValue = convertCompareValue(compareValue);
- detectComparatorType();
- matchTypeAndComparator();
-
}
/**
@@ -105,11 +104,21 @@ public class TriggerConfiguration{
public TriggerConfiguration(TriggerType triggerType, String scrapeInterval) throws ScraperException {
this.triggerType = triggerType;
this.scrapeInterval = parseScrapeInterval(scrapeInterval);
- this.triggerVariable = null;
- this.comparator = null;
- this.compareValue = null;
- this.plcField = null;
- this.comparatorType = null;
+ this.triggerElementList = new ArrayList<>();
+ }
+
+ /**
+ * checks the trigger list for correct syntax
+ * @throws ScraperException if syntax isn't correct an exception is thrown
+ */
+ private void checkTriggerVarList() throws ScraperException {
+ boolean first = true;
+ for(TriggerElement triggerElement:this.triggerElementList){
+ if(!first && triggerElement.getConcatType()==null){
+ throw new ScraperException("A concat for the second and following trigger must be given!");
+ }
+ first = false;
+ }
}
/**
@@ -131,191 +140,215 @@ public class TriggerConfiguration{
/**
* evaluates the trigger dependent of base type and converts acquired respectively ref-value to the needed datatype
- * @param value acquired value
+ * @param acquiredValues acquired value
* @return true when condition is matched, false otherwise
* @throws ScraperException when something goes wrong
*/
- boolean evaluateTrigger(Object value) throws ScraperException {
- if(validateDataType().equals(Boolean.class)){
- boolean currentValue;
- boolean refValue;
- try{
- currentValue = (boolean) value;
- refValue = (boolean) compareValue;
- }
- catch (Exception e){
- handleException(e);
- return false;
- }
- if(this.comparatorType.equals(Comparators.EQUAL)){
- return currentValue == refValue;
- }
- else {
- return currentValue != refValue;
- }
- }
- if(validateDataType().equals(Double.class)
- || validateDataType().equals(Integer.class)
- || validateDataType().equals(Long.class)) {
- double currentValue;
- double refValue;
- try{
- refValue = (double) compareValue;
- if(value instanceof Short){
- currentValue = ((Short) value).doubleValue();
- }
- else {
- if (value instanceof Integer) {
- currentValue = ((Integer) value).doubleValue();
- }
- else {
- if (value instanceof Long) {
- currentValue = ((Long) value).doubleValue();
- }
- else{
- if (value instanceof Double) {
- currentValue = (Double) value;
- }else {
- currentValue = (double) value;
- }
- }
- }
-
- }
-
- //
-
- }
- catch (Exception e){
- handleException(e);
- return false;
- }
-
- switch (this.comparatorType) {
- case EQUAL:
- return isApproximately(currentValue,refValue, TOLERANCE_FLOATING_EQUALITY);
- case UNEQUAL:
- return !isApproximately(currentValue,refValue, TOLERANCE_FLOATING_EQUALITY);
- case SMALLER:
- return currentValue < refValue;
- case SMALLER_EQUAL:
- return currentValue <= refValue;
- case GREATER:
- return currentValue > refValue;
- case GREATER_EQUAL:
- return currentValue >= refValue;
- }
-
- }
- //should not happen, as fallback return false which always implies that no data is collected
- return false;
- }
-
- /**
- * convertes parsed comparator from regex to ComparatorType
- * @throws ScraperException when no valid comparator has been used
- */
- private void detectComparatorType() throws ScraperException {
- switch (this.comparator){
- case "==":
- this.comparatorType= Comparators.EQUAL;
- break;
- case "!=":
- this.comparatorType= Comparators.UNEQUAL;
- break;
- case "<=":
- this.comparatorType= Comparators.SMALLER_EQUAL;
- break;
- case "<":
- this.comparatorType= Comparators.SMALLER;
- break;
- case ">=":
- this.comparatorType= Comparators.GREATER_EQUAL;
- break;
- case ">":
- this.comparatorType= Comparators.GREATER;
- break;
- default:
- throw new ScraperException("Invalid comparator detected!");
- }
+ boolean evaluateTrigger(List<Object> acquiredValues) throws ScraperException {
+ TriggerEvaluation triggerEvaluation = new TriggerEvaluation(acquiredValues,triggerElementList);
+ return triggerEvaluation.evaluateTrigger();
}
- /**
- * matches data-type and comparator for a valid combination
- * @throws ScraperException when invalid combination is detected
- */
- private void matchTypeAndComparator() throws ScraperException {
- if(validateDataType().equals(Boolean.class)
- && !(this.comparatorType.equals(Comparators.EQUAL) || this.comparatorType.equals(Comparators.UNEQUAL))){
- String exceptionMessage = String.format("Trigger-Data-Type (%s) and Comparator (%s) do not match",this.plcField.getDefaultJavaType(),this.comparatorType);
- throw new ScraperException(exceptionMessage);
- }
- //all other combinations are valid
- }
/**
* defines the used base type for comparison
* @return the detected base type
- * @throws ScraperException when an unsupported S7-Type is choosen,which is not (yet) implemented for comparison
+ * @throws ScraperException when an unsupported S7-Type is chosen,which is not (yet) implemented for comparison
* ToDo check how to handle time-variables if needed
*/
- private Class<?> validateDataType() throws ScraperException {
- if(this.plcField!=null){
- Class<?> javaDataType = this.plcField.getDefaultJavaType();
+ private static Class<?> validateDataType(PlcField plcField) throws ScraperException {
+ if(plcField!=null){
+ Class<?> javaDataType = plcField.getDefaultJavaType();
if(!javaDataType.equals(Boolean.class)
&& !javaDataType.equals(Integer.class)
&& !javaDataType.equals(Long.class)
&& !javaDataType.equals(Double.class)
){
- String exceptionMessage = String.format("Unsupported plc-trigger variable %s with converted data-type %s used",this.plcField,this.plcField.getDefaultJavaType());
+ String exceptionMessage = String.format("Unsupported plc-trigger variable %s with converted data-type %s used",plcField,plcField.getDefaultJavaType());
throw new ScraperException(exceptionMessage);
}
return javaDataType;
}
else{
- String exceptionMessage = String.format("Unsupported plc-trigger variable %s with converted data-type %s used",this.plcField,this.plcField.getDefaultJavaType());
+ String exceptionMessage = String.format("Unsupported plc-trigger variable %s with converted data-type %s used",plcField,plcField.getDefaultJavaType());
throw new ScraperException(exceptionMessage);
}
}
+
/**
- * parses the ref-value to a given value, as well as checking if ref-value matches to the given data-type
- * @param compareValue compare-value extracted by regex
- * @return converted object to needed data-type
- * @throws ScraperException when something does not match or parsing fails
+ * nested class performing the trigger evaluation
*/
- private Object convertCompareValue(String compareValue) throws ScraperException {
- Class<?> javaDataType =validateDataType();
- if(javaDataType.equals(Boolean.class)){
- switch (compareValue){
- case "1":
- case "true":
- return true;
- case "0":
- case "false":
- return false;
- default:
- String exceptionMessage = String.format("No valid compare Value at DataType Boolean for trigger for Job %s: %s",triggeredScrapeJobImpl.getJobName(),compareValue);
- throw new ScraperException(exceptionMessage);
- }
+ class TriggerEvaluation{
+ private List<Object> acquiredValuesList;
+ private List<TriggerElement> triggerElementList;
+
+ TriggerEvaluation(List<Object> acquiredValuesList, List<TriggerElement> triggerElementList) {
+ this.acquiredValuesList = acquiredValuesList;
+ this.triggerElementList = triggerElementList;
}
- if(javaDataType.equals(Double.class)
- || javaDataType.equals(Integer.class)
- || javaDataType.equals(Long.class)){
- try {
- //everything fits to Double for conversion ... so for first step use only double
- //ToDo if different handling dependent on specific datatype is needed then differ
- return Double.parseDouble(compareValue);
+
+ /**
+ * does the evaluation of the trigger conditions are met
+ * //ToDo refactor this
+ * @return true if trigger conditions are met, false otherwise
+ * @throws ScraperException
+ */
+ boolean evaluateTrigger() throws ScraperException {
+ List<Boolean> triggerResultList = new ArrayList<>();
+ if(logger.isTraceEnabled()){
+ logger.trace("eval values: {}",acquiredValuesList);
}
- catch (Exception e){
- logger.debug(e.getMessage(), e);
- String exceptionMessage = String.format("No valid compare Value at DataType Numeric for trigger for Job %s: %s",triggeredScrapeJobImpl.getJobName(),compareValue);
- throw new ScraperException(exceptionMessage);
+ for(int countElements=0; countElements<acquiredValuesList.size();countElements++){
+ TriggerElement triggerElement = triggerElementList.get(countElements);
+ Object acquiredObject = acquiredValuesList.get(countElements);
+ if(validateDataType(triggerElement.getPlcField()).equals(Boolean.class)){
+ boolean currentValue;
+ boolean refValue;
+ try{
+ currentValue = (boolean) acquiredObject;
+ refValue = (boolean) triggerElement.getCompareValue();
+ }
+ catch (Exception e){
+ handleException(e);
+ return false;
+ }
+ if(triggerElement.getComparatorType().equals(Comparator.EQUAL)){
+ triggerResultList.add(currentValue == refValue);
+ }
+ else {
+ triggerResultList.add(currentValue != refValue);
+ }
+ }
+ if(validateDataType(triggerElement.getPlcField()).equals(Double.class)
+ || validateDataType(triggerElement.getPlcField()).equals(Integer.class)
+ || validateDataType(triggerElement.getPlcField()).equals(Long.class)) {
+ boolean skipComparison = false;
+ double currentValue;
+ double refValue = 0;
+ try{
+
+ if(acquiredObject instanceof Short){
+ currentValue = ((Short) acquiredObject).doubleValue();
+ }
+ else {
+ if (acquiredObject instanceof Integer) {
+ currentValue = ((Integer) acquiredObject).doubleValue();
+ }
+ else {
+ if (acquiredObject instanceof Long) {
+ currentValue = ((Long) acquiredObject).doubleValue();
+ }
+ else{
+ if (acquiredObject instanceof Double) {
+ currentValue = (Double) acquiredObject;
+ }else {
+ currentValue = (double) acquiredObject;
+ }
+ }
+ }
+
+ }
+ if(triggerElement.getPreviousMode()){
+ if(triggerElement.getCompareValue()==null){
+ triggerElement.setCompareValue(currentValue);
+ triggerElement.setReservedCompareValue(currentValue);
+ triggerResultList.add(true);
+ if(logger.isTraceEnabled()) {
+ logger.trace("Initially set compare value to {}", currentValue);
+ }
+ skipComparison=true;
+ }
+ else{
+ refValue = (double) triggerElement.getCompareValue();
+ }
+ }
+ else {
+ refValue = (double) triggerElement.getCompareValue();
+ }
+
+ }
+ catch (Exception e){
+ handleException(e);
+ return false;
+ }
+
+ boolean triggerResult = false;
+ if(!skipComparison) {
+ switch (triggerElement.getComparatorType()) {
+ case EQUAL:
+ triggerResult = isApproximately(currentValue, refValue, TOLERANCE_FLOATING_EQUALITY);
+ break;
+ case UNEQUAL:
+ triggerResult = !isApproximately(currentValue, refValue, TOLERANCE_FLOATING_EQUALITY);
+ break;
+ case SMALLER:
+ triggerResult = currentValue < refValue;
+ break;
+ case SMALLER_EQUAL:
+ triggerResult = currentValue <= refValue;
+ break;
+ case GREATER:
+ triggerResult = currentValue > refValue;
+ break;
+ case GREATER_EQUAL:
+ triggerResult = currentValue >= refValue;
+ break;
+ default:
+ triggerResult = false;
+ }
+ }
+
+ if(triggerResult && triggerElement.getPreviousMode()){
+ triggerElement.setReservedCompareValue(currentValue);
+ if(logger.isTraceEnabled()) {
+ logger.trace("Subcondition matched. Previous value: {}, current compare value {} for Job {}",
+ triggerElement.getReservedCompareValue(),
+ triggerElement.getCompareValue(),
+ triggerElement.getTriggerJob());
+ }
+ }
+ triggerResultList.add(triggerResult);
+
+ }
+
+ }
+ if(triggerResultList.isEmpty()){
+ if(logger.isDebugEnabled()) {
+ logger.debug("No results could be acquired - setting trigger to false");
+ }
+ return false;
+ }
+ if(triggerResultList.size()>1) {
+ logger.trace("{}",triggerResultList);
+ boolean combinedResult=triggerResultList.get(0);
+ for (int countElements = 1; countElements < acquiredValuesList.size(); countElements++) {
+ switch (triggerElementList.get(countElements).getConcatType()){
+ case AND:
+ combinedResult = combinedResult && triggerResultList.get(countElements);
+ break;
+ case OR:
+ combinedResult = combinedResult || triggerResultList.get(countElements);
+ break;
+ default:
+ //should not happen
+ combinedResult = false;
+ }
+ }
+ if(combinedResult) {
+ triggerElementList.forEach(triggerElement -> triggerElement.overrideCompareValue());
+ }
+ return combinedResult;
+ }
+ else{
+ if(triggerResultList.get(0)) {
+ triggerElementList.forEach(triggerElement -> triggerElement.overrideCompareValue());
+ }
+ //return first result because its the only one
+ return triggerResultList.get(0);
}
}
- String exceptionMessage = "Invalid Datatype detected ... should not happen and be catcht earlier - please report";
- throw new ScraperException(exceptionMessage);
}
/**
@@ -329,28 +362,65 @@ public class TriggerConfiguration{
Matcher matcher = TRIGGER_STRATEGY_PATTERN.matcher(jobTriggerStrategy);
if(matcher.matches()){
- String strat = matcher.group("strategy");
+ String triggerStrategy = matcher.group("strategy");
String scheduledMs = matcher.group("scheduledInterval");
- logger.debug("Strategy: {}, scheduled ms: {}",strat,scheduledMs);
+ logger.debug("Strategy: {}, scheduled ms: {}",triggerStrategy,scheduledMs);
String triggerVar = matcher.group("triggerVar");
String comparatorString = matcher.group("comp");
String comparatorVariable = matcher.group("compVar");
- switch (strat){
+ switch (triggerStrategy){
case S_7_TRIGGER_VAR:
+
if(triggerVar ==null || comparatorString==null || comparatorVariable==null){
throw new ScraperException("S7_TRIGGER_VAR trigger strategy needs the trigger-condition - information missing! given configString: "+jobTriggerStrategy);
}
- return new TriggerConfiguration(TriggerType.S7_TRIGGER_VAR,scheduledMs,triggerVar,comparatorString,comparatorVariable,triggeredScrapeJob);
+
+ List<TriggerElement> triggerElements = new ArrayList<>();
+
+ TriggerElement triggerElement = new TriggerElement(
+ comparatorString,
+ null,
+ comparatorVariable,
+ triggerVar,
+ triggerStrategy);
+
+ triggerElement.setTriggerJob(triggeredScrapeJob.getJobName());
+
+ triggerElements.add(triggerElement);
+
+ String concatConn = matcher.group("concatConn");
+ String triggerVar2 = matcher.group("triggerVar2");
+ String comparatorString2 = matcher.group("comp2");
+ String comparatorVariable2 = matcher.group("compVar2");
+
+ if(triggerVar2 != null && comparatorString2 != null && comparatorVariable2 != null && concatConn != null){
+ TriggerElement triggerElement2 = new TriggerElement(
+ comparatorString2,
+ concatConn,
+ comparatorVariable2,
+ triggerVar2,
+ triggerStrategy);
+
+
+ triggerElement2.setTriggerJob(triggeredScrapeJob.getJobName());
+ triggerElements.add(triggerElement2);
+
+ }
+
+ //ToDo add clever Strategy to concat more than two conditions if needed
+
+
+ return new TriggerConfiguration(TriggerType.S7_TRIGGER_VAR,scheduledMs,triggerElements,triggeredScrapeJob);
case SCHEDULED:
if(triggerVar !=null || comparatorString!=null || comparatorVariable!=null){
throw new ScraperException("SCHEDULED trigger strategy must only be used with scheduled interval - nothing more! given configString: "+jobTriggerStrategy);
}
return new TriggerConfiguration(TriggerType.SCHEDULED,scheduledMs);
default:
- throw new ScraperException("Unknown Trigger Strategy "+strat);
+ throw new ScraperException("Unknown Trigger Strategy "+triggerStrategy);
}
@@ -371,16 +441,8 @@ public class TriggerConfiguration{
return scrapeInterval;
}
- String getTriggerVariable() {
- return triggerVariable;
- }
-
- Comparators getComparatorType() {
- return comparatorType;
- }
-
- Object getCompareValue() {
- return compareValue;
+ public List<TriggerElement> getTriggerElementList() {
+ return triggerElementList;
}
/**
@@ -395,7 +457,7 @@ public class TriggerConfiguration{
return Math.abs(self - other) <= within;
}
- public enum Comparators{
+ public enum Comparator {
EQUAL,
UNEQUAL,
GREATER,
@@ -409,4 +471,234 @@ public class TriggerConfiguration{
SCHEDULED,
S7_TRIGGER_VAR
}
+
+ public enum ConcatType {
+ AND,
+ OR
+ }
+
+ public static class TriggerElement{
+ private Comparator comparatorType;
+ private ConcatType concatType;
+ //if trigger should be compared to previous value
+ private Boolean previousMode;
+ private Object compareValue;
+ private PlcField plcField;
+ private String plcFieldString;
+
+ private String uuid;
+
+ private String triggerJob;
+
+ //storage for overwrite if condition matched
+ private Object reservedCompareValue;
+
+ public TriggerElement() {
+ this.comparatorType = null;
+ this.concatType = null;
+ this.previousMode = false;
+ this.compareValue = null;
+ this.plcField = null;
+ this.plcFieldString = null;
+ this.reservedCompareValue = null;
+ this.triggerJob = "Not yet defined";
+ this.uuid = "";
+ }
+
+ public TriggerElement(Comparator comparatorType, ConcatType concatType, Boolean previousMode, Object compareValue, PlcField plcField, String plcFieldString) {
+ this.comparatorType = comparatorType;
+ this.concatType = concatType;
+ this.previousMode = previousMode;
+ this.compareValue = compareValue;
+ this.plcField = plcField;
+ this.plcFieldString = plcFieldString;
+ }
+
+ public TriggerElement(Comparator comparatorType, Object compareValue, PlcField plcField) {
+ this();
+ this.comparatorType = comparatorType;
+ this.compareValue = compareValue;
+ this.plcField = plcField;
+ }
+
+ TriggerElement(String comparator, String concatType, String compareValue, String plcField, String triggerStrategy) throws ScraperException {
+ this();
+ this.plcFieldString = plcField;
+ if(triggerStrategy.equals(S_7_TRIGGER_VAR)){
+ try {
+ this.plcField = S7Field.of(this.plcFieldString);
+ }
+ catch (Exception e){
+ if(logger.isDebugEnabled()) {
+ logger.debug("Exception occurred parsing a S7Field");
+ }
+ throw new ScraperException("Exception on parsing S7Field (" + plcField + "): " + e.getMessage());
+ }
+ this.compareValue = convertCompareValue(compareValue,this.plcField);
+ this.comparatorType = detectComparatorType(comparator);
+ matchTypeAndComparator();
+ }
+
+ this.concatType = detectConcatType(concatType);
+
+ }
+
+
+ /**
+ * parses the ref-value to a given value, as well as checking if ref-value matches to the given data-type
+ * @param compareValue compare-value extracted by regex
+ * @return converted object to needed data-type
+ * @throws ScraperException when something does not match or parsing fails
+ */
+ private Object convertCompareValue(String compareValue, PlcField plcField) throws ScraperException {
+ Class<?> javaDataType = validateDataType(plcField);
+ if(javaDataType.equals(Boolean.class)){
+ switch (compareValue){
+ case "1":
+ case "true":
+ return true;
+ case "0":
+ case "false":
+ return false;
+ default:
+ String exceptionMessage = String.format("No valid compare Value at DataType Boolean for trigger: %s",compareValue);
+ throw new ScraperException(exceptionMessage);
+ }
+ }
+ if(javaDataType.equals(Double.class)
+ || javaDataType.equals(Integer.class)
+ || javaDataType.equals(Long.class)){
+ try {
+ //everything fits to Double for conversion ... so for first step use only double
+ //ToDo if different handling dependent on specific datatype is needed then differ
+ if(PREVIOUS_DEF.equals(compareValue)){
+ this.previousMode=true;
+ return null;
+ }
+ return Double.parseDouble(compareValue);
+ }
+ catch (Exception e){
+ logger.debug(e.getMessage(), e);
+ String exceptionMessage = String.format("No valid compare Value at DataType Numeric for trigger: %s",compareValue);
+ throw new ScraperException(exceptionMessage);
+ }
+ }
+ String exceptionMessage = "Invalid Datatype detected ... should not happen and be catcht earlier - please report";
+ throw new ScraperException(exceptionMessage);
+ }
+
+ /**
+ * converts parsed comparator from regex to ComparatorType
+ * @throws ScraperException when no valid comparator has been used
+ */
+ private Comparator detectComparatorType(String comparator) throws ScraperException {
+ switch (comparator){
+ case "==":
+ return Comparator.EQUAL;
+ case "!=":
+ return Comparator.UNEQUAL;
+ case "<=":
+ return Comparator.SMALLER_EQUAL;
+ case "<":
+ return Comparator.SMALLER;
+ case ">=":
+ return Comparator.GREATER_EQUAL;
+ case ">":
+ return Comparator.GREATER;
+ default:
+ throw new ScraperException("Invalid comparator detected!");
+ }
+ }
+
+ /**
+ * convertes parsed comparator from regex to ComparatorType
+ * @throws ScraperException when no valid comparator has been used
+ */
+ private ConcatType detectConcatType(String concat) throws ScraperException {
+ //concat is not necessary in every case, correct usage is checked later on
+ if(concat==null){
+ return null;
+ }
+ switch (concat){
+ case "AND":
+ return ConcatType.AND;
+ case "OR":
+ return ConcatType.OR;
+ default:
+ throw new ScraperException("Invalid concat between triggerVars detected: "+concat);
+ }
+ }
+
+ /**
+ * matches data-type and comparator for a valid combination
+ * @throws ScraperException when invalid combination is detected
+ */
+ private void matchTypeAndComparator() throws ScraperException {
+ if(validateDataType(this.plcField).equals(Boolean.class)
+ && !(this.comparatorType.equals(Comparator.EQUAL) || this.comparatorType.equals(Comparator.UNEQUAL))){
+ String exceptionMessage = String.format("Trigger-Data-Type (%s) and Comparator (%s) do not match",this.plcField.getDefaultJavaType(),this.comparatorType);
+ throw new ScraperException(exceptionMessage);
+ }
+ //all other combinations are valid
+ }
+
+ Comparator getComparatorType() {
+ return comparatorType;
+ }
+
+ ConcatType getConcatType() {
+ return concatType;
+ }
+
+ Boolean getPreviousMode() {
+ return previousMode;
+ }
+
+ Object getCompareValue() {
+ return compareValue;
+ }
+
+ PlcField getPlcField() {
+ return plcField;
+ }
+
+ String getPlcFieldString() {
+ return plcFieldString;
+ }
+
+ void setCompareValue(Object compareValue) {
+ this.compareValue = compareValue;
+ }
+
+ Object getReservedCompareValue() {
+ return reservedCompareValue;
+ }
+
+ void setReservedCompareValue(Object reservedCompareValue) {
+ this.reservedCompareValue = reservedCompareValue;
+ }
+
+ String getTriggerJob() {
+ return triggerJob;
+ }
+
+ void setTriggerJob(String triggerJob) {
+ this.triggerJob = triggerJob;
+ }
+
+ void overrideCompareValue(){
+ if(this.previousMode && this.reservedCompareValue!=null){
+ logger.info("Compare value overridden, before: {}, now: {}; for Trigger {}",this.compareValue,this.reservedCompareValue,this.triggerJob);
+ this.compareValue = this.reservedCompareValue;
+ }
+ }
+
+ public String getUuid() {
+ return uuid;
+ }
+
+ public void setUuid(String uuid) {
+ this.uuid = uuid;
+ }
+ }
}
\ No newline at end of file
diff --git a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerHandler.java b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerHandler.java
index a2f45bb..98582e0 100644
--- a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerHandler.java
+++ b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerHandler.java
@@ -18,114 +18,6 @@
*/
package org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler;
-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;
-import org.apache.plc4x.java.api.messages.PlcReadResponse;
-import org.apache.plc4x.java.scraper.exception.ScraperException;
-import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScrapeJobImpl;
-import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScraperTask;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-
-/**
- * holds the handler for the regarding trigger-scraper on rising-trigger edge
- */
-public class TriggerHandler {
- private static final Logger LOGGER = LoggerFactory.getLogger(TriggerHandler.class);
- private static final String TRIGGER = "trigger";
-
- private final TriggerConfiguration triggerConfiguration;
- private final TriggeredScraperTask parentScraperTask;
-
- //used to enable trigger only on rising edge
- private boolean lastTriggerState;
-
- public TriggerHandler(String triggerStrategy, TriggeredScrapeJobImpl triggeredScrapeJob,TriggeredScraperTask parentScraperTask) throws ScraperException {
- this.triggerConfiguration = TriggerConfiguration.createConfiguration(triggerStrategy,triggeredScrapeJob);
- this.parentScraperTask = parentScraperTask;
- this.lastTriggerState = false;
- }
-
- /**
- * checks rising edge of trigger event
- * @return true on detection of rising edge, false otherwise
- */
- public boolean checkTrigger(){
- switch (this.triggerConfiguration.getTriggerType()){
- case SCHEDULED:
- //used base scheduling -> trigger is always true
- return true;
- case S7_TRIGGER_VAR:
- return checkS7TriggerVariable();
- default:
- //should not happen
- return false;
- }
- }
-
- /**
- * acquires the given S7Field from S7 and evaluates if trigger is released
- * @return true if rising-edge of trigger is detected, false otherwise
- */
- private boolean checkS7TriggerVariable(){
-
- CompletableFuture<PlcConnection> future = CompletableFuture.supplyAsync(() -> {
- try {
- return parentScraperTask.getDriverManager().getConnection(parentScraperTask.getConnectionString());
- } catch (PlcConnectionException e) {
- LOGGER.warn("Unable to instantiate connection to " + parentScraperTask.getConnectionString(), e);
- throw new PlcRuntimeException(e);
- }
- }, parentScraperTask.getExecutorService());
- PlcConnection connection = null;
- try {
- connection = future.get(parentScraperTask.getRequestTimeoutMs(), TimeUnit.MILLISECONDS);
- LOGGER.trace("Connection to {} established: {}", parentScraperTask.getConnectionString(), connection);
- PlcReadRequest.Builder builder = connection.readRequestBuilder();
- builder.addItem(TRIGGER, triggerConfiguration.getTriggerVariable());
- PlcReadResponse response = builder
- .build()
- .execute()
- .get(parentScraperTask.getRequestTimeoutMs(), TimeUnit.MILLISECONDS);
-
- //check if trigger condition from TriggerConfiguration is fulfilled
- boolean trigger = triggerConfiguration.evaluateTrigger(response.getObject(TRIGGER));
-
- //only trigger scraping of data on rising edge of trigger
- if(trigger && !this.lastTriggerState){
- this.lastTriggerState = true;
- return true;
- }
- else{
- this.lastTriggerState = trigger;
- return false;
- }
-
- } catch (Exception e) {
- // Handle execution exception
- parentScraperTask.handleException(e);
- return false;
- }
- finally {
- if (connection != null) {
- try {
- connection.close();
- } catch (Exception e) {
- LOGGER.warn("Error on closing connection",e);
- // intentionally do nothing
- }
- }
- }
- }
-
-
-
-
-
-
+public interface TriggerHandler {
+ boolean checkTrigger();
}
diff --git a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerHandlerImpl.java b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerHandlerImpl.java
new file mode 100644
index 0000000..432b575
--- /dev/null
+++ b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerHandlerImpl.java
@@ -0,0 +1,112 @@
+/*
+ * 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.plc4x.java.scraper.triggeredscraper.triggerhandler;
+
+import org.apache.plc4x.java.scraper.exception.ScraperException;
+import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScrapeJobImpl;
+import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScraperTask;
+import org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+
+/**
+ * holds the handler for the regarding trigger-scraper on rising-trigger edge
+ */
+public class TriggerHandlerImpl implements TriggerHandler {
+ private static final Logger LOGGER = LoggerFactory.getLogger(TriggerHandlerImpl.class);
+
+ private final TriggerConfiguration triggerConfiguration;
+
+ private final TriggerCollector triggerCollector;
+
+ //used to enable trigger only on rising edge
+ private boolean lastTriggerState;
+
+ public TriggerHandlerImpl(String triggerStrategy, TriggeredScrapeJobImpl triggeredScrapeJob, TriggeredScraperTask parentScraperTask, TriggerCollector triggerCollector) throws ScraperException {
+ this.triggerConfiguration = TriggerConfiguration.createConfiguration(triggerStrategy,triggeredScrapeJob);
+
+ //transmit needed trigger to triggerCollection
+ for(TriggerConfiguration.TriggerElement triggerElement:triggerConfiguration.getTriggerElementList()){
+ triggerElement.setUuid(triggerCollector.submitTrigger(triggerElement.getPlcFieldString(),parentScraperTask.getConnectionString(),this.triggerConfiguration.getScrapeInterval()));
+ }
+
+ this.lastTriggerState = false;
+ this.triggerCollector = triggerCollector;
+ }
+
+ /**
+ * checks rising edge of trigger event
+ * @return true on detection of rising edge, false otherwise
+ */
+ @Override
+ public boolean checkTrigger(){
+ switch (this.triggerConfiguration.getTriggerType()){
+ case SCHEDULED:
+ //used base scheduling -> trigger is always true
+ return true;
+ case S7_TRIGGER_VAR:
+ return checkS7TriggerVariable();
+ default:
+ //should not happen
+ return false;
+ }
+ }
+
+ /**
+ * acquires the given S7Field from S7 and evaluates if trigger is released
+ * @return true if rising-edge of trigger is detected, false otherwise
+ */
+ private boolean checkS7TriggerVariable(){
+
+ List<Object> acquiredValuesList = new ArrayList<>();
+ for(TriggerConfiguration.TriggerElement triggerElement:triggerConfiguration.getTriggerElementList()){
+ try {
+ Object result = triggerCollector.requestResult(triggerElement.getUuid());
+ if(result==null){
+ return false;
+ }
+ acquiredValuesList.add(result);
+ } catch (ScraperException e) {
+ LOGGER.warn("Went wrong",e);
+ }
+ }
+
+ //check if trigger condition from TriggerConfiguration is fulfilled
+ boolean trigger = false;
+ try {
+ trigger = triggerConfiguration.evaluateTrigger(acquiredValuesList);
+ } catch (ScraperException e) {
+ LOGGER.warn("Could not evaluate trigger");
+ }
+
+ //only trigger scraping of data on rising edge of trigger
+ if(trigger && !this.lastTriggerState){
+ this.lastTriggerState = true;
+ return true;
+ }
+ else{
+ this.lastTriggerState = trigger;
+ return false;
+ }
+
+ }
+
+}
diff --git a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/collector/TriggerCollector.java b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/collector/TriggerCollector.java
new file mode 100644
index 0000000..37d2eed
--- /dev/null
+++ b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/collector/TriggerCollector.java
@@ -0,0 +1,61 @@
+/*
+ * 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.plc4x.java.scraper.triggeredscraper.triggerhandler.collector;
+
+import org.apache.plc4x.java.scraper.exception.ScraperException;
+
+/**
+ * defines the interface for implementing a TriggerCollector that handles and acquires all triggerRequests that need a PlcConnection
+ */
+public interface TriggerCollector {
+
+ /**
+ * submits a trigger request to TriggerCollector
+ * @param plcField a (plc) field that is used for triggering procedure
+ * @param plcConnectionString the connection string to the regarding source
+ * @param maxAwaitingTime max awaiting time until request shall be submitted
+ * @return a uuid under that the request is handled internally
+ */
+ String submitTrigger(String plcField, String plcConnectionString, long maxAwaitingTime) throws ScraperException;
+
+ /**
+ * requests the result of submitted plc request with default timeout
+ * @param uuid uuid that represents the request
+ * @return the object acquired by requesting plc instance
+ */
+ Object requestResult(String uuid) throws ScraperException;
+
+ /**
+ * requests the result of submitted plc request
+ * @param uuid uuid that represents the request
+ * @param timeout timeout until response shall be acquired
+ * @return the object acquired by requesting plc instance
+ */
+ Object requestResult(String uuid, long timeout) throws ScraperException;
+
+ /**
+ * starts the acquirement of triggers
+ */
+ void start();
+
+ /**
+ * stops acquirement of triggers
+ */
+ void stop();
+}
diff --git a/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/collector/TriggerCollectorImpl.java b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/collector/TriggerCollectorImpl.java
new file mode 100644
index 0000000..77b95d8
--- /dev/null
+++ b/plc4j/utils/scraper/src/main/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/collector/TriggerCollectorImpl.java
@@ -0,0 +1,316 @@
+/*
+ * 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.plc4x.java.scraper.triggeredscraper.triggerhandler.collector;
+
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.apache.plc4x.java.PlcDriverManager;
+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;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.scraper.Scraper;
+import org.apache.plc4x.java.scraper.exception.ScraperException;
+import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScraperImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.*;
+import java.util.concurrent.*;
+
+/**
+ * default implementation for TriggerCollector
+ */
+public class TriggerCollectorImpl implements TriggerCollector {
+ private static final Logger logger = LoggerFactory.getLogger( TriggerCollectorImpl.class );
+
+ private static final int DEFAULT_SCHEDULED_TRIGGER_INTERVAL = 1000;
+ private static final int FUTURE_TIMEOUT = 1000;
+ private static final int READ_REQUEST_TIMEOUT = 2000;
+ private final PlcDriverManager plcDriverManager;
+ private final Map<String,RequestElement> currentRequestElements;
+ private long schedulerInterval;
+ private final long futureTimeout;
+
+ private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3,
+ new BasicThreadFactory.Builder()
+ .namingPattern("triggercollector-scheduler-thread-%d")
+ .daemon(false)
+ .build()
+ );
+ private final ExecutorService executorService = Executors.newFixedThreadPool(10,
+ new BasicThreadFactory.Builder()
+ .namingPattern("triggercollector-executer-thread-%d")
+ .daemon(true)
+ .build()
+ );
+
+
+ public TriggerCollectorImpl(PlcDriverManager plcDriverManager, long schedulerInterval, long futureTimeout) {
+ this.plcDriverManager = plcDriverManager;
+ this.currentRequestElements = new ConcurrentHashMap<>();
+ this.schedulerInterval = schedulerInterval;
+ this.futureTimeout = futureTimeout;
+ }
+
+ public TriggerCollectorImpl(PlcDriverManager plcDriverManager) {
+ this(plcDriverManager,DEFAULT_SCHEDULED_TRIGGER_INTERVAL, FUTURE_TIMEOUT);
+ }
+
+ /**
+ * submits a trigger request to TriggerCollector
+ *
+ * @param plcField a (plc) field that is used for triggering procedure
+ * @param plcConnectionString the connection string to the regarding source
+ * @param interval max awaiting time until request shall be submitted
+ * @return a uuid under that the request is handled internally
+ */
+ @Override
+ public String submitTrigger(String plcField, String plcConnectionString, long interval) throws ScraperException {
+ String uuid = UUID.randomUUID().toString();
+
+ if(this.schedulerInterval>interval){
+ this.schedulerInterval=interval;
+ }
+
+ RequestElement requestElement = new RequestElement(plcConnectionString,plcField,interval, uuid);
+ if(!currentRequestElements.containsValue(requestElement)){
+ currentRequestElements.put(uuid,requestElement);
+ if(logger.isInfoEnabled()) {
+ logger.info("Received request to: {} for PLC: {}", plcField, plcConnectionString);
+ }
+ return uuid;
+ }
+ else{
+ if(logger.isTraceEnabled()) {
+ logger.trace("Received a placed trigger");
+ }
+ for(RequestElement requestElementFromMap:currentRequestElements.values()){
+ if(requestElementFromMap.equals(requestElement)){
+ //detect shortest interval if trigger used more than once
+ if(requestElementFromMap.getScanIntervalMs()>interval){
+ requestElementFromMap.setScanIntervalMs(interval);
+ }
+ return requestElementFromMap.getUuid();
+ }
+ }
+
+ //should not happen
+ throw new ScraperException(String.format("Could not evaluate UUID for given trigger (%s,%s). Should not happen please report!",plcField,plcConnectionString));
+ }
+
+ }
+
+ /**
+ * acquire all triggers within given interval from definition
+ */
+ private void processActiveTrigger(){
+ LocalDateTime currentTimestamp = LocalDateTime.now();
+ Map<String, PlcReadRequest.Builder> plcReadRequestBuilderMap = new HashMap<>();
+ Map<String, PlcReadResponse> plcReadResponseMap = new HashMap<>();
+ List<RequestElement> activeRequestElements = new ArrayList<>();
+ for(Map.Entry<String,RequestElement> entry:currentRequestElements.entrySet()){
+ if(entry.getValue().getLastAcquirement().isBefore(
+ currentTimestamp
+ .minus(entry.getValue().scanIntervalMs,ChronoUnit.MILLIS))
+ ){
+ String plcConnectionString = entry.getValue().plcConnectionString;
+ if(!plcReadRequestBuilderMap.containsKey(plcConnectionString)){
+ try {
+ PlcConnection plcConnection = TriggeredScraperImpl.getPlcConnection(plcDriverManager,plcConnectionString,executorService,futureTimeout);
+ plcReadRequestBuilderMap.put(plcConnectionString,plcConnection.readRequestBuilder());
+ plcReadRequestBuilderMap.get(plcConnectionString).addItem(entry.getKey(),entry.getValue().getPlcField());
+ activeRequestElements.add(entry.getValue());
+ } catch (InterruptedException e) {
+ logger.warn("Acquirement of PLC-Connection was interrupted",e);
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) {
+ logger.warn("Acquirement of PLC-Connection could not be executed",e);
+ } catch (TimeoutException e) {
+ logger.warn("Acquirement of PLC-Connection was timeouted",e);
+ }
+ }
+ else{
+ plcReadRequestBuilderMap.get(plcConnectionString).addItem(entry.getKey(),entry.getValue().getPlcField());
+ activeRequestElements.add(entry.getValue());
+ }
+ }
+ }
+
+ for(Map.Entry<String,PlcReadRequest.Builder> entry: plcReadRequestBuilderMap.entrySet()){
+ try {
+ plcReadResponseMap.put(entry.getKey(),entry.getValue().build().execute().get(futureTimeout,TimeUnit.MILLISECONDS));
+ } catch (InterruptedException e) {
+ logger.warn("Extraction of PlcResponse was interrupted",e);
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) {
+ logger.warn("Extraction of PlcResponse could not be executed",e);
+ } catch (TimeoutException e) {
+ logger.warn("Extraction of PlcResponse was timeouted",e);
+ }
+ }
+
+ LocalDateTime currentTime = LocalDateTime.now();
+ for(RequestElement requestElement:activeRequestElements){
+ requestElement.setResult(plcReadResponseMap.get(requestElement.getPlcConnectionString()).getObject(requestElement.getUuid()));
+ requestElement.setLastAcquirement(currentTime);
+ }
+ }
+
+ /**
+ * requests the result of submitted plc request with default timeout
+ *
+ * @param uuid uuid that represents the request
+ * @return the object acquired by requesting plc instance
+ */
+ @Override
+ public Object requestResult(String uuid) throws ScraperException {
+ return requestResult(uuid, READ_REQUEST_TIMEOUT);
+ }
+
+ /**
+ * requests the result of submitted plc request
+ *
+ * @param uuid uuid that represents the request
+ * @return the object acquired by requesting plc instance
+ */
+ @Override
+ public Object requestResult(String uuid, long timeout) throws ScraperException {
+ Object result = currentRequestElements.get(uuid).getResult();
+ /*
+ if(result==null){
+ throw new ScraperException("No result acquired yet");
+ }
+ */
+ return result;
+ }
+
+ /**
+ * starts the acquirement of triggers
+ */
+ @Override
+ public void start() {
+ this.scheduler.scheduleAtFixedRate(() -> processActiveTrigger(), 1_000, this.schedulerInterval, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * stops acquirement of triggers
+ */
+ @Override
+ public void stop() {
+ //ToDo stop everything the right way
+ }
+
+ class RequestElement{
+ private String plcConnectionString;
+ private String plcField;
+ private LocalDateTime submitTimeOut;
+ private LocalDateTime lastAcquirement;
+ private Object result;
+ private String uuid;
+ private boolean submitted;
+ private CompletableFuture<PlcReadResponse> plcReadResponse;
+ private long scanIntervalMs;
+
+
+ public RequestElement(String plcConnectionString, String plcField, long scanIntervalMs, String uuid) {
+ this.plcConnectionString = plcConnectionString;
+ this.plcField = plcField;
+ this.submitted = false;
+ this.submitTimeOut = LocalDateTime.now().plus(scanIntervalMs, ChronoUnit.MILLIS);
+ this.uuid = uuid;
+ this.scanIntervalMs = scanIntervalMs;
+ //set initial acquirement to a long time ago
+ this.lastAcquirement = LocalDateTime.of(1,1,1,1,1,1);
+ }
+
+ public String getPlcConnectionString() {
+ return plcConnectionString;
+ }
+
+ public String getPlcField() {
+ return plcField;
+ }
+
+ public LocalDateTime getSubmitTimeOut() {
+ return submitTimeOut;
+ }
+
+ public boolean isSubmitted() {
+ return submitted;
+ }
+
+ public void setSubmitted(boolean submitted) {
+ this.submitted = submitted;
+ }
+
+ public Object getResult() {
+ return result;
+ }
+
+ public void setResult(Object result) {
+ this.result = result;
+ }
+
+ public String getUuid() {
+ return uuid;
+ }
+
+ public CompletableFuture<PlcReadResponse> getPlcReadResponse() {
+ return plcReadResponse;
+ }
+
+ public void setPlcReadResponse(CompletableFuture<PlcReadResponse> plcReadResponse) {
+ this.plcReadResponse = plcReadResponse;
+ }
+
+ public long getScanIntervalMs() {
+ return scanIntervalMs;
+ }
+
+ public void setScanIntervalMs(long scanIntervalMs) {
+ this.scanIntervalMs = scanIntervalMs;
+ }
+
+ public LocalDateTime getLastAcquirement() {
+ return lastAcquirement;
+ }
+
+ public void setLastAcquirement(LocalDateTime lastAcquirement) {
+ this.lastAcquirement = lastAcquirement;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ RequestElement that = (RequestElement) o;
+ return Objects.equals(plcConnectionString, that.plcConnectionString) &&
+ Objects.equals(plcField, that.plcField);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(plcConnectionString, plcField);
+ }
+ }
+
+}
diff --git a/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/TriggeredScraperRunner.java b/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/TriggeredScraperRunner.java
index 3f5461f..5c22893 100644
--- a/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/TriggeredScraperRunner.java
+++ b/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/TriggeredScraperRunner.java
@@ -19,11 +19,15 @@
package org.apache.plc4x.java.scraper;
+import org.apache.plc4x.java.PlcDriverManager;
import org.apache.plc4x.java.scraper.config.triggeredscraper.TriggeredScraperConfiguration;
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.utils.connectionpool.PooledPlcDriverManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,8 +42,12 @@ public class TriggeredScraperRunner {
*/
public static void main(String[] args) throws IOException, ScraperException {
TriggeredScraperConfiguration configuration = TriggeredScraperConfiguration.fromFile("plc4j/utils/scraper/src/test/resources/example_triggered_scraper.yml");
- TriggeredScraperImpl scraper = new TriggeredScraperImpl(configuration, (j, a, m) -> LOGGER.info("Results from {}/{}: {}", j, a, m));
+
+ PlcDriverManager plcDriverManager = new PooledPlcDriverManager();
+ TriggerCollector triggerCollector = new TriggerCollectorImpl(plcDriverManager);
+ TriggeredScraperImpl scraper = new TriggeredScraperImpl(configuration, (j, a, m) -> LOGGER.info("Results from {}/{}: {}", j, a, m),triggerCollector);
scraper.start();
+ triggerCollector.start();
}
}
diff --git a/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/TriggeredScraperRunnerModbus.java b/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/TriggeredScraperRunnerModbus.java
index 3ed4fe3..345b3b8 100644
--- a/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/TriggeredScraperRunnerModbus.java
+++ b/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/TriggeredScraperRunnerModbus.java
@@ -19,10 +19,13 @@
package org.apache.plc4x.java.scraper;
+import org.apache.plc4x.java.PlcDriverManager;
import org.apache.plc4x.java.modbus.connection.ModbusConnectionFactory;
import org.apache.plc4x.java.scraper.config.triggeredscraper.TriggeredScraperConfiguration;
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.TriggerCollectorImpl;
+import org.apache.plc4x.java.utils.connectionpool.PooledPlcDriverManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,14 +42,19 @@ public class TriggeredScraperRunnerModbus {
*/
public static void main(String[] args) throws IOException, ScraperException {
TriggeredScraperConfiguration configuration = TriggeredScraperConfiguration.fromFile("plc4j/utils/scraper/src/test/resources/example_triggered_scraper_modbus.yml");
- TriggeredScraperImpl scraper = new TriggeredScraperImpl(configuration, (j, a, m) -> {
+ PlcDriverManager plcDriverManager = new PooledPlcDriverManager();
+ TriggeredScraperImpl scraper = new TriggeredScraperImpl(
+ configuration,
+ plcDriverManager,
+ (j, a, m) -> {
LOGGER.info("Results from {}/{}: {}", j, a, m);
for(Map.Entry<String, Object> entry:m.entrySet()){
for(Object object:(List<Object>)entry.getValue()){
LOGGER.info("{}",object);
}
}
- });
+ },
+ new TriggerCollectorImpl(plcDriverManager));
scraper.start();
}
diff --git a/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerConfigurationTest.java b/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerConfigurationTest.java
index 3bd4d98..68402a9 100644
--- a/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerConfigurationTest.java
+++ b/plc4j/utils/scraper/src/test/java/org/apache/plc4x/java/scraper/triggeredscraper/triggerhandler/TriggerConfigurationTest.java
@@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory;
import java.util.stream.Stream;
+import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.core.IsEqual.equalTo;
import static org.hamcrest.core.IsNull.notNullValue;
import static org.junit.Assert.assertThat;
@@ -44,15 +45,26 @@ class TriggerConfigurationTest {
private static Stream<Arguments> validTriggerPattern() {
return Stream.of(
- Arguments.of("(S7_TRIGGER_VAR,50,(%I0.1:BOOL)==(true))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparators.EQUAL, true),
- Arguments.of("(S7_TRIGGER_VAR,50,(%I0.1:BOOL)!=(0))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparators.UNEQUAL, false),
- Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBW10:INT)<=(33))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparators.SMALLER_EQUAL, 33.0),
- Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBB10:USINT)>=(33))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparators.GREATER_EQUAL, 33.0),
- Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:DINT)<(33))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparators.SMALLER, 33.0),
- Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(33.3))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparators.GREATER, 33.3),
- Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(33.3))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparators.GREATER, 33.3),
- Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(-1))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparators.GREATER, -1.0),
- Arguments.of("(SCHEDULED,1000)",TriggerConfiguration.TriggerType.SCHEDULED, 1000, null, null)
+
+ Arguments.of("(S7_TRIGGER_VAR,50,(%I0.1:BOOL)==(true))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.EQUAL, true,false,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%I0.1:BOOL)!=(0))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.UNEQUAL, false,false,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBW10:INT)<=(33))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.SMALLER_EQUAL, 33.0,false,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBB10:USINT)>=(33))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.GREATER_EQUAL, 33.0,false,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:DINT)<(33))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.SMALLER, 33.0,false,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(33.3))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.GREATER, 33.3,false,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(33.3))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.GREATER, 33.3,false,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(-1))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.GREATER, -1.0,false,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(PREV))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.GREATER, null,true,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(PREV))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50, TriggerConfiguration.Comparator.GREATER, null,true,null,null,null,null),
+ Arguments.of("(SCHEDULED,1000)",TriggerConfiguration.TriggerType.SCHEDULED, 1000, null, null,null,null,null,null,null),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(PREV)AND(%DB111:DBD20:REAL)>(PREV))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 50,
+ TriggerConfiguration.Comparator.GREATER, null,true,
+ TriggerConfiguration.Comparator.GREATER, null,true,
+ TriggerConfiguration.ConcatType.AND),
+ Arguments.of("(S7_TRIGGER_VAR,200,(%DB111:DBD10:REAL)>(PREV)OR(%DB111:DBD20:REAL)>(PREV))",TriggerConfiguration.TriggerType.S7_TRIGGER_VAR, 200,
+ TriggerConfiguration.Comparator.GREATER, null,true,
+ TriggerConfiguration.Comparator.GREATER, null,true,
+ TriggerConfiguration.ConcatType.OR)
);
}
@@ -69,7 +81,10 @@ class TriggerConfigurationTest {
Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBX10.1:BOOL)<=(true))"),
Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBW10:INT)<=(true))"),
Arguments.of("(MODBUS_TRIGGER_VAR,50)"),
- Arguments.of("(MODBUS_TRIGGER_VAR,50,(%DB111:DBW10:INT)<=(11))")
+ Arguments.of("(MODBUS_TRIGGER_VAR,50,(%DB111:DBW10:INT)<=(11))"),
+ Arguments.of("(S7_TRIGGER_VAR,50,(%DB111:DBD10:REAL)>(prev))"),
+ Arguments.of("(S7_TRIGGER_VAR,200,(%DB111:DBD10:REAL)>(PREV)OR(%DB111:DBD20:INT)>(17))"),
+ Arguments.of("(S7_TRIGGER_VAR,200,(%DB111:DBD10:REAL)>(PREV)AND)")
);
}
@@ -79,8 +94,14 @@ class TriggerConfigurationTest {
void testValidFieldQueryParsing(String triggerConfig,
TriggerConfiguration.TriggerType triggerType,
long scrapeInterval,
- TriggerConfiguration.Comparators comparator,
- Object refValue) {
+ TriggerConfiguration.Comparator comparator1,
+ Object refValue1,
+ Boolean previousMode1,
+ TriggerConfiguration.Comparator comparator2,
+ Object refValue2,
+ Boolean previousMode2,
+ TriggerConfiguration.ConcatType concatType
+ ) {
TriggeredScrapeJobImpl triggeredScrapeJob = Mockito.mock(TriggeredScrapeJobImpl.class);
TriggerConfiguration triggerConfiguration = null;
try {
@@ -92,8 +113,19 @@ class TriggerConfigurationTest {
assertThat(triggerConfiguration, notNullValue());
assertThat(triggerConfiguration.getScrapeInterval(), equalTo(scrapeInterval));
assertThat(triggerConfiguration.getTriggerType(), equalTo(triggerType));
- assertThat(triggerConfiguration.getComparatorType(), equalTo(comparator));
- assertThat(triggerConfiguration.getCompareValue(), equalTo(refValue));
+ if(!triggerConfiguration.getTriggerElementList().isEmpty()) {
+ assertThat(triggerConfiguration.getTriggerElementList().get(0).getComparatorType(), equalTo(comparator1));
+ assertThat(triggerConfiguration.getTriggerElementList().get(0).getCompareValue(), equalTo(refValue1));
+ assertThat(triggerConfiguration.getTriggerElementList().get(0).getPreviousMode(), equalTo(previousMode1));
+ assertThat(triggerConfiguration.getTriggerElementList().get(0).getConcatType(), nullValue());
+
+ if (triggerConfiguration.getTriggerElementList().size() > 1) {
+ assertThat(triggerConfiguration.getTriggerElementList().get(1).getComparatorType(), equalTo(comparator2));
+ assertThat(triggerConfiguration.getTriggerElementList().get(1).getCompareValue(), equalTo(refValue2));
+ assertThat(triggerConfiguration.getTriggerElementList().get(1).getPreviousMode(), equalTo(previousMode2));
+ assertThat(triggerConfiguration.getTriggerElementList().get(1).getConcatType(), equalTo(concatType));
+ }
+ }
}
diff --git a/plc4j/utils/scraper/src/test/resources/example_triggered_scraper.yml b/plc4j/utils/scraper/src/test/resources/example_triggered_scraper.yml
index b3c0c75..e701e12 100644
--- a/plc4j/utils/scraper/src/test/resources/example_triggered_scraper.yml
+++ b/plc4j/utils/scraper/src/test/resources/example_triggered_scraper.yml
@@ -22,23 +22,21 @@ sources:
jobs:
- name: scheduled-demo-job1
- triggerConfig: (SCHEDULED,100)
+ triggerConfig: (SCHEDULED,10000)
sources:
- S7_PI
fields:
test1: '%DB810:DBB0:USINT'
-
- name: triggered-demo-job1
- triggerConfig: (S7_TRIGGER_VAR,10,(%M0.3:BOOL)==(true))
+ triggerConfig: (S7_TRIGGER_VAR,100,(%M0.3:BOOL)==(true))
sources:
- S7_PI
fields:
test1: '%DB810:DBW0:INT'
-
- name: triggered-demo-job2
- triggerConfig: (S7_TRIGGER_VAR,10,(%M0.7:BOOL)==(true))
+ triggerConfig: (S7_TRIGGER_VAR,250,(%M0.7:BOOL)==(true))
sources:
- S7_PI
fields:
@@ -53,4 +51,18 @@ jobs:
test9: '%DB811:DBX280:STRING'
test10: '%DB811:DBB1000:BYTE[8]'
test11: '%DB811:DBX268.3:BOOL'
- test12: '%DB811:DBB270:BYTE[8]'
\ No newline at end of file
+ test12: '%DB811:DBB270:BYTE[8]'
+
+ - name: triggered-demo-job3-prev_value
+ triggerConfig: (S7_TRIGGER_VAR,500,(%M0:USINT)>=(PREV))
+ sources:
+ - S7_PI
+ fields:
+ test1: '%DB810:DBW0:INT'
+
+ - name: triggered-demo-job3-combinded-condition
+ triggerConfig: (S7_TRIGGER_VAR,400,(%M0.6:BOOL)==(true)OR(%M0.7:BOOL)==(true))
+ sources:
+ - S7_PI
+ fields:
+ test1: '%M0:USINT'
\ No newline at end of file