This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 7dd8ca07c3 fix(#2856): improve handling of invalid UNIX timestamps
(#2858)
7dd8ca07c3 is described below
commit 7dd8ca07c38f9ac587511c9396015dad236685ee
Author: Tim <[email protected]>
AuthorDate: Wed May 15 08:03:07 2024 +0200
fix(#2856): improve handling of invalid UNIX timestamps (#2858)
* docs: improve documentation of file stream adapter
* fix(#2856): improve handling of invalid UNIX timestamps
* minor adaptions
* style: fix line length
* fix: make type casting more robust
* fix: improve handling of missing timestamps
* style: add missing whitespace
* tests: adapt E2E test
* style: adapt file formatting
---
.../iiot/protocol/stream/FileReplayAdapter.java | 50 ++++++++++++-------
.../documentation.md | 1 +
.../documentation.md | 56 +++++++++++++++++++++-
.../strings.en | 10 ++--
ui/cypress/README.md | 2 +-
ui/cypress/support/utils/connect/ConnectUtils.ts | 16 +++++--
.../tests/adapter/rules/schemaRules.smoke.spec.ts | 3 +-
ui/cypress/tests/adapter/rules/streamRules.spec.ts | 6 ++-
ui/cypress/tests/adapter/rules/valueRules.ts | 3 +-
9 files changed, 114 insertions(+), 33 deletions(-)
diff --git
a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileReplayAdapter.java
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileReplayAdapter.java
index 04f32643a3..dd0647c6d5 100644
---
a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileReplayAdapter.java
+++
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileReplayAdapter.java
@@ -33,6 +33,7 @@ import
org.apache.streampipes.extensions.management.connect.adapter.parser.xml.X
import org.apache.streampipes.model.AdapterType;
import org.apache.streampipes.model.connect.guess.GuessSchema;
import org.apache.streampipes.model.connect.rules.schema.RenameRuleDescription;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
import org.apache.streampipes.sdk.StaticProperties;
import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
import org.apache.streampipes.sdk.helpers.Alternatives;
@@ -40,7 +41,6 @@ import org.apache.streampipes.sdk.helpers.Filetypes;
import org.apache.streampipes.sdk.helpers.Labels;
import org.apache.streampipes.sdk.helpers.Locales;
import org.apache.streampipes.sdk.helpers.Options;
-import org.apache.streampipes.sdk.utils.Assets;
import org.apache.streampipes.vocabulary.SO;
import org.slf4j.Logger;
@@ -91,7 +91,7 @@ public class FileReplayAdapter implements StreamPipesAdapter {
new XmlParser(),
new ImageParser()
)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
.withLocales(Locales.EN)
.withCategory(AdapterType.Generic)
.requiredFile(
@@ -192,7 +192,7 @@ public class FileReplayAdapter implements
StreamPipesAdapter {
*
* @param extractor An instance of {@code IAdapterParameterExtractor} used
for extracting adapter parameters.
* @throws AdapterException If there are multiple renaming rules detected
affecting the timestamp property,
- * indicating an invalid configuration.
+ * indicating an invalid configuration.
*/
private void determineSourceTimestampField(IAdapterParameterExtractor
extractor) throws AdapterException {
var renamingRulesTimestamp = getRenamingRulesTimestamp(extractor);
@@ -245,23 +245,37 @@ public class FileReplayAdapter implements
StreamPipesAdapter {
.parse(inputStream, (event) -> {
long actualEventTimestamp = -1;
- if (event.get(timestampSourceFieldName) instanceof Long) {
- actualEventTimestamp = (long)
event.get(timestampSourceFieldName);
- } else {
- LOG.error(
- "The timestamp field is not a unix timestamp in ms. Value:
%s"
- .formatted(event.get(timestampSourceFieldName)));
+ var timestampFieldValue = event.get(timestampSourceFieldName);
+ if (timestampFieldValue instanceof Long) {
+ actualEventTimestamp = (Long) timestampFieldValue;
+ } else if (timestampFieldValue instanceof Integer) {
+ actualEventTimestamp = (Integer) timestampFieldValue;
+ } else if (!(timestampFieldValue == null && replaceTimestamp)){
+ adapterRuntimeContext
+ .getLogger()
+ .error(
+ new AdapterException("Timestamp field is not a unix
timestamp in ms, skipping event. Value: %s"
+ .formatted(event.get(timestampSourceFieldName))
+ ));
+ }
+
+ if (actualEventTimestamp == -1 && !replaceTimestamp) {
+ // Do not emit any data if timestamp could not be processed
+ return;
}
- if (timestampLastEvent != -1) {
- long sleepTime = (long) ((actualEventTimestamp -
timestampLastEvent) / speedUp);
- // speed up is set to Float.MAX_VALUE when user selected fastest
option
- if (sleepTime > 0 && speedUp != Float.MAX_VALUE) {
- try {
- Thread.sleep(sleepTime);
- } catch (InterruptedException e) {
- LOG.info("File stream adapter was stopped, the current
replay is interuppted", e);
- }
+ long sleepTime;
+ if (timestampLastEvent != -1 && actualEventTimestamp != -1) {
+ sleepTime = (long) ((actualEventTimestamp - timestampLastEvent)
/ speedUp);
+ } else {
+ sleepTime = 1;
+ }
+ // speed up is set to Float.MAX_VALUE when user selected fastest
option
+ if (sleepTime > 0 && speedUp != Float.MAX_VALUE) {
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException e) {
+ LOG.info("File stream adapter was stopped, the current replay
is interrupted", e);
}
}
diff --git
a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.oi4/documentation.md
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.oi4/documentation.md
index fccb004c9c..1a9bf14903 100644
---
a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.oi4/documentation.md
+++
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.oi4/documentation.md
@@ -79,3 +79,4 @@ The output consists of all values from the selected sensor
type. Below is an exa
"Device status": 0,
"timestamp": 1685525380729
}
+```
diff --git
a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/documentation.md
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/documentation.md
index a03969e8fa..34f3bde459 100644
---
a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/documentation.md
+++
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/documentation.md
@@ -26,7 +26,59 @@
## Description
-Continuously streams the content from a file
+The File Stream Adapter enables continuous streaming of file contents to
Apache StreamPipes, creating a data stream for utilization within StreamPipes.
It's particularly handy when you prefer not to connect directly to the data
source via StreamPipes or for testing and demonstration purposes. Currently, it
supports the following file types:
-***
+- CSV
+- JSON
+- XML
+
+### Example
+
+Suppose we have a CSV file (`temperature.csv`) containing data from a
temperature sensor recording data every second:
+
+```text
+time,temperature
+1715593295000,36.3
+1715593296000,37.5
+1715593297000,37.0
+1715593298000,37.2
+1715593299000,37.2
+1715593210000,37.6
+1715593211000,37.4
+1715593212000,37.5
+1715593213000,37.5
+1715593214000,37.7
+```
+
+When creating a new File Stream Adapter:
+- Upload the file
+- Select `yes` for `Replay Once`
+- Choose `CSV` as the `Format` with `,` as the `delimiter`, check `Header`
+
+After creating the adapter, it will output one line of the CSV as an event
every second.
+Further details on configuration options are provided below.
+
+---
+
+## Configuration
+
+### File
+
+This section determines the file to be streamed by the adapter. Options
include:
+
+- `Choose existing file`: Select from files already present in StreamPipes.
+- `Upload new file`: Upload a new file, also available for other adapters.
Supports `.csv`, `.json`, and `.xml` file types.
+
+### Overwrite file time
+Enable this option to always pass the current system time as the timestamp
when emitting an event. If your file lacks timestamp information, this should
be enabled. Conversely, if your file has timestamp information, enabling this
option will overwrite it with the current system time. By default, this option
is disabled, leaving timestamp information unaffected.
+
+### Replay Once
+Distinguishes between replaying all data contained in the file only once or in
a loop until the adapter is manually stopped.
+If enabled, this will cause events from the file to be emitted multiple times.
In this case, it is recommended to enable `Overwrite file time` if the
resulting stream is to be persisted in StreamPipes, otherwise existing events
with the same timestamp will be overwritten.
+
+### Replay Speed
+Configures the event frequency:
+- **Keep original time**: Events are emitted based on the timestamp
information in the file.
+- **Fastest**: All data in the file is replayed as quickly as possible, with
no waiting time.
+- **Speed Up Factor**: Adjusts the waiting time of the adapter based on the
provided speed up factor, considering the time between two events in the file.
diff --git
a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/strings.en
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/strings.en
index 1f6fecb2d8..86a6a38d31 100644
---
a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/strings.en
+++
b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.file/strings.en
@@ -17,16 +17,16 @@
org.apache.streampipes.connect.iiot.protocol.stream.file.title=File Stream
-org.apache.streampipes.connect.iiot.protocol.stream.file.description=Continuously
streams the content from a file.
+org.apache.streampipes.connect.iiot.protocol.stream.file.description=allows
you to continuously stream the contents of a file to Apache StreamPipes,
creating a data stream that allows you to use the data within StreamPipes
filePath.title=File
-filePath.description=File Path
+filePath.description=Choose between an already existing file or upload a new
file
-replaceTimestamp.title=Use current time
-replaceTimestamp.description=Replace Event Time with Current Timestamp
+replaceTimestamp.title=Overwrite file time
+replaceTimestamp.description=Always use the current time as the timestamp when
emitting the event, regardless of any time value in the file. Needs to be
activated if file does not contain a timestamp.
speed.title=Replay Speed
-speed.description=
+speed.description=Determine the event frequency.
speedUp.title=Speed Up
speedUp.description=original = 1; speedup 2x = 2; half speed = 0.5
diff --git a/ui/cypress/README.md b/ui/cypress/README.md
index 075cbaeaef..faf4e25194 100644
--- a/ui/cypress/README.md
+++ b/ui/cypress/README.md
@@ -51,7 +51,7 @@ This folder contains a WIP framework for automated E2E tests
of StreamPipes.
## Automated test runs
-- Each night the whole test suite is run within github actions
+- Each night the whole test suite is run within GitHub actions
- See:
- To add a test to the test suite add the suffix .spec.ts
- Each PR triggers the smoke tests to detect errors before the branch is
merged into the development
diff --git a/ui/cypress/support/utils/connect/ConnectUtils.ts
b/ui/cypress/support/utils/connect/ConnectUtils.ts
index 5a8a77a57c..cfe06bfc6a 100644
--- a/ui/cypress/support/utils/connect/ConnectUtils.ts
+++ b/ui/cypress/support/utils/connect/ConnectUtils.ts
@@ -274,7 +274,9 @@ export class ConnectUtils {
);
}
- public static setUpPreprocessingRuleTest(): AdapterInput {
+ public static setUpPreprocessingRuleTest(
+ overwriteTimestamp: boolean,
+ ): AdapterInput {
const adapterConfiguration = AdapterBuilder.create('File_Stream')
.setStoreInDataLake()
.setTimestampProperty('timestamp')
@@ -287,8 +289,16 @@ export class ConnectUtils {
.setName('Adapter to test rules')
.setFormat('csv')
.addFormatInput('input', ConnectBtns.csvDelimiter(), ';')
- .addFormatInput('checkbox', ConnectBtns.csvHeader(), 'check')
- .build();
+ .addFormatInput('checkbox', ConnectBtns.csvHeader(), 'check');
+
+ if (overwriteTimestamp) {
+ adapterConfiguration.addProtocolInput(
+ 'checkbox',
+ 'replaceTimestamp',
+ 'check',
+ );
+ }
+ adapterConfiguration = adapterConfiguration.build();
ConnectUtils.goToConnect();
ConnectUtils.goToNewAdapterPage();
diff --git a/ui/cypress/tests/adapter/rules/schemaRules.smoke.spec.ts
b/ui/cypress/tests/adapter/rules/schemaRules.smoke.spec.ts
index 3399ffd41b..39462cfc06 100644
--- a/ui/cypress/tests/adapter/rules/schemaRules.smoke.spec.ts
+++ b/ui/cypress/tests/adapter/rules/schemaRules.smoke.spec.ts
@@ -27,7 +27,8 @@ describe('Connect schema rule transformations', () => {
});
it('Perform Test', () => {
- const adapterConfiguration = ConnectUtils.setUpPreprocessingRuleTest();
+ const adapterConfiguration =
+ ConnectUtils.setUpPreprocessingRuleTest(true);
// Add static value to event
ConnectEventSchemaUtils.addStaticProperty('staticPropertyName', 'id1');
diff --git a/ui/cypress/tests/adapter/rules/streamRules.spec.ts
b/ui/cypress/tests/adapter/rules/streamRules.spec.ts
index 48867115d6..dbfab9dab7 100644
--- a/ui/cypress/tests/adapter/rules/streamRules.spec.ts
+++ b/ui/cypress/tests/adapter/rules/streamRules.spec.ts
@@ -27,7 +27,8 @@ describe('Connect aggregation rule transformations', () => {
});
it('Perform Test', () => {
- const adapterConfiguration = ConnectUtils.setUpPreprocessingRuleTest();
+ const adapterConfiguration =
+ ConnectUtils.setUpPreprocessingRuleTest(false);
ConnectEventSchemaUtils.markPropertyAsTimestamp('timestamp');
ConnectEventSchemaUtils.finishEventSchemaConfiguration();
@@ -50,7 +51,8 @@ describe('Remove duplicates rule transformations', () => {
});
it('Perform Test', () => {
- const adapterConfiguration = ConnectUtils.setUpPreprocessingRuleTest();
+ const adapterConfiguration =
+ ConnectUtils.setUpPreprocessingRuleTest(false);
ConnectEventSchemaUtils.markPropertyAsTimestamp('timestamp');
ConnectEventSchemaUtils.finishEventSchemaConfiguration();
diff --git a/ui/cypress/tests/adapter/rules/valueRules.ts
b/ui/cypress/tests/adapter/rules/valueRules.ts
index d53de16de0..60ffdbf21a 100644
--- a/ui/cypress/tests/adapter/rules/valueRules.ts
+++ b/ui/cypress/tests/adapter/rules/valueRules.ts
@@ -27,7 +27,8 @@ describe('Connect value rule transformations', () => {
});
it('Perform Test', () => {
- const adapterConfiguration = ConnectUtils.setUpPreprocessingRuleTest();
+ const adapterConfiguration =
+ ConnectUtils.setUpPreprocessingRuleTest(false);
// Edit timestamp property
ConnectEventSchemaUtils.editTimestampProperty(
