This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch
4442-add-adapter-to-import-wincc-archived-alarm-messages
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/4442-add-adapter-to-import-wincc-archived-alarm-messages by this
push:
new d8cc65aa30 feat(#4442): Add WinCC archived alarm adapter
d8cc65aa30 is described below
commit d8cc65aa30d52e30e6c2221e3b845d95f8f525cc
Author: Dominik Riemer <[email protected]>
AuthorDate: Wed May 6 23:36:16 2026 +0200
feat(#4442): Add WinCC archived alarm adapter
---
streampipes-extensions/pom.xml | 1 +
.../streampipes-connectors-filewatcher/pom.xml | 80 ++++++
.../FileWatcherConnectorsModuleExport.java | 46 +++
.../adapter/WinCCAlarmArchiveAdapter.java | 264 +++++++++++++++++
.../filewatcher/adapter/WinCCAlarmEventMapper.java | 126 ++++++++
.../adapter/WinCCTimestampConverter.java | 121 ++++++++
.../filewatcher/model/CsvParserSettings.java | 46 +++
.../filewatcher/model/FileFingerprint.java | 83 ++++++
.../filewatcher/model/FileWatcherCheckpoint.java | 77 +++++
.../filewatcher/model/FileWatcherConfig.java | 30 ++
.../filewatcher/runtime/CsvFileReader.java | 145 ++++++++++
.../filewatcher/runtime/EventDelayExecutor.java | 25 ++
.../filewatcher/runtime/EventMapper.java | 31 ++
.../filewatcher/runtime/FileSetWatcher.java | 317 +++++++++++++++++++++
.../connectors/filewatcher/runtime/FileSlot.java | 26 ++
.../filewatcher/runtime/FileWatchReadResult.java | 24 ++
.../runtime/FileWatcherCheckpointStore.java | 70 +++++
.../documentation.md | 45 +++
.../strings.en | 28 ++
.../adapter/WinCCAlarmEventMapperTest.java | 74 +++++
.../adapter/WinCCTimestampConverterTest.java | 50 ++++
.../filewatcher/runtime/CsvFileReaderTest.java | 79 +++++
.../filewatcher/runtime/FileSetWatcherTest.java | 180 ++++++++++++
.../streampipes-extensions-all-iiot/pom.xml | 5 +
.../extensions/all/iiot/AllExtensionsIIoTInit.java | 2 +
.../streampipes-extensions-all-jvm/pom.xml | 5 +
.../extensions/all/jvm/AllExtensionsInit.java | 2 +
27 files changed, 1982 insertions(+)
diff --git a/streampipes-extensions/pom.xml b/streampipes-extensions/pom.xml
index d540061ea2..749c3bbf24 100644
--- a/streampipes-extensions/pom.xml
+++ b/streampipes-extensions/pom.xml
@@ -35,6 +35,7 @@
<module>streampipes-connect-adapters-iiot</module>
<module>streampipes-connectors-camel</module>
+ <module>streampipes-connectors-filewatcher</module>
<module>streampipes-connectors-influx</module>
<module>streampipes-connectors-kafka</module>
<module>streampipes-connectors-mqtt</module>
diff --git a/streampipes-extensions/streampipes-connectors-filewatcher/pom.xml
b/streampipes-extensions/streampipes-connectors-filewatcher/pom.xml
new file mode 100644
index 0000000000..4627d76eba
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-filewatcher/pom.xml
@@ -0,0 +1,80 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ ~
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-extensions</artifactId>
+ <version>0.99.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>streampipes-connectors-filewatcher</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-extensions-api</artifactId>
+ <version>0.99.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-extensions-management</artifactId>
+ <version>0.99.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-sdk</artifactId>
+ <version>0.99.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-serializers-json</artifactId>
+ <version>0.99.0-SNAPSHOT</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.opencsv</groupId>
+ <artifactId>opencsv</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/FileWatcherConnectorsModuleExport.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/FileWatcherConnectorsModuleExport.java
new file mode 100644
index 0000000000..d87c1777d8
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/FileWatcherConnectorsModuleExport.java
@@ -0,0 +1,46 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher;
+
+import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter;
+import org.apache.streampipes.extensions.api.declarer.IExtensionModuleExport;
+import org.apache.streampipes.extensions.api.migration.IModelMigrator;
+import org.apache.streampipes.extensions.api.pe.IStreamPipesPipelineElement;
+import
org.apache.streampipes.extensions.connectors.filewatcher.adapter.WinCCAlarmArchiveAdapter;
+
+import java.util.Collections;
+import java.util.List;
+
+public class FileWatcherConnectorsModuleExport implements
IExtensionModuleExport {
+
+ @Override
+ public List<StreamPipesAdapter> adapters() {
+ return List.of(new WinCCAlarmArchiveAdapter());
+ }
+
+ @Override
+ public List<IStreamPipesPipelineElement<?>> pipelineElements() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List<IModelMigrator<?, ?>> migrators() {
+ return Collections.emptyList();
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmArchiveAdapter.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmArchiveAdapter.java
new file mode 100644
index 0000000000..21391d0218
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmArchiveAdapter.java
@@ -0,0 +1,264 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.adapter;
+
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.extensions.api.connect.IAdapterConfiguration;
+import org.apache.streampipes.extensions.api.connect.IEventCollector;
+import org.apache.streampipes.extensions.api.connect.IPullAdapter;
+import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter;
+import
org.apache.streampipes.extensions.api.connect.context.IAdapterGuessSchemaContext;
+import
org.apache.streampipes.extensions.api.connect.context.IAdapterRuntimeContext;
+import
org.apache.streampipes.extensions.api.extractor.IAdapterParameterExtractor;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.CsvParserSettings;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.FileWatcherConfig;
+import
org.apache.streampipes.extensions.connectors.filewatcher.runtime.CsvFileReader;
+import
org.apache.streampipes.extensions.connectors.filewatcher.runtime.FileSetWatcher;
+import
org.apache.streampipes.extensions.connectors.filewatcher.runtime.FileWatcherCheckpointStore;
+import
org.apache.streampipes.extensions.management.connect.PullAdapterScheduler;
+import
org.apache.streampipes.extensions.management.connect.adapter.parser.CsvParser;
+import
org.apache.streampipes.extensions.management.connect.adapter.util.PollingSettings;
+import org.apache.streampipes.model.connect.guess.SampleData;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
+import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
+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.helpers.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+
+public class WinCCAlarmArchiveAdapter implements StreamPipesAdapter,
IPullAdapter {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(WinCCAlarmArchiveAdapter.class);
+
+ public static final String ID =
"org.apache.streampipes.connectors.wincc.alarm.archive";
+
+ private static final String DIRECTORY_PATH = "directory-path";
+ private static final String ARCHIVE_BASE_NAME = "archive-base-name";
+ private static final String SEGMENTED_CIRCULAR_LOG_ENABLED =
"segmented-circular-log-enabled";
+ private static final String ARCHIVE_SEGMENT_COUNT = "archive-segment-count";
+ private static final String POLL_INTERVAL_SECONDS = "poll-interval-seconds";
+ private static final String INTER_EVENT_DELAY_MS = "inter-event-delay-ms";
+
+ private static final String SEGMENTED_CIRCULAR_LOG_ON =
"segmented-circular-log-on";
+ private static final String SEGMENTED_CIRCULAR_LOG_OFF =
"segmented-circular-log-off";
+ private static final CsvParserSettings WINCC_CSV_SETTINGS = new
CsvParserSettings(true, ';');
+ private static final WinCCAlarmEventMapper EVENT_MAPPER = new
WinCCAlarmEventMapper();
+
+ private PullAdapterScheduler scheduler;
+ private FileSetWatcher watcher;
+ private String adapterElementId;
+ private IEventCollector collector;
+ private PollingSettings pollingSettings;
+
+ @Override
+ public IAdapterConfiguration declareConfig() {
+ return AdapterConfigurationBuilder
+ .create(ID, 0, WinCCAlarmArchiveAdapter::new)
+ .withAssets(ExtensionAssetType.DOCUMENTATION)
+ .withLocales(Locales.EN)
+ .requiredTextParameter(Labels.from(
+ DIRECTORY_PATH,
+ "Archive directory",
+ "Absolute path to the WinCC archive directory that contains the
exported alarm CSV files."
+ ))
+ .requiredTextParameter(Labels.from(
+ ARCHIVE_BASE_NAME,
+ "Archive base name",
+ "Base file name of the WinCC alarm archive, for example
Meldungsarchiv."
+ ))
+ .requiredSingleValueSelection(
+ Labels.from(
+ SEGMENTED_CIRCULAR_LOG_ENABLED,
+ "Segmented circular log",
+ "Whether WinCC writes the alarm archive as a segmented
circular log with rotating CSV segments."
+ ),
+ Options.from(
+ new Tuple2<>("Enabled", SEGMENTED_CIRCULAR_LOG_ON),
+ new Tuple2<>("Disabled", SEGMENTED_CIRCULAR_LOG_OFF)
+ )
+ )
+ .requiredIntegerParameter(Labels.from(
+ ARCHIVE_SEGMENT_COUNT,
+ "Segment count",
+ "Number of archive segments in segmented circular log mode.
Ignored when the segmented circular log is disabled."
+ ))
+ .requiredIntegerParameter(Labels.from(
+ POLL_INTERVAL_SECONDS,
+ "Polling interval (seconds)",
+ "How often the WinCC archive files should be scanned for new alarm
entries."
+ ))
+ .requiredIntegerParameter(Labels.from(
+ INTER_EVENT_DELAY_MS,
+ "Inter-event delay (ms)",
+ "Delay between replayed alarm events in milliseconds. Use 0 to
disable throttling."
+ ), 0)
+ .buildConfiguration();
+ }
+
+ @Override
+ public void onAdapterStarted(IAdapterParameterExtractor extractor,
+ IEventCollector collector,
+ IAdapterRuntimeContext adapterRuntimeContext)
throws AdapterException {
+ FileWatcherConfig config = toConfig(extractor);
+ this.adapterElementId = extractor.getAdapterDescription().getElementId();
+ this.collector = collector;
+ this.pollingSettings = PollingSettings.from(TimeUnit.SECONDS,
config.pollIntervalSeconds());
+ this.watcher = new FileSetWatcher(
+ config,
+ new FileWatcherCheckpointStore(),
+ new CsvFileReader(),
+ EVENT_MAPPER
+ );
+ LOG.debug(
+ "Starting WinCC alarm archive adapter '{}': directory='{}',
pattern='{}', pollIntervalSeconds={}, "
+ + "singleFileGrowthMode={}, interEventDelayMs={}.",
+ adapterElementId,
+ config.directory(),
+ config.filePattern().pattern(),
+ config.pollIntervalSeconds(),
+ config.singleFileGrowthMode(),
+ config.interEventDelayMs()
+ );
+ this.scheduler = new PullAdapterScheduler();
+ this.scheduler.schedule(this, adapterElementId);
+ }
+
+ @Override
+ public void onAdapterStopped(IAdapterParameterExtractor extractor,
+ IAdapterRuntimeContext adapterRuntimeContext) {
+ if (scheduler != null) {
+ scheduler.shutdown();
+ }
+ }
+
+ @Override
+ public SampleData onSampleDataRequested(IAdapterParameterExtractor extractor,
+ IAdapterGuessSchemaContext
adapterGuessSchemaContext) throws AdapterException {
+ FileWatcherConfig config = toConfig(extractor);
+ try (var files = Files.list(config.directory())) {
+ Path sampleFile = files
+ .filter(Files::isRegularFile)
+ .filter(path ->
config.filePattern().matcher(path.getFileName().toString()).matches())
+ .sorted()
+ .findFirst()
+ .orElseThrow(() -> new AdapterException("No matching WinCC alarm
archive files found in " + config.directory()));
+
+ try (var inputStream = Files.newInputStream(sampleFile)) {
+ Map<String, Object> rawSample = new CsvParser(true,
';').getSampleData(inputStream).getSamples().get(0);
+ return
org.apache.streampipes.sdk.builder.adapter.SampleDataBuilder.create()
+ .sample(EVENT_MAPPER.map(rawSample))
+ .build();
+ }
+ } catch (IOException e) {
+ throw new AdapterException("Could not load sample data from directory "
+ config.directory(), e);
+ }
+ }
+
+ @Override
+ public void pullData() {
+ try {
+ watcher.poll(adapterElementId, collector);
+ } catch (IOException | RuntimeException e) {
+ throw new CompletionException("Could not poll files from WinCC alarm
archive adapter", e);
+ }
+ }
+
+ @Override
+ public PollingSettings getPollingInterval() {
+ return pollingSettings;
+ }
+
+ private FileWatcherConfig toConfig(IAdapterParameterExtractor extractor)
throws AdapterException {
+ var staticPropertyExtractor = extractor.getStaticPropertyExtractor();
+ var directory =
Path.of(staticPropertyExtractor.singleValueParameter(DIRECTORY_PATH,
String.class));
+ if (!Files.isDirectory(directory)) {
+ throw new AdapterException("Configured directory does not exist or is
not a directory: " + directory);
+ }
+
+ var baseName =
normalizeBaseName(staticPropertyExtractor.singleValueParameter(ARCHIVE_BASE_NAME,
String.class));
+ var segmentedCircularLogMode =
+
staticPropertyExtractor.selectedSingleValueInternalName(SEGMENTED_CIRCULAR_LOG_ENABLED,
String.class);
+ var segmentCount =
staticPropertyExtractor.singleValueParameter(ARCHIVE_SEGMENT_COUNT,
Integer.class);
+ var interval =
staticPropertyExtractor.singleValueParameter(POLL_INTERVAL_SECONDS,
Integer.class);
+ var interEventDelayMs =
staticPropertyExtractor.singleValueParameter(INTER_EVENT_DELAY_MS,
Integer.class);
+ if (interEventDelayMs < 0) {
+ throw new AdapterException("Inter-event delay must be greater than or
equal to 0.");
+ }
+
+ Pattern filePattern =
SEGMENTED_CIRCULAR_LOG_ON.equals(segmentedCircularLogMode)
+ ? buildSegmentPattern(baseName, segmentCount)
+ : buildSingleSegmentPattern(baseName);
+
+ return new FileWatcherConfig(
+ directory,
+ filePattern,
+ WINCC_CSV_SETTINGS,
+ interval,
+ SEGMENTED_CIRCULAR_LOG_OFF.equals(segmentedCircularLogMode),
+ interEventDelayMs
+ );
+ }
+
+ private String normalizeBaseName(String configuredBaseName) {
+ String trimmedBaseName = configuredBaseName.trim();
+ if (trimmedBaseName.endsWith(".csv")) {
+ return trimmedBaseName.substring(0, trimmedBaseName.length() - 4);
+ }
+
+ return trimmedBaseName;
+ }
+
+ private Pattern buildSingleSegmentPattern(String baseName) {
+ return Pattern.compile(Pattern.quote(baseName) + "1\\.csv",
Pattern.CASE_INSENSITIVE);
+ }
+
+ private Pattern buildSegmentPattern(String baseName, int segmentCount)
throws AdapterException {
+ if (segmentCount < 1) {
+ throw new AdapterException("Segment count must be at least 1 when the
segmented circular log is enabled.");
+ }
+
+ return Pattern.compile(
+ Pattern.quote(baseName) + "(" + buildAllowedSegments(segmentCount) +
")\\.csv",
+ Pattern.CASE_INSENSITIVE
+ );
+ }
+
+ private String buildAllowedSegments(int segmentCount) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 1; i <= segmentCount; i++) {
+ if (i > 1) {
+ builder.append("|");
+ }
+ builder.append(i);
+ }
+
+ return builder.toString();
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmEventMapper.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmEventMapper.java
new file mode 100644
index 0000000000..ebca63feda
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmEventMapper.java
@@ -0,0 +1,126 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.adapter;
+
+import
org.apache.streampipes.extensions.connectors.filewatcher.runtime.EventMapper;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class WinCCAlarmEventMapper implements EventMapper {
+
+ @Override
+ public Map<String, Object> map(Map<String, Object> rawEvent) {
+ Map<String, Object> normalizedEvent = new LinkedHashMap<>();
+
+ normalizedEvent.put("timestamp",
+ WinCCTimestampConverter.toUnixTimestampMillis(rawEvent.get("Time_ms"),
rawEvent.get("TimeString")));
+ normalizedEvent.put("timestamp_ms", asString(rawEvent.get("Time_ms")));
+ normalizedEvent.put("timestamp_string",
asString(rawEvent.get("TimeString")));
+ normalizedEvent.put("message_text", asString(rawEvent.get("MsgText")));
+ normalizedEvent.put("message_number",
toInteger(rawEvent.get("MsgNumber")));
+ normalizedEvent.put("alarm_procedure_code",
toInteger(rawEvent.get("MsgProc")));
+ normalizedEvent.put("alarm_procedure",
alarmProcedure(rawEvent.get("MsgProc")));
+ normalizedEvent.put("state_after_code",
toInteger(rawEvent.get("StateAfter")));
+ normalizedEvent.put("state_after", stateAfter(rawEvent.get("StateAfter")));
+ normalizedEvent.put("alarm_class_code",
toInteger(rawEvent.get("MsgClass")));
+ normalizedEvent.put("alarm_class", alarmClass(rawEvent.get("MsgClass")));
+ normalizedEvent.put("plc", asString(rawEvent.get("PLC")));
+
+ for (int i = 1; i <= 8; i++) {
+ normalizedEvent.put("parameter_" + i, asString(rawEvent.get("Var" + i)));
+ }
+
+ // Keep only the original timestamp representations in addition to the
normalized fields.
+ normalizedEvent.put("raw_Time_ms", asString(rawEvent.get("Time_ms")));
+ normalizedEvent.put("raw_TimeString",
asString(rawEvent.get("TimeString")));
+ return normalizedEvent;
+ }
+
+ private String alarmProcedure(Object code) {
+ return switch (toInt(code)) {
+ case 1 -> "system_event";
+ case 2 -> "alarm_bit_operation";
+ case 3 -> "alarm_s";
+ case 4 -> "diagnostics_event";
+ case 7 -> "analog_alarm";
+ case 9 -> "program_alarm";
+ case 100 -> "alarm_bit_fault";
+ default -> "unknown";
+ };
+ }
+
+ private String stateAfter(Object code) {
+ return switch (toInt(code)) {
+ case 0 -> "incoming_outgoing";
+ case 1 -> "incoming";
+ case 2 -> "incoming_acknowledged_outgoing";
+ case 3 -> "incoming_acknowledged";
+ case 4 -> "pending_after_plc_reset";
+ case 6 -> "incoming_outgoing_acknowledged";
+ default -> "unknown";
+ };
+ }
+
+ private String alarmClass(Object code) {
+ int numericCode = toInt(code);
+ return switch (numericCode) {
+ case 0 -> "no_alarm_class";
+ case 1 -> "errors";
+ case 2 -> "warnings";
+ case 3 -> "system";
+ case 4 -> "diagnostic_events";
+ default -> numericCode >= 64 ? "user_defined" : "unknown";
+ };
+ }
+
+ private int toInt(Object code) {
+ Integer value = toInteger(code);
+ if (value != null) {
+ return value;
+ }
+
+ return -1;
+ }
+
+ private Integer toInteger(Object code) {
+ if (code instanceof Number number) {
+ return number.intValue();
+ }
+
+ if (code instanceof String stringValue) {
+ String trimmedValue = stringValue.trim();
+ if (trimmedValue.isEmpty()) {
+ return null;
+ }
+
+ try {
+ return Integer.parseInt(trimmedValue);
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ return null;
+ }
+
+ private String asString(Object value) {
+ return value == null ? "" : String.valueOf(value);
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCTimestampConverter.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCTimestampConverter.java
new file mode 100644
index 0000000000..056fe03432
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCTimestampConverter.java
@@ -0,0 +1,121 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.adapter;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.util.List;
+
+public final class WinCCTimestampConverter {
+
+ private static final BigDecimal EXCEL_UNIX_EPOCH_OFFSET_DAYS =
BigDecimal.valueOf(25569L);
+ private static final BigDecimal MILLIS_PER_DAY =
BigDecimal.valueOf(86_400_000L);
+ private static final BigDecimal SCALED_DAY_DIVISOR =
BigDecimal.valueOf(1_000_000L);
+ private static final List<DateTimeFormatter> TIME_STRING_FORMATS = List.of(
+ DateTimeFormatter.ofPattern("dd.MM.yy HH:mm:ss"),
+ DateTimeFormatter.ofPattern("dd.MM.yy HH:mm")
+ );
+
+ private WinCCTimestampConverter() {
+ }
+
+ public static Long toUnixTimestampMillis(Object winccTimestamp,
+ Object timeString) {
+ BigDecimal serialDays = normalizeSerialDays(winccTimestamp);
+ if (serialDays != null) {
+ return serialDays
+ .subtract(EXCEL_UNIX_EPOCH_OFFSET_DAYS)
+ .multiply(MILLIS_PER_DAY)
+ .setScale(0, RoundingMode.HALF_UP)
+ .longValue();
+ }
+
+ return parseTimeString(timeString);
+ }
+
+ private static BigDecimal normalizeSerialDays(Object winccTimestamp) {
+ BigDecimal rawValue = toBigDecimal(winccTimestamp);
+ if (rawValue == null) {
+ return null;
+ }
+
+ if (isScaledDayValue(rawValue, winccTimestamp)) {
+ return rawValue.divide(SCALED_DAY_DIVISOR, 10, RoundingMode.HALF_UP);
+ }
+
+ return rawValue;
+ }
+
+ private static boolean isScaledDayValue(BigDecimal rawValue,
+ Object originalValue) {
+ if (originalValue instanceof String stringValue) {
+ if (stringValue.contains(",") || stringValue.contains(".")) {
+ return false;
+ }
+ }
+
+ return rawValue.compareTo(BigDecimal.valueOf(1_000_000L)) > 0;
+ }
+
+ private static BigDecimal toBigDecimal(Object winccTimestamp) {
+ if (winccTimestamp instanceof Number number) {
+ return BigDecimal.valueOf(number.doubleValue());
+ }
+
+ if (winccTimestamp instanceof String stringValue) {
+ String normalizedValue = stringValue.trim().replace(',', '.');
+ if (normalizedValue.isEmpty()) {
+ return null;
+ }
+
+ try {
+ return new BigDecimal(normalizedValue);
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ return null;
+ }
+
+ private static Long parseTimeString(Object timeString) {
+ if (!(timeString instanceof String stringValue)) {
+ return null;
+ }
+
+ String trimmedValue = stringValue.trim();
+ if (trimmedValue.isEmpty()) {
+ return null;
+ }
+
+ for (DateTimeFormatter formatter : TIME_STRING_FORMATS) {
+ try {
+ LocalDateTime dateTime = LocalDateTime.parse(trimmedValue, formatter);
+ return
dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+ } catch (DateTimeParseException ignored) {
+ }
+ }
+
+ return null;
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/CsvParserSettings.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/CsvParserSettings.java
new file mode 100644
index 0000000000..89fcd9c22e
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/CsvParserSettings.java
@@ -0,0 +1,46 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.model;
+
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.extensions.api.connect.IParser;
+import
org.apache.streampipes.extensions.management.connect.adapter.parser.CsvParser;
+
+import java.lang.reflect.Field;
+
+public record CsvParserSettings(boolean hasHeader, char delimiter) {
+
+ public static CsvParserSettings from(IParser parser) throws AdapterException
{
+ if (!(parser instanceof CsvParser csvParser)) {
+ throw new AdapterException("The file watcher currently supports only the
CSV parser.");
+ }
+
+ try {
+ Field headerField = CsvParser.class.getDeclaredField("header");
+ headerField.setAccessible(true);
+
+ Field delimiterField = CsvParser.class.getDeclaredField("delimiter");
+ delimiterField.setAccessible(true);
+
+ return new CsvParserSettings(headerField.getBoolean(csvParser),
delimiterField.getChar(csvParser));
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new AdapterException("Could not extract CSV parser settings.", e);
+ }
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileFingerprint.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileFingerprint.java
new file mode 100644
index 0000000000..16a3093f97
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileFingerprint.java
@@ -0,0 +1,83 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.model;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class FileFingerprint implements Serializable {
+
+ private long size;
+ private long lastModified;
+ private String contentHash;
+
+ public FileFingerprint() {
+ }
+
+ public FileFingerprint(long size, long lastModified, String contentHash) {
+ this.size = size;
+ this.lastModified = lastModified;
+ this.contentHash = contentHash;
+ }
+
+ public long getSize() {
+ return size;
+ }
+
+ public void setSize(long size) {
+ this.size = size;
+ }
+
+ public long getLastModified() {
+ return lastModified;
+ }
+
+ public void setLastModified(long lastModified) {
+ this.lastModified = lastModified;
+ }
+
+ public String getContentHash() {
+ return contentHash;
+ }
+
+ public void setContentHash(String contentHash) {
+ this.contentHash = contentHash;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof FileFingerprint other)) {
+ return false;
+ }
+ return size == other.size
+ && lastModified == other.lastModified
+ && Objects.equals(contentHash, other.contentHash);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = Long.hashCode(size);
+ result = 31 * result + Long.hashCode(lastModified);
+ result = 31 * result + Objects.hashCode(contentHash);
+ return result;
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileWatcherCheckpoint.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileWatcherCheckpoint.java
new file mode 100644
index 0000000000..3b0cb0d7fb
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileWatcherCheckpoint.java
@@ -0,0 +1,77 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.model;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+public class FileWatcherCheckpoint implements Serializable {
+
+ private String currentFileName;
+ private long currentSequence;
+ private long lastProcessedRecord;
+ private FileFingerprint currentFingerprint;
+ private Map<String, FileFingerprint> processedGenerations;
+
+ public FileWatcherCheckpoint() {
+ this.lastProcessedRecord = -1L;
+ this.processedGenerations = new HashMap<>();
+ }
+
+ public String getCurrentFileName() {
+ return currentFileName;
+ }
+
+ public void setCurrentFileName(String currentFileName) {
+ this.currentFileName = currentFileName;
+ }
+
+ public long getCurrentSequence() {
+ return currentSequence;
+ }
+
+ public void setCurrentSequence(long currentSequence) {
+ this.currentSequence = currentSequence;
+ }
+
+ public long getLastProcessedRecord() {
+ return lastProcessedRecord;
+ }
+
+ public void setLastProcessedRecord(long lastProcessedRecord) {
+ this.lastProcessedRecord = lastProcessedRecord;
+ }
+
+ public FileFingerprint getCurrentFingerprint() {
+ return currentFingerprint;
+ }
+
+ public void setCurrentFingerprint(FileFingerprint currentFingerprint) {
+ this.currentFingerprint = currentFingerprint;
+ }
+
+ public Map<String, FileFingerprint> getProcessedGenerations() {
+ return processedGenerations;
+ }
+
+ public void setProcessedGenerations(Map<String, FileFingerprint>
processedGenerations) {
+ this.processedGenerations = processedGenerations;
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileWatcherConfig.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileWatcherConfig.java
new file mode 100644
index 0000000000..9a2d534ebc
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/model/FileWatcherConfig.java
@@ -0,0 +1,30 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.model;
+
+import java.nio.file.Path;
+import java.util.regex.Pattern;
+
+public record FileWatcherConfig(Path directory,
+ Pattern filePattern,
+ CsvParserSettings parserSettings,
+ int pollIntervalSeconds,
+ boolean singleFileGrowthMode,
+ int interEventDelayMs) {
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/CsvFileReader.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/CsvFileReader.java
new file mode 100644
index 0000000000..de59b5ad45
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/CsvFileReader.java
@@ -0,0 +1,145 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+import org.apache.streampipes.commons.exceptions.connect.ParseException;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.CsvParserSettings;
+
+import com.opencsv.CSVParserBuilder;
+import com.opencsv.CSVReader;
+import com.opencsv.CSVReaderBuilder;
+import com.opencsv.exceptions.CsvValidationException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.charset.MalformedInputException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
+import java.util.stream.IntStream;
+
+public class CsvFileReader {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CsvFileReader.class);
+
+ private static final List<Charset> FALLBACK_CHARSETS = List.of(
+ StandardCharsets.UTF_8,
+ Charset.forName("windows-1252"),
+ StandardCharsets.ISO_8859_1
+ );
+
+ public FileWatchReadResult readFrom(Path path,
+ CsvParserSettings parserSettings,
+ long startRecordIndex,
+ BiConsumer<Long, Map<String, Object>>
eventConsumer) throws IOException {
+ IOException lastException = null;
+ for (Charset charset : FALLBACK_CHARSETS) {
+ try {
+ LOG.debug("Trying to read CSV file '{}' with charset '{}'.",
path.getFileName(), charset.name());
+ return readFrom(path, parserSettings, startRecordIndex, eventConsumer,
charset);
+ } catch (MalformedInputException e) {
+ LOG.debug("Could not decode CSV file '{}' with charset '{}': {}.",
+ path.getFileName(), charset.name(), e.getMessage());
+ lastException = e;
+ }
+ }
+
+ throw new IOException("Could not decode CSV file " + path.getFileName() +
" using supported charsets.", lastException);
+ }
+
+ private FileWatchReadResult readFrom(Path path,
+ CsvParserSettings parserSettings,
+ long startRecordIndex,
+ BiConsumer<Long, Map<String, Object>>
eventConsumer,
+ Charset charset) throws IOException {
+ try (var bufferedReader = Files.newBufferedReader(path, charset);
+ var csvReader = newCsvReader(bufferedReader, parserSettings)) {
+ LOG.debug("Reading CSV file '{}' using charset '{}'.",
path.getFileName(), charset.name());
+
+ String[] header = parserSettings.hasHeader() ? csvReader.readNext() :
null;
+ String[] row;
+ long currentRecordIndex = -1L;
+ long lastProcessedRecord = startRecordIndex - 1;
+
+ while ((row = csvReader.readNext()) != null) {
+ currentRecordIndex++;
+
+ if (header == null) {
+ header = IntStream.range(0, row.length)
+ .mapToObj(i -> "key_" + i)
+ .toArray(String[]::new);
+ }
+
+ if (currentRecordIndex < startRecordIndex) {
+ continue;
+ }
+
+ eventConsumer.accept(currentRecordIndex, toMap(header, row));
+ lastProcessedRecord = currentRecordIndex;
+ }
+
+ int emittedEvents = lastProcessedRecord < startRecordIndex
+ ? 0
+ : (int) (lastProcessedRecord - startRecordIndex + 1);
+
+ LOG.debug("Completed reading CSV file '{}'. Emitted {} records starting
from record {}.",
+ path.getFileName(), emittedEvents, startRecordIndex);
+
+ return new FileWatchReadResult(lastProcessedRecord, emittedEvents, true);
+ } catch (CsvValidationException e) {
+ throw new ParseException("Could not parse CSV file " +
path.getFileName(), e);
+ }
+ }
+
+ private CSVReader newCsvReader(BufferedReader reader, CsvParserSettings
parserSettings) {
+ var csvParser = new CSVParserBuilder()
+ .withSeparator(parserSettings.delimiter())
+ .build();
+
+ return new CSVReaderBuilder(reader)
+ .withSkipLines(0)
+ .withCSVParser(csvParser)
+ .build();
+ }
+
+ private Map<String, Object> toMap(String[] header, String[] row) {
+ Map<String, Object> event = new LinkedHashMap<>();
+ for (int i = 0; i < header.length; i++) {
+ event.put(header[i], i < row.length ? row[i] : "");
+ }
+
+ if (row.length > header.length && header.length > 0) {
+ String lastHeader = header[header.length - 1];
+ String mergedValue = String.valueOf(event.get(lastHeader));
+ for (int i = header.length; i < row.length; i++) {
+ mergedValue = mergedValue + row[i];
+ }
+ event.put(lastHeader, mergedValue);
+ }
+
+ return event;
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/EventDelayExecutor.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/EventDelayExecutor.java
new file mode 100644
index 0000000000..515888105f
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/EventDelayExecutor.java
@@ -0,0 +1,25 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+@FunctionalInterface
+public interface EventDelayExecutor {
+
+ void sleep(long delayMs) throws InterruptedException;
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/EventMapper.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/EventMapper.java
new file mode 100644
index 0000000000..294e565b6f
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/EventMapper.java
@@ -0,0 +1,31 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+import java.util.Map;
+
+@FunctionalInterface
+public interface EventMapper {
+
+ Map<String, Object> map(Map<String, Object> rawEvent);
+
+ static EventMapper identity() {
+ return rawEvent -> rawEvent;
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSetWatcher.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSetWatcher.java
new file mode 100644
index 0000000000..549285de10
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSetWatcher.java
@@ -0,0 +1,317 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+import org.apache.streampipes.extensions.api.connect.IEventCollector;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.FileFingerprint;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.FileWatcherCheckpoint;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.FileWatcherConfig;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.regex.Matcher;
+
+public class FileSetWatcher {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(FileSetWatcher.class);
+
+ private final FileWatcherConfig config;
+ private final FileWatcherCheckpointStore checkpointStore;
+ private final CsvFileReader csvFileReader;
+ private final EventMapper eventMapper;
+ private final EventDelayExecutor eventDelayExecutor;
+
+ public FileSetWatcher(FileWatcherConfig config,
+ FileWatcherCheckpointStore checkpointStore,
+ CsvFileReader csvFileReader) {
+ this(config, checkpointStore, csvFileReader, EventMapper.identity());
+ }
+
+ public FileSetWatcher(FileWatcherConfig config,
+ FileWatcherCheckpointStore checkpointStore,
+ CsvFileReader csvFileReader,
+ EventMapper eventMapper) {
+ this(config, checkpointStore, csvFileReader, eventMapper, Thread::sleep);
+ }
+
+ public FileSetWatcher(FileWatcherConfig config,
+ FileWatcherCheckpointStore checkpointStore,
+ CsvFileReader csvFileReader,
+ EventMapper eventMapper,
+ EventDelayExecutor eventDelayExecutor) {
+ this.config = config;
+ this.checkpointStore = checkpointStore;
+ this.csvFileReader = csvFileReader;
+ this.eventMapper = eventMapper;
+ this.eventDelayExecutor = eventDelayExecutor;
+ }
+
+ public void poll(String adapterElementId, IEventCollector collector) throws
IOException {
+ LOG.debug("Polling WinCC archive files in directory '{}' for adapter
'{}'.",
+ config.directory(), adapterElementId);
+ var files = discoverFiles();
+ if (files.isEmpty()) {
+ LOG.debug("No matching WinCC archive files found in '{}'.",
config.directory());
+ return;
+ }
+
+ var checkpoint = checkpointStore.load(adapterElementId);
+ LOG.debug(
+ "Loaded checkpoint for adapter '{}': currentFile='{}', sequence={},
lastProcessedRecord={}, knownGenerations={}.",
+ adapterElementId,
+ checkpoint.getCurrentFileName(),
+ checkpoint.getCurrentSequence(),
+ checkpoint.getLastProcessedRecord(),
+ checkpoint.getProcessedGenerations().size()
+ );
+ processCurrentFile(adapterElementId, collector, checkpoint, files);
+ processFollowingFiles(adapterElementId, collector, checkpoint, files);
+ }
+
+ private void processCurrentFile(String adapterElementId,
+ IEventCollector collector,
+ FileWatcherCheckpoint checkpoint,
+ List<FileSlot> files) throws IOException {
+ if (checkpoint.getCurrentFileName() == null ||
checkpoint.getCurrentFingerprint() == null) {
+ LOG.debug("No existing checkpoint state found. Starting from the first
available matching file.");
+ return;
+ }
+
+ FileSlot currentSlot = files.stream()
+ .filter(file ->
file.fileName().equals(checkpoint.getCurrentFileName()))
+ .findFirst()
+ .orElse(null);
+
+ if (currentSlot == null) {
+ LOG.debug("Checkpoint file '{}' is no longer present in '{}'. Continuing
with following files.",
+ checkpoint.getCurrentFileName(), config.directory());
+ return;
+ }
+
+ long startRecord = determineStartRecord(currentSlot, checkpoint);
+ LOG.debug("Resuming file '{}' at record {}.", currentSlot.fileName(),
startRecord);
+
+ readFile(adapterElementId, collector, checkpoint, currentSlot,
startRecord);
+ }
+
+ private long determineStartRecord(FileSlot currentSlot,
FileWatcherCheckpoint checkpoint) {
+ if (currentSlot.fingerprint().equals(checkpoint.getCurrentFingerprint())) {
+ LOG.debug("File '{}' fingerprint unchanged. Continuing after record {}.",
+ currentSlot.fileName(), checkpoint.getLastProcessedRecord());
+ return checkpoint.getLastProcessedRecord() + 1;
+ }
+
+ if (config.singleFileGrowthMode() &&
isAppendedInPlace(currentSlot.fingerprint(),
checkpoint.getCurrentFingerprint())) {
+ LOG.debug("File '{}' grew in place. Continuing after record {}.",
+ currentSlot.fileName(), checkpoint.getLastProcessedRecord());
+ return checkpoint.getLastProcessedRecord() + 1;
+ }
+
+ LOG.debug("File '{}' fingerprint changed. Restarting from record 0.",
currentSlot.fileName());
+ return 0;
+ }
+
+ private boolean isAppendedInPlace(FileFingerprint currentFingerprint,
FileFingerprint previousFingerprint) {
+ return currentFingerprint.getSize() > previousFingerprint.getSize()
+ && currentFingerprint.getLastModified() >=
previousFingerprint.getLastModified();
+ }
+
+ private void processFollowingFiles(String adapterElementId,
+ IEventCollector collector,
+ FileWatcherCheckpoint checkpoint,
+ List<FileSlot> files) throws IOException {
+ int startIndex = startIndex(files, checkpoint);
+ LOG.debug("Processing following files starting at index {}.", startIndex);
+ for (int offset = 0; offset < files.size(); offset++) {
+ FileSlot candidate = files.get((startIndex + offset) % files.size());
+ FileFingerprint processedFingerprint =
checkpoint.getProcessedGenerations().get(candidate.fileName());
+ if (processedFingerprint != null &&
processedFingerprint.equals(candidate.fingerprint())) {
+ LOG.debug("Stopping at file '{}' because this generation was already
processed.", candidate.fileName());
+ break;
+ }
+
+ LOG.debug("Reading new file generation '{}' from record 0.",
candidate.fileName());
+ readFile(adapterElementId, collector, checkpoint, candidate, 0);
+ }
+ }
+
+ private void readFile(String adapterElementId,
+ IEventCollector collector,
+ FileWatcherCheckpoint checkpoint,
+ FileSlot fileSlot,
+ long startRecord) throws IOException {
+ LOG.debug(
+ "Reading file '{}' (sequence={}, size={}, lastModified={}) from record
{} with interEventDelayMs={}.",
+ fileSlot.fileName(),
+ fileSlot.sequence(),
+ fileSlot.fingerprint().getSize(),
+ fileSlot.fingerprint().getLastModified(),
+ startRecord,
+ config.interEventDelayMs()
+ );
+ final boolean[] firstEvent = {true};
+ var result = csvFileReader.readFrom(fileSlot.path(),
config.parserSettings(), startRecord, (recordIndex, event) -> {
+ delayBeforeNextEvent(firstEvent[0], fileSlot.fileName(), recordIndex);
+ firstEvent[0] = false;
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Collected record {} from file '{}'.", recordIndex,
fileSlot.fileName());
+ }
+ collector.collect(eventMapper.map(event));
+ checkpoint.setCurrentFileName(fileSlot.fileName());
+ checkpoint.setCurrentSequence(fileSlot.sequence());
+ checkpoint.setCurrentFingerprint(fileSlot.fingerprint());
+ checkpoint.setLastProcessedRecord(recordIndex);
+ checkpoint.getProcessedGenerations().put(fileSlot.fileName(),
fileSlot.fingerprint());
+
+ try {
+ checkpointStore.save(adapterElementId, checkpoint);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Checkpoint saved for file '{}' at record {}.",
fileSlot.fileName(), recordIndex);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ checkpoint.setCurrentFileName(fileSlot.fileName());
+ checkpoint.setCurrentSequence(fileSlot.sequence());
+ checkpoint.setCurrentFingerprint(fileSlot.fingerprint());
+ checkpoint.setLastProcessedRecord(result.lastProcessedRecord());
+ checkpoint.getProcessedGenerations().put(fileSlot.fileName(),
fileSlot.fingerprint());
+ checkpointStore.save(adapterElementId, checkpoint);
+ LOG.debug(
+ "Finished file '{}'. Emitted {} events, lastProcessedRecord={}.",
+ fileSlot.fileName(),
+ result.emittedEvents(),
+ result.lastProcessedRecord()
+ );
+ }
+
+ private void delayBeforeNextEvent(boolean firstEvent,
+ String fileName,
+ long recordIndex) {
+ if (firstEvent || config.interEventDelayMs() <= 0) {
+ return;
+ }
+
+ try {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Sleeping {} ms before record {} from file '{}'.",
+ config.interEventDelayMs(), recordIndex, fileName);
+ }
+ eventDelayExecutor.sleep(config.interEventDelayMs());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while delaying replay for file "
+ fileName, e);
+ }
+ }
+
+ List<FileSlot> discoverFiles() throws IOException {
+ List<FileSlot> files = new ArrayList<>();
+ try (var stream = Files.list(config.directory())) {
+ stream.filter(Files::isRegularFile)
+ .filter(path ->
config.filePattern().matcher(path.getFileName().toString()).matches())
+ .forEach(path -> files.add(toFileSlot(path)));
+ }
+
+
files.sort(Comparator.comparingLong(FileSlot::sequence).thenComparing(FileSlot::fileName));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Discovered {} matching WinCC archive files: {}",
+ files.size(),
+ files.stream().map(file -> file.fileName() + "@" +
file.sequence()).toList());
+ }
+ return files;
+ }
+
+ private int startIndex(List<FileSlot> files, FileWatcherCheckpoint
checkpoint) {
+ if (checkpoint.getCurrentFileName() == null) {
+ return 0;
+ }
+
+ for (int i = 0; i < files.size(); i++) {
+ FileSlot slot = files.get(i);
+ if (slot.sequence() > checkpoint.getCurrentSequence()) {
+ return i;
+ }
+ if (slot.sequence() == checkpoint.getCurrentSequence()
+ && slot.fileName().compareTo(checkpoint.getCurrentFileName()) > 0) {
+ return i;
+ }
+ }
+
+ return 0;
+ }
+
+ private FileSlot toFileSlot(Path path) {
+ String fileName = path.getFileName().toString();
+ return new FileSlot(fileName, path, sequence(fileName), fingerprint(path));
+ }
+
+ private long sequence(String fileName) {
+ Matcher matcher = config.filePattern().matcher(fileName);
+ if (matcher.matches() && matcher.groupCount() >= 1) {
+ try {
+ return Long.parseLong(matcher.group(1));
+ } catch (NumberFormatException ignored) {
+ return Long.MAX_VALUE;
+ }
+ }
+
+ return Long.MAX_VALUE;
+ }
+
+ private FileFingerprint fingerprint(Path path) {
+ try {
+ long size = Files.size(path);
+ long lastModified = Files.getLastModifiedTime(path).toMillis();
+ String hash = sha256(path);
+ return new FileFingerprint(size, lastModified, hash);
+ } catch (IOException e) {
+ throw new RuntimeException("Could not create fingerprint for " +
path.getFileName(), e);
+ }
+ }
+
+ private String sha256(Path path) throws IOException {
+ try {
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ try (InputStream in = Files.newInputStream(path)) {
+ byte[] buffer = new byte[8192];
+ int read;
+ while ((read = in.read(buffer)) != -1) {
+ digest.update(buffer, 0, read);
+ }
+ }
+
+ return HexFormat.of().formatHex(digest.digest());
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 is not available", e);
+ }
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSlot.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSlot.java
new file mode 100644
index 0000000000..9c75ed052d
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSlot.java
@@ -0,0 +1,26 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.FileFingerprint;
+
+import java.nio.file.Path;
+
+public record FileSlot(String fileName, Path path, long sequence,
FileFingerprint fingerprint) {
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileWatchReadResult.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileWatchReadResult.java
new file mode 100644
index 0000000000..ff64ff4f12
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileWatchReadResult.java
@@ -0,0 +1,24 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+public record FileWatchReadResult(long lastProcessedRecord,
+ int emittedEvents,
+ boolean reachedEndOfFile) {
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileWatcherCheckpointStore.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileWatcherCheckpointStore.java
new file mode 100644
index 0000000000..15150c69bb
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileWatcherCheckpointStore.java
@@ -0,0 +1,70 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+import org.apache.streampipes.commons.environment.Environments;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.FileWatcherCheckpoint;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+public class FileWatcherCheckpointStore {
+
+ private static final ObjectMapper MAPPER =
JacksonSerializer.getObjectMapper();
+ private final Path baseDirectory;
+
+ public FileWatcherCheckpointStore() {
+ this(resolveDefaultDirectory());
+ }
+
+ public FileWatcherCheckpointStore(Path baseDirectory) {
+ this.baseDirectory = baseDirectory;
+ }
+
+ public FileWatcherCheckpoint load(String adapterElementId) throws
IOException {
+ Path checkpointPath = checkpointPath(adapterElementId);
+ if (!Files.exists(checkpointPath)) {
+ return new FileWatcherCheckpoint();
+ }
+
+ return MAPPER.readValue(checkpointPath.toFile(),
FileWatcherCheckpoint.class);
+ }
+
+ public void save(String adapterElementId, FileWatcherCheckpoint checkpoint)
throws IOException {
+ Files.createDirectories(baseDirectory);
+ MAPPER.writeValue(checkpointPath(adapterElementId).toFile(), checkpoint);
+ }
+
+ private Path checkpointPath(String adapterElementId) {
+ return baseDirectory.resolve(adapterElementId + ".json");
+ }
+
+ private static Path resolveDefaultDirectory() {
+ String baseDir = Environments
+ .getEnvironment()
+ .getExtAssetBaseDir()
+ .getValueOrReturn(System.getProperty("user.home"));
+
+ return Path.of(baseDir, ".streampipes", "service",
"filewatcher-checkpoints");
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/resources/org.apache.streampipes.connectors.wincc.alarm.archive/documentation.md
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/resources/org.apache.streampipes.connectors.wincc.alarm.archive/documentation.md
new file mode 100644
index 0000000000..80c5e7556d
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/resources/org.apache.streampipes.connectors.wincc.alarm.archive/documentation.md
@@ -0,0 +1,45 @@
+<!--
+ ~ 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.
+ ~
+ -->
+
+The WinCC Alarm Archive adapter reads Siemens WinCC alarm archives exported as
CSV files.
+
+It is intended for the WinCC alarm CSV structure and for both archive modes:
+
+- single archive file
+- segmented circular log with multiple rotating CSV segments
+
+Configuration is Siemens-specific:
+
+- archive directory
+- archive base name
+- segmented circular log enabled or disabled
+- segment count in segmented circular log mode
+- polling interval
+- optional delay between replayed events in milliseconds
+
+The adapter assumes the WinCC alarm CSV layout with a header row and
semicolon-separated columns.
+The configured base name should not include the segment number or file suffix.
For example, use `Meldungsarchiv`
+to match `Meldungsarchiv1.csv` or `Meldungsarchiv1.csv` to
`MeldungsarchivN.csv`.
+
+In segmented circular log mode the adapter treats reused file names as new
generations by fingerprinting file content,
+so a segment like `Meldungsarchiv1.csv` can be consumed again after WinCC
rotates the archive.
+
+Use the inter-event delay when consumers or brokers cannot handle large replay
bursts. A value like `1` ms can reduce
+event loss on slower consumers, while `0` disables throttling.
+
+The configured directory must be reachable from the extension runtime process
or container.
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/main/resources/org.apache.streampipes.connectors.wincc.alarm.archive/strings.en
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/resources/org.apache.streampipes.connectors.wincc.alarm.archive/strings.en
new file mode 100644
index 0000000000..0012cb3226
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/main/resources/org.apache.streampipes.connectors.wincc.alarm.archive/strings.en
@@ -0,0 +1,28 @@
+# 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.
+
+org.apache.streampipes.connectors.wincc.alarm.archive.title=WinCC Alarm Archive
+org.apache.streampipes.connectors.wincc.alarm.archive.description=Reads
Siemens WinCC alarm archive CSV files with fixed WinCC CSV semantics and
segmented circular log rotation
+directory-path.title=Archive directory
+directory-path.description=Absolute path to the WinCC alarm archive directory
+archive-base-name.title=Archive base name
+archive-base-name.description=Base name of the WinCC alarm archive files
without segment number or .csv suffix, for example Meldungsarchiv
+segmented-circular-log-enabled.title=Segmented circular log
+segmented-circular-log-enabled.description=Whether the WinCC alarm archive
uses rotating file segments
+archive-segment-count.title=Segment count
+archive-segment-count.description=Number of archive segments in segmented
circular log mode
+poll-interval-seconds.title=Polling interval (seconds)
+poll-interval-seconds.description=Interval used to scan the alarm archive
files for new entries
+inter-event-delay-ms.title=Inter-event delay (ms)
+inter-event-delay-ms.description=Delay between replayed alarm events in
milliseconds. Use 0 to disable throttling
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmEventMapperTest.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmEventMapperTest.java
new file mode 100644
index 0000000000..911f3ddb7e
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCAlarmEventMapperTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.adapter;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class WinCCAlarmEventMapperTest {
+
+ @Test
+ void shouldMapRawWinCCAlarmToNormalizedEvent() {
+ var rawEvent = Map.<String, Object>ofEntries(
+ Map.entry("Time_ms", "46120366239"),
+ Map.entry("MsgProc", "1"),
+ Map.entry("StateAfter", "3"),
+ Map.entry("MsgClass", "2"),
+ Map.entry("MsgNumber", "110001"),
+ Map.entry("Var1", "A"),
+ Map.entry("Var2", "B"),
+ Map.entry("Var3", ""),
+ Map.entry("Var4", ""),
+ Map.entry("Var5", ""),
+ Map.entry("Var6", ""),
+ Map.entry("Var7", ""),
+ Map.entry("Var8", ""),
+ Map.entry("TimeString", "08.04.26 08:47"),
+ Map.entry("MsgText", "Alarm text"),
+ Map.entry("PLC", "PLC_1")
+ );
+
+ var mappedEvent = new WinCCAlarmEventMapper().map(rawEvent);
+
+ assertEquals(1775638043050L, mappedEvent.get("timestamp"));
+ assertEquals("46120366239", mappedEvent.get("timestamp_ms"));
+ assertEquals("08.04.26 08:47", mappedEvent.get("timestamp_string"));
+ assertEquals("Alarm text", mappedEvent.get("message_text"));
+ assertEquals(110001, mappedEvent.get("message_number"));
+ assertEquals(1, mappedEvent.get("alarm_procedure_code"));
+ assertEquals("system_event", mappedEvent.get("alarm_procedure"));
+ assertEquals(3, mappedEvent.get("state_after_code"));
+ assertEquals("incoming_acknowledged", mappedEvent.get("state_after"));
+ assertEquals(2, mappedEvent.get("alarm_class_code"));
+ assertEquals("warnings", mappedEvent.get("alarm_class"));
+ assertEquals("PLC_1", mappedEvent.get("plc"));
+ assertEquals("A", mappedEvent.get("parameter_1"));
+ assertEquals("B", mappedEvent.get("parameter_2"));
+ assertFalse(mappedEvent.containsKey("raw_Var1"));
+ assertNull(mappedEvent.get("raw_MsgText"));
+ assertNull(mappedEvent.get("raw_StateAfter"));
+ assertEquals("46120366239", mappedEvent.get("raw_Time_ms"));
+ assertEquals("08.04.26 08:47", mappedEvent.get("raw_TimeString"));
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCTimestampConverterTest.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCTimestampConverterTest.java
new file mode 100644
index 0000000000..5f1c05add2
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/adapter/WinCCTimestampConverterTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.adapter;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class WinCCTimestampConverterTest {
+
+ @Test
+ void shouldConvertWinCCTimestampToUnixMillis() {
+ assertEquals(1072876370999L,
+ WinCCTimestampConverter.toUnixTimestampMillis("37986.55059027",
"30.12.03 13:12:51"));
+ assertEquals(1072876370999L,
+ WinCCTimestampConverter.toUnixTimestampMillis("37986,55059027",
"30.12.03 13:12:51"));
+ assertEquals(1775638043050L,
+ WinCCTimestampConverter.toUnixTimestampMillis(46120366239L, "08.04.26
08:47"));
+ }
+
+ @Test
+ void shouldReturnNullForInvalidInput() {
+ assertNull(WinCCTimestampConverter.toUnixTimestampMillis(null, null));
+ assertNull(WinCCTimestampConverter.toUnixTimestampMillis("", ""));
+
assertNull(WinCCTimestampConverter.toUnixTimestampMillis("not-a-timestamp",
"also-invalid"));
+ }
+
+ @Test
+ void shouldFallbackToTimeStringWhenNeeded() {
+ assertEquals(1775630820000L,
+ WinCCTimestampConverter.toUnixTimestampMillis(null, "08.04.26 08:47"));
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/CsvFileReaderTest.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/CsvFileReaderTest.java
new file mode 100644
index 0000000000..ecef4cf6b4
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/CsvFileReaderTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.CsvParserSettings;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class CsvFileReaderTest {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void shouldKeepQuotedSeparatorsInsideOneField() throws IOException {
+ Path file = tempDir.resolve("Meldungsarchiv1.csv");
+ Files.writeString(file, "Time_ms;MsgText;PLC\n46120366239;\"Alarm;
text\";PLC_1\n");
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ new CsvFileReader().readFrom(file, new CsvParserSettings(true, ';'), 0,
(recordIndex, event) -> events.add(event));
+
+ assertEquals(1, events.size());
+ assertEquals("Alarm; text", events.get(0).get("MsgText"));
+ }
+
+ @Test
+ void shouldPadMissingColumnsWithEmptyStrings() throws IOException {
+ Path file = tempDir.resolve("Meldungsarchiv1.csv");
+ Files.writeString(file, "Time_ms;MsgText;PLC\n46120366239;Alarm text\n");
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ new CsvFileReader().readFrom(file, new CsvParserSettings(true, ';'), 0,
(recordIndex, event) -> events.add(event));
+
+ assertEquals(1, events.size());
+ assertEquals("", events.get(0).get("PLC"));
+ }
+
+ @Test
+ void shouldFallbackToWindows1252ForWinCCExports() throws IOException {
+ Path file = tempDir.resolve("Meldungsarchiv1.csv");
+ Files.write(
+ file,
+ "Time_ms;MsgText;PLC\n46120366239;Störung
Überfüllung;PLC_1\n".getBytes(Charset.forName("windows-1252"))
+ );
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ new CsvFileReader().readFrom(file, new CsvParserSettings(true, ';'), 0,
(recordIndex, event) -> events.add(event));
+
+ assertEquals(1, events.size());
+ assertEquals("Störung Überfüllung", events.get(0).get("MsgText"));
+ }
+}
diff --git
a/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSetWatcherTest.java
b/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSetWatcherTest.java
new file mode 100644
index 0000000000..5cc92c543f
--- /dev/null
+++
b/streampipes-extensions/streampipes-connectors-filewatcher/src/test/java/org/apache/streampipes/extensions/connectors/filewatcher/runtime/FileSetWatcherTest.java
@@ -0,0 +1,180 @@
+/*
+ * 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.streampipes.extensions.connectors.filewatcher.runtime;
+
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.CsvParserSettings;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.FileWatcherCheckpoint;
+import
org.apache.streampipes.extensions.connectors.filewatcher.model.FileWatcherConfig;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Pattern;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class FileSetWatcherTest {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void shouldReadInitialFilesAndOnlyNewGenerationAfterWrapAround() throws
IOException {
+ write("Meldungsarchiv1", "id,value\n1,a\n2,b\n");
+ write("Meldungsarchiv2", "id,value\n3,c\n");
+
+ FileSetWatcher watcher = new FileSetWatcher(
+ new FileWatcherConfig(tempDir,
Pattern.compile("Meldungsarchiv(\\d+)"), new CsvParserSettings(true, ','), 1,
false, 0),
+ new FileWatcherCheckpointStore(tempDir.resolve("checkpoints")),
+ new CsvFileReader(),
+ EventMapper.identity()
+ );
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ watcher.poll("adapter-1", events::add);
+
+ assertEquals(3, events.size());
+ assertEquals("1", events.get(0).get("id"));
+ assertEquals("3", events.get(2).get("id"));
+
+ events.clear();
+ watcher.poll("adapter-1", events::add);
+ assertEquals(0, events.size());
+
+ write("Meldungsarchiv1", "id,value\n4,d\n5,e\n");
+ watcher.poll("adapter-1", events::add);
+
+ assertEquals(2, events.size());
+ assertEquals("4", events.get(0).get("id"));
+ assertEquals("5", events.get(1).get("id"));
+ }
+
+ @Test
+ void shouldResumeWithinSameGenerationFromCheckpoint() throws IOException {
+ write("Meldungsarchiv1", "id,value\n1,a\n2,b\n3,c\n");
+ Path checkpointDir = tempDir.resolve("checkpoints");
+ FileWatcherCheckpointStore checkpointStore = new
FileWatcherCheckpointStore(checkpointDir);
+
+ FileSetWatcher watcher = new FileSetWatcher(
+ new FileWatcherConfig(tempDir,
Pattern.compile("Meldungsarchiv(\\d+)"), new CsvParserSettings(true, ','), 1,
false, 0),
+ checkpointStore,
+ new CsvFileReader(),
+ EventMapper.identity()
+ );
+
+ FileSlot slot = watcher.discoverFiles().get(0);
+ FileWatcherCheckpoint checkpoint = new FileWatcherCheckpoint();
+ checkpoint.setCurrentFileName(slot.fileName());
+ checkpoint.setCurrentSequence(slot.sequence());
+ checkpoint.setCurrentFingerprint(slot.fingerprint());
+ checkpoint.setLastProcessedRecord(1L);
+ checkpoint.getProcessedGenerations().put(slot.fileName(),
slot.fingerprint());
+ checkpointStore.save("adapter-2", checkpoint);
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ watcher.poll("adapter-2", events::add);
+
+ assertEquals(1, events.size());
+ assertEquals("3", events.get(0).get("id"));
+ }
+
+ @Test
+ void shouldContinueFromLastLineWhenSingleFileGrows() throws IOException {
+ write("Meldungsarchiv.csv", "id,value\n1,a\n2,b\n");
+
+ FileSetWatcher watcher = new FileSetWatcher(
+ new FileWatcherConfig(tempDir,
Pattern.compile("Meldungsarchiv\\.csv"), new CsvParserSettings(true, ','), 1,
true, 0),
+ new FileWatcherCheckpointStore(tempDir.resolve("checkpoints-single")),
+ new CsvFileReader(),
+ EventMapper.identity()
+ );
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ watcher.poll("adapter-single", events::add);
+
+ assertEquals(2, events.size());
+ events.clear();
+
+ write("Meldungsarchiv.csv", "id,value\n1,a\n2,b\n3,c\n4,d\n");
+ watcher.poll("adapter-single", events::add);
+
+ assertEquals(2, events.size());
+ assertEquals("3", events.get(0).get("id"));
+ assertEquals("4", events.get(1).get("id"));
+ }
+
+ @Test
+ void shouldRestartFromBeginningWhenSingleFileIsTruncatedOrReplaced() throws
IOException {
+ write("Meldungsarchiv.csv", "id,value\n1,a\n2,b\n3,c\n");
+
+ FileSetWatcher watcher = new FileSetWatcher(
+ new FileWatcherConfig(tempDir,
Pattern.compile("Meldungsarchiv\\.csv"), new CsvParserSettings(true, ','), 1,
true, 0),
+ new FileWatcherCheckpointStore(tempDir.resolve("checkpoints-replace")),
+ new CsvFileReader(),
+ EventMapper.identity()
+ );
+
+ watcher.poll("adapter-replace", event -> {
+ });
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ write("Meldungsarchiv.csv", "id,value\n10,x\n11,y\n");
+ watcher.poll("adapter-replace", events::add);
+
+ assertEquals(2, events.size());
+ assertEquals("10", events.get(0).get("id"));
+ assertEquals("11", events.get(1).get("id"));
+ }
+
+ @Test
+ void shouldDelayBetweenEventsWhenConfigured() throws IOException {
+ write("Meldungsarchiv1", "id,value\n1,a\n2,b\n3,c\n");
+ AtomicInteger delayCalls = new AtomicInteger();
+ List<Long> appliedDelays = new ArrayList<>();
+
+ FileSetWatcher watcher = new FileSetWatcher(
+ new FileWatcherConfig(tempDir,
Pattern.compile("Meldungsarchiv(\\d+)"), new CsvParserSettings(true, ','), 1,
false, 5),
+ new FileWatcherCheckpointStore(tempDir.resolve("checkpoints-delay")),
+ new CsvFileReader(),
+ EventMapper.identity(),
+ delayMs -> {
+ delayCalls.incrementAndGet();
+ appliedDelays.add(delayMs);
+ }
+ );
+
+ List<Map<String, Object>> events = new ArrayList<>();
+ watcher.poll("adapter-delay", events::add);
+
+ assertEquals(3, events.size());
+ assertEquals(2, delayCalls.get());
+ assertEquals(List.of(5L, 5L), appliedDelays);
+ }
+
+ private void write(String fileName, String content) throws IOException {
+ Files.writeString(tempDir.resolve(fileName), content);
+ }
+}
diff --git a/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
b/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
index 4507aff55c..47e559cda0 100644
--- a/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
+++ b/streampipes-extensions/streampipes-extensions-all-iiot/pom.xml
@@ -56,6 +56,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-connectors-filewatcher</artifactId>
+ <version>0.99.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-connectors-influx</artifactId>
diff --git
a/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java
b/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java
index fb64c58b37..179f9b4ede 100644
---
a/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java
+++
b/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.extensions.all.iiot;
import org.apache.streampipes.connect.iiot.IIoTAdaptersExtensionModuleExport;
import org.apache.streampipes.connectors.ros.RosConnectorsModuleExport;
+import
org.apache.streampipes.extensions.connectors.filewatcher.FileWatcherConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.kafka.KafkaConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.mqtt.MqttConnectorsModuleExport;
@@ -65,6 +66,7 @@ public class AllExtensionsIIoTInit extends
StreamPipesExtensionsServiceBase {
.registerModules(
new IIoTAdaptersExtensionModuleExport(),
+ new FileWatcherConnectorsModuleExport(),
new InfluxConnectorsModuleExport(),
new KafkaConnectorsModuleExport(),
new MqttConnectorsModuleExport(),
diff --git a/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
b/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
index f6da620463..a165ea8b03 100644
--- a/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-extensions-all-jvm/pom.xml
@@ -74,6 +74,11 @@
<artifactId>streampipes-connectors-camel-azure</artifactId>
<version>0.99.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-connectors-filewatcher</artifactId>
+ <version>0.99.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-connectors-influx</artifactId>
diff --git
a/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
b/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
index e39dbdf0d8..ef031f35f3 100644
---
a/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
+++
b/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
@@ -21,6 +21,7 @@ import
org.apache.streampipes.connect.GeneralAdaptersExtensionModuleExport;
import org.apache.streampipes.connect.iiot.IIoTAdaptersExtensionModuleExport;
import org.apache.streampipes.connectors.ros.RosConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.camel.azure.CamelAzureConnectorsModuleExport;
+import
org.apache.streampipes.extensions.connectors.filewatcher.FileWatcherConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.kafka.KafkaConnectorsModuleExport;
import
org.apache.streampipes.extensions.connectors.mqtt.MqttConnectorsModuleExport;
@@ -69,6 +70,7 @@ public class AllExtensionsInit extends
StreamPipesExtensionsServiceBase {
new IIoTAdaptersExtensionModuleExport(),
new CamelAzureConnectorsModuleExport(),
+ new FileWatcherConnectorsModuleExport(),
new InfluxConnectorsModuleExport(),
new KafkaConnectorsModuleExport(),
new MqttConnectorsModuleExport(),
