Repository: apache/streampipes
Pull request #2759 (2024-04-23 07:52:34)

refactor: change `FilterCondition` class to record
<!--
  ~ 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.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no



Patch:
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
index d2f518246b..8c968ca885 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
@@ -170,7 +170,7 @@ public class DataLakeInfluxQueryBuilder implements IDataLakeQueryBuilder<Query>
   public IDataLakeQueryBuilder<Query> withInclusiveFilter(List<FilterCondition> filterConditions) {
     List<ConjunctionClause> and = new ArrayList<>();
     filterConditions.forEach(c -> and
-        .add(new AndConjunction(new SimpleClause(c.getField(), c.getOperator(), c.getCondition()))));
+        .add(new AndConjunction(new SimpleClause(c.field(), c.operator(), c.condition()))));
 
     addNestedWhereClause(and);
 
@@ -251,7 +251,7 @@ public class DataLakeInfluxQueryBuilder implements IDataLakeQueryBuilder<Query>
         this.selectionQuery.from(env.getTsStorageBucket().getValueOrDefault(), escapeIndex(measurementId));
     this.whereClauses.forEach(selectQuery::where);
 
-    if (this.groupByClauses.size() > 0) {
+    if (!this.groupByClauses.isEmpty()) {
       selectQuery.groupBy(this.groupByClauses.toArray());
     }
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java
index 396bed21d2..ed2592d8d3 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java
@@ -17,28 +17,5 @@
  */
 package org.apache.streampipes.dataexplorer.querybuilder;
 
-public class FilterCondition {
-
-  private String field;
-  private String operator;
-  private Object condition;
-
-  public FilterCondition(String field, String operator, Object condition) {
-    this.field = field;
-    this.operator = operator;
-    this.condition = condition;
-  }
-
-  public String getField() {
-    return field;
-  }
-
-  public String getOperator() {
-    return operator;
-  }
-
-  public Object getCondition() {
-    return condition;
-  }
-
+public record FilterCondition(String field, String operator, Object condition) {
 }
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
 ) {
 
 }


Test patch:
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
index 6859a41531..984aee2bed 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
@@ -53,9 +53,9 @@ public class WhereStatementParamsTest {
   private void assertWhereCondition(WhereClauseParams result, FilterCondition expected) {
     assertEquals(1, result.getWhereConditions().size());
     FilterCondition resultingFilterCondition = result.getWhereConditions().get(0);
-    assertEquals(expected.getField(), resultingFilterCondition.getField());
-    assertEquals(expected.getOperator(), resultingFilterCondition.getOperator());
-    assertEquals(expected.getCondition(), resultingFilterCondition.getCondition());
+    assertEquals(expected.field(), resultingFilterCondition.field());
+    assertEquals(expected.operator(), resultingFilterCondition.operator());
+    assertEquals(expected.condition(), resultingFilterCondition.condition());
   }