You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by bo...@apache.org on 2024/04/24 06:07:31 UTC
(streampipes) branch rel/0.95.0 updated: fix: OI4 adapter can only read one sensor measurement per event (#2756)
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch rel/0.95.0
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/rel/0.95.0 by this push:
new bfba078584 fix: OI4 adapter can only read one sensor measurement per event (#2756)
bfba078584 is described below
commit bfba0785848602155e099a0502f20d67e3791ba7
Author: Tim <50...@users.noreply.github.com>
AuthorDate: Wed Apr 24 08:06:13 2024 +0200
fix: OI4 adapter can only read one sensor measurement per event (#2756)
* refactor: switch to `ExtensionAssetType` instead of deprecated `Assets`
* fix: adapt message extraction behavior of OI4 adapter
* style: introduce whitespace
* refactor: remove unused `throws` annotation
---
.../connect/iiot/adapters/oi4/Oi4Adapter.java | 46 +++++++++++-----------
.../iiot/adapters/oi4/model/DataSetMessage.java | 3 +-
2 files changed, 25 insertions(+), 24 deletions(-)
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
index cdecf2a56d..6f6a0477ae 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
@@ -37,13 +37,13 @@ import org.apache.streampipes.extensions.management.connect.adapter.parser.json.
import org.apache.streampipes.messaging.InternalEventProcessor;
import org.apache.streampipes.model.AdapterType;
import org.apache.streampipes.model.connect.guess.GuessSchema;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
import org.apache.streampipes.model.schema.EventSchema;
import org.apache.streampipes.sdk.StaticProperties;
import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
import org.apache.streampipes.sdk.helpers.Alternatives;
import org.apache.streampipes.sdk.helpers.Labels;
import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.utils.Assets;
import org.apache.streampipes.sdk.utils.Datatypes;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -60,7 +60,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -87,7 +86,6 @@ public class Oi4Adapter implements StreamPipesAdapter {
private List<String> selectedSensors;
private String givenSensorType;
- private String appId;
protected final ObjectMapper mapper;
@@ -101,7 +99,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
return AdapterConfigurationBuilder
.create(ID, 0, Oi4Adapter::new)
.withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
.withCategory(AdapterType.Generic, AdapterType.Manufacturing)
.requiredTextParameter(MqttConnectUtils.getBrokerUrlLabel())
.requiredAlternatives(MqttConnectUtils.getAccessModeLabel(), MqttConnectUtils.getAlternativesOne(),
@@ -174,7 +172,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
OI4AdapterLabels.LABEL_KEY_SENSORS_ALTERNATIVES
);
- appId = extractor.textParameter(OI4AdapterLabels.LABEL_KEY_APP_ID);
+ String appId = extractor.textParameter(OI4AdapterLabels.LABEL_KEY_APP_ID);
if (selectedAlternativeSelectedSensors.equals(OI4AdapterLabels.LABEL_KEY_SENSORS_ALL_ALTERNATIVE)) {
// An empty list is used to indicated that messages from all sensors should be collected
@@ -197,7 +195,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
public void onAdapterStopped(
IAdapterParameterExtractor extractor,
IAdapterRuntimeContext adapterRuntimeContext
- ) throws AdapterException {
+ ) {
mqttConsumer.close();
}
@@ -311,40 +309,42 @@ public class Oi4Adapter implements StreamPipesAdapter {
private Map<String, Object> extractPayload(NetworkMessage message) throws ParseException {
- var dataMessageOpt = findProcessDataInputMessage(message);
+ var dataMessages = findProcessDataInputMessage(message);
- if (dataMessageOpt.isPresent()) {
+ if (!dataMessages.isEmpty()) {
- var dataMessage = dataMessageOpt.get();
+ for (var dataMessage: dataMessages) {
- var sensorId = getSensorIdFromSource(dataMessage.source());
+ var sensorId = getSensorIdFromSource(dataMessage.source());
- // Verify that the message corresponds to the designated sensor type.
- // This validation relies on the assumption that the source information includes the sensor type.
- if (dataMessage.source()
- .contains(givenSensorType)) {
+ // Verify that the message corresponds to the designated sensor type.
+ // This validation relies on the assumption that the source information includes the sensor type.
+ if (dataMessage.source()
+ .contains(givenSensorType)) {
- // an empty list of selected sensors means that we want to collect data from all sensors available
- if (selectedSensors.isEmpty() || selectedSensors.contains(sensorId)) {
+ // an empty list of selected sensors means that we want to collect data from all sensors available
+ if (selectedSensors.isEmpty() || selectedSensors.contains(sensorId)) {
- return extractAndEnrichMessagePayload(dataMessage, sensorId);
+ return extractAndEnrichMessagePayload(dataMessage, sensorId);
+ }
}
}
}
throw new ParseException("No process data message");
}
- private Optional<DataSetMessage> findProcessDataInputMessage(NetworkMessage message) {
+ private List<DataSetMessage> findProcessDataInputMessage(NetworkMessage message) {
return message.messages()
- .stream()
- .filter(msg -> msg.filter()
- .equals(OI4AdapterLabels.MESSAGE_VALUE_FILTER))
- .findFirst();
+ .stream()
+ .filter(msg -> msg.filter()
+ .equals(OI4AdapterLabels.MESSAGE_VALUE_FILTER))
+ .toList();
}
private Map<String, Object> extractAndEnrichMessagePayload(DataSetMessage dataSetMessage, String sensorId) {
var payload = dataSetMessage
- .payload();
+ .payload()
+ .get(0);
try {
payload.put(
OI4AdapterLabels.EVENT_KEY_TIMESTAMP,
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
index 6fb62b3300..f50dd2f76a 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
@@ -21,6 +21,7 @@ package org.apache.streampipes.connect.iiot.adapters.oi4.model;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
import java.util.Map;
/**
@@ -36,7 +37,7 @@ public record DataSetMessage(
@JsonProperty("Timestamp") String timestamp,
@JsonProperty("Filter") String filter,
@JsonProperty("Source") String source,
- @JsonProperty("Payload") Map<String, Object> payload
+ @JsonProperty("Payload") List<Map<String, Object>> payload
) {
}