This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 709d6e31b refactor(#2246): Refactor InfluxStore.onEvent (#2247)
709d6e31b is described below
commit 709d6e31b12251e07d39576d85d9ffed70c11200
Author: Philipp Zehnder <[email protected]>
AuthorDate: Fri Dec 1 20:15:44 2023 +0100
refactor(#2246): Refactor InfluxStore.onEvent (#2247)
---
streampipes-data-explorer-commons/pom.xml | 12 +-
.../dataexplorer/commons/TimeSeriesStore.java | 3 +-
.../commons/influx/InfluxClientProvider.java | 80 +++++
.../commons/influx/InfluxRequests.java | 46 ---
.../dataexplorer/commons/influx/InfluxStore.java | 321 ++++++++---------
.../commons/influx/PropertyHandler.java | 193 ++++++++++
.../commons/influx/InfluxClientProviderTest.java | 103 ++++++
.../commons/influx/InfluxStoreTest.java | 395 +++++++++++++++++++++
.../connectors/influx/adapter/InfluxDbClient.java | 7 +-
.../connectors/influx/sink/InfluxDbClient.java | 7 +-
...lder.java => EventPropertyListTestBuilder.java} | 31 +-
.../EventPropertyPrimitiveTestBuilder.java | 11 +
12 files changed, 975 insertions(+), 234 deletions(-)
diff --git a/streampipes-data-explorer-commons/pom.xml
b/streampipes-data-explorer-commons/pom.xml
index cfcbea0c2..0377e69c6 100644
--- a/streampipes-data-explorer-commons/pom.xml
+++ b/streampipes-data-explorer-commons/pom.xml
@@ -42,13 +42,23 @@
<artifactId>streampipes-commons</artifactId>
<version>0.95.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-test-utils</artifactId>
+ <version>0.95.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
-
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<!-- Others -->
<dependency>
diff --git
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java
index 5b063fff7..31ce5cbf3 100644
---
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java
+++
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java
@@ -22,6 +22,7 @@ import org.apache.streampipes.client.api.IStreamPipesClient;
import org.apache.streampipes.commons.environment.Environment;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.dataexplorer.commons.image.ImageStore;
+import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
import org.apache.streampipes.dataexplorer.commons.influx.InfluxStore;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
import org.apache.streampipes.model.runtime.Event;
@@ -50,7 +51,7 @@ public class TimeSeriesStore {
this.imageStore = new ImageStore(measure, environment);
}
- this.influxStore = new InfluxStore(measure, environment);
+ this.influxStore = new InfluxStore(measure, environment, new
InfluxClientProvider());
}
diff --git
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java
index b4dd11e2f..9a9e7a127 100644
---
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java
+++
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java
@@ -20,12 +20,21 @@ package org.apache.streampipes.dataexplorer.commons.influx;
import org.apache.streampipes.commons.environment.Environment;
import org.apache.streampipes.commons.environment.Environments;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
+import org.influxdb.dto.Query;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
public class InfluxClientProvider {
+ private static final Logger LOG =
LoggerFactory.getLogger(InfluxClientProvider.class);
+
/**
* Create a new InfluxDB client from environment variables
*
@@ -58,7 +67,78 @@ public class InfluxClientProvider {
}
}
+
+ public InfluxDB getInitializedInfluxDBClient(Environment environment) {
+
+ var settings = InfluxConnectionSettings.from(environment);
+ var influxDb = InfluxClientProvider.getInfluxDBClient(settings);
+ var databaseName = settings.getDatabaseName();
+
+ // Checking, if server is available
+ var response = influxDb.ping();
+ if (response.getVersion()
+ .equalsIgnoreCase("unknown")) {
+ throw new SpRuntimeException("Could not connect to InfluxDb Server: " +
settings.getConnectionUrl());
+ }
+
+ // Checking whether the database exists
+ if (!databaseExists(influxDb, databaseName)) {
+ LOG.info("Database '" + databaseName + "' not found. Gets created ...");
+ createDatabase(influxDb, databaseName);
+ }
+
+ // setting up the database
+ influxDb.setDatabase(databaseName);
+ var batchSize = 2000;
+ var flushDuration = 500;
+ influxDb.enableBatch(batchSize, flushDuration, TimeUnit.MILLISECONDS);
+
+ return influxDb;
+ }
+
+ /**
+ * Creates a new database with the given name
+ *
+ * @param influxDb The InfluxDB client
+ * @param dbName The name of the database which should be created
+ */
+ public void createDatabase(
+ InfluxDB influxDb,
+ String dbName
+ ) throws SpRuntimeException {
+ if (!dbName.matches("^[a-zA-Z_]\\w*$")) {
+ throw new SpRuntimeException(
+ "Database name '" + dbName + "' not allowed. Allowed names:
^[a-zA-Z_][a-zA-Z0-9_]*$");
+ }
+ influxDb.query(new Query("CREATE DATABASE \"" + dbName + "\"", ""));
+ }
+
+ /**
+ * Checks whether the given database exists.
+ *
+ * @param influxDb The InfluxDB client instance
+ * @param dbName The name of the database, the method should look for
+ * @return True if the database exists, false otherwise
+ */
+ public boolean databaseExists(
+ InfluxDB influxDb,
+ String dbName
+ ) {
+ var queryResult = influxDb.query(new Query("SHOW DATABASES", ""));
+ for (List<Object> a : queryResult.getResults()
+ .get(0)
+ .getSeries()
+ .get(0)
+ .getValues()) {
+ if (!a.isEmpty() && dbName.equals(a.get(0))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
private static Environment getEnvironment() {
return Environments.getEnvironment();
}
+
}
diff --git
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxRequests.java
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxRequests.java
deleted file mode 100644
index 6522a8ee9..000000000
---
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxRequests.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.dataexplorer.commons.influx;
-
-import org.influxdb.InfluxDB;
-import org.influxdb.dto.Query;
-import org.influxdb.dto.QueryResult;
-
-import java.util.List;
-
-public class InfluxRequests {
-
- /**
- * Checks whether the given database exists.
- *
- * @param influxDb The InfluxDB client instance
- * @param dbName The name of the database, the method should look for
- * @return True if the database exists, false otherwise
- */
- public static boolean databaseExists(InfluxDB influxDb,
- String dbName) {
- QueryResult queryResult = influxDb.query(new Query("SHOW DATABASES", ""));
- for (List<Object> a :
queryResult.getResults().get(0).getSeries().get(0).getValues()) {
- if (a.get(0).equals(dbName)) {
- return true;
- }
- }
- return false;
- }
-}
diff --git
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
index 4abb33d67..0a107b1dd 100644
---
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
+++
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
@@ -20,219 +20,161 @@ package
org.apache.streampipes.dataexplorer.commons.influx;
import org.apache.streampipes.commons.environment.Environment;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import
org.apache.streampipes.dataexplorer.commons.influx.serializer.RawFieldSerializer;
import org.apache.streampipes.model.datalake.DataLakeMeasure;
import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.model.runtime.field.PrimitiveField;
+import org.apache.streampipes.model.schema.EventProperty;
import org.apache.streampipes.model.schema.EventPropertyPrimitive;
-import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.vocabulary.SO;
-import org.apache.streampipes.vocabulary.XSD;
import org.influxdb.InfluxDB;
import org.influxdb.dto.Point;
-import org.influxdb.dto.Query;
-import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
public class InfluxStore {
private static final Logger LOG = LoggerFactory.getLogger(InfluxStore.class);
- DataLakeMeasure measure;
- Map<String, String> sanitizedRuntimeNames = new HashMap<>();
- private InfluxDB influxDb = null;
- private RawFieldSerializer rawFieldSerializer = new RawFieldSerializer();
+ private final DataLakeMeasure measure;
+ private final List<EventProperty> allEventProperties;
+ private final Map<String, String> sanitizedRuntimeNames = new HashMap<>();
+ private final InfluxDB influxDb;
- public InfluxStore(DataLakeMeasure measure,
- InfluxConnectionSettings settings) {
- this.measure = measure;
- // store sanitized target property runtime names in local variable
- measure.getEventSchema()
- .getEventProperties()
- .forEach(ep -> sanitizedRuntimeNames.put(ep.getRuntimeName(),
- InfluxNameSanitizer.renameReservedKeywords(ep.getRuntimeName())));
+ private final PropertyHandler propertyHandler;
- connect(settings);
- }
- public InfluxStore(DataLakeMeasure measure,
- Environment environment) throws SpRuntimeException {
- this(measure, InfluxConnectionSettings.from(environment));
+ public InfluxStore(
+ DataLakeMeasure measure,
+ Environment environment,
+ InfluxClientProvider influxClientProvider
+ ) throws SpRuntimeException {
+ this.measure = measure;
+ storeSanitizedRuntimeNames(measure);
+ allEventProperties = getAllEventPropertiesExceptTimestamp(measure);
+ influxDb = influxClientProvider.getInitializedInfluxDBClient(environment);
+ propertyHandler = new PropertyHandler();
}
/**
- * Connects to the InfluxDB Server, sets the database and initializes the
batch-behaviour
+ * Takes an StreamPipes event, transforms it to an InfluxDB point and writes
it to the InfluxDB
*
- * @throws SpRuntimeException If not connection can be established or if the
database could not
- * be found
+ * @param event The event which should be saved
+ * @throws SpRuntimeException If the column name (key-value of the event
map) is not allowed
*/
- private void connect(InfluxConnectionSettings settings) throws
SpRuntimeException {
- influxDb = InfluxClientProvider.getInfluxDBClient(settings);
+ public void onEvent(Event event) throws SpRuntimeException {
- // Checking, if server is available
- var response = influxDb.ping();
- if (response.getVersion().equalsIgnoreCase("unknown")) {
- throw new SpRuntimeException("Could not connect to InfluxDb Server: " +
settings.getConnectionUrl());
- }
+ validateInputEventAndLogMissingFields(event);
- String databaseName = settings.getDatabaseName();
- // Checking whether the database exists
- if (!InfluxRequests.databaseExists(influxDb, databaseName)) {
- LOG.info("Database '" + databaseName + "' not found. Gets created ...");
- createDatabase(databaseName);
- }
+ sanitizeRuntimeNamesInEvent(event);
+
+ var point = initializePointWithTimestamp(event);
- // setting up the database
- influxDb.setDatabase(databaseName);
- var batchSize = 2000;
- var flushDuration = 500;
- influxDb.enableBatch(batchSize, flushDuration, TimeUnit.MILLISECONDS);
+ iterateOverallEventProperties(event, point);
+
+ influxDb.write(point.build());
+ }
+
+ private void validateInputEventAndLogMissingFields(Event event) {
+ checkEventIsNotNull(event);
+
+ logMissingFields(event);
+
+ logNullFields(event);
}
/**
- * Creates a new database with the given name
- *
- * @param dbName The name of the database which should be created
+ * Logs all fields which are present in the schema, but not in the provided
event
*/
- private void createDatabase(String dbName) throws SpRuntimeException {
- if (!dbName.matches("^[a-zA-Z_]\\w*$")) {
- throw new SpRuntimeException(
- "Database name '" + dbName + "' not allowed. Allowed names:
^[a-zA-Z_][a-zA-Z0-9_]*$");
+ private void logMissingFields(Event event) {
+ var missingFields = getMissingProperties(allEventProperties, event);
+ if (!missingFields.isEmpty()) {
+ LOG.debug(
+ "Ignored {} fields which were present in the schema, but not in the
provided event: {}",
+ missingFields.size(),
+ String.join(", ", missingFields)
+ );
}
- influxDb.query(new Query("CREATE DATABASE \"" + dbName + "\"", ""));
}
+
/**
- * Saves an event to the connected InfluxDB database
- *
- * @param event The event which should be saved
- * @throws SpRuntimeException If the column name (key-value of the event
map) is not allowed
+ * Logs all fields that contain null values
*/
- public void onEvent(Event event) throws SpRuntimeException {
- var missingFields = new ArrayList<String>();
- var nullFields = new ArrayList<String>();
- if (event == null) {
- throw new SpRuntimeException("event is null");
- }
-
- // sanitize event
- for (var key : event.getRaw().keySet()) {
- if (InfluxDbReservedKeywords.KEYWORD_LIST.stream().anyMatch(k ->
k.equalsIgnoreCase(key))) {
- event.renameFieldByRuntimeName(key, key + "_");
- }
+ private void logNullFields(Event event) {
+ List<String> nullFields = allEventProperties
+ .stream()
+ .filter(ep -> ep instanceof EventPropertyPrimitive)
+ .filter(ep -> {
+ var runtimeName = ep.getRuntimeName();
+ var field = event.getOptionalFieldByRuntimeName(runtimeName);
+
+ return field.isPresent() && field.get()
+ .getAsPrimitive()
+ .getRawValue() == null;
+ })
+ .map(EventProperty::getRuntimeName)
+ .collect(Collectors.toList());
+
+ if (!nullFields.isEmpty()) {
+ LOG.warn("Ignored {} fields which had a value 'null': {}",
nullFields.size(), String.join(", ", nullFields));
}
+ }
- var timestampValue =
event.getFieldBySelector(measure.getTimestampField()).getAsPrimitive().getAsLong();
- var point =
- Point.measurement(measure.getMeasureName()).time((long)
timestampValue, TimeUnit.MILLISECONDS);
+ private void iterateOverallEventProperties(
+ Event event,
+ Point.Builder point
+ ) {
- for (var ep : measure.getEventSchema().getEventProperties()) {
+ allEventProperties.forEach(ep -> {
var runtimeName = ep.getRuntimeName();
- // timestamp should not be added as a field
- if (!measure.getTimestampField().endsWith(runtimeName)) {
- var sanitizedRuntimeName = sanitizedRuntimeNames.get(runtimeName);
- var field = event.getOptionalFieldByRuntimeName(runtimeName);
- try {
- if (ep instanceof EventPropertyPrimitive) {
- if (field.isPresent()) {
- var eventPropertyPrimitiveField = field.get().getAsPrimitive();
- if (eventPropertyPrimitiveField.getRawValue() == null) {
- nullFields.add(sanitizedRuntimeName);
- } else {
-
- // store property as tag when the field is a dimension property
- if
(PropertyScope.DIMENSION_PROPERTY.name().equals(ep.getPropertyScope())) {
- point.tag(sanitizedRuntimeName,
eventPropertyPrimitiveField.getAsString());
- } else {
- handleMeasurementProperty(
- point,
- (EventPropertyPrimitive) ep,
- sanitizedRuntimeName,
- eventPropertyPrimitiveField);
- }
- }
- } else {
- missingFields.add(runtimeName);
- }
- } else {
- // Since InfluxDB can't store non-primitive types, store them as
string
- // and deserialize later in downstream processes
- if (field.isPresent()) {
- handleNonPrimitiveMeasurementProperty(point, event,
sanitizedRuntimeName);
- } else {
- missingFields.add(runtimeName);
- }
- }
- } catch (SpRuntimeException iae) {
- LOG.warn("Runtime exception while extracting field value of field {}
- this field will be ignored",
- runtimeName, iae);
+ var sanitizedRuntimeName = sanitizedRuntimeNames.get(runtimeName);
+ var fieldOptional = event.getOptionalFieldByRuntimeName(runtimeName);
+
+ fieldOptional.ifPresent(field -> {
+ if (ep instanceof EventPropertyPrimitive) {
+ propertyHandler.handlePrimitiveProperty(
+ point,
+ (EventPropertyPrimitive) ep,
+ field.getAsPrimitive(),
+ sanitizedRuntimeName
+ );
+ } else {
+ propertyHandler.handleNonPrimitiveProperty(
+ point,
+ event,
+ sanitizedRuntimeName
+ );
}
- }
- }
+ });
+ });
+ }
- if (missingFields.size() > 0) {
- LOG.debug("Ignored {} fields which were present in the schema, but not
in the provided event: {}",
- missingFields.size(),
- String.join(", ", missingFields));
- }
- if (nullFields.size() > 0) {
- LOG.warn("Ignored {} fields which had a value 'null': {}",
nullFields.size(), String.join(", ", nullFields));
- }
-
- influxDb.write(point.build());
+ /**
+ * Returns a list of the runtime names that are missing within the event
+ */
+ private List<String> getMissingProperties(
+ List<EventProperty> allEventProperties,
+ Event event
+ ) {
+ return allEventProperties.stream()
+ .map(EventProperty::getRuntimeName)
+ .filter(runtimeName ->
event.getOptionalFieldByRuntimeName(runtimeName)
+ .isEmpty())
+ .collect(Collectors.toList());
}
- private void handleMeasurementProperty(Point.Builder p,
- @NotNull EventPropertyPrimitive ep,
- String preparedRuntimeName,
- PrimitiveField
eventPropertyPrimitiveField) {
- try {
- // Store property according to property type
- var runtimeType = ep.getRuntimeType();
- if (XSD.INTEGER.toString().equals(runtimeType)) {
- try {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsInt());
- } catch (NumberFormatException ef) {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsFloat());
- }
- } else if (XSD.LONG.toString().equals(runtimeType)) {
- try {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsLong());
- } catch (NumberFormatException ef) {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsFloat());
- }
- } else if (XSD.FLOAT.toString().equals(runtimeType)) {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsFloat());
- } else if (XSD.DOUBLE.toString().equals(runtimeType)) {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsDouble());
- } else if (XSD.BOOLEAN.toString().equals(runtimeType)) {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsBoolean());
- } else if (SO.NUMBER.equals(runtimeType)) {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsDouble());
- } else {
- p.addField(preparedRuntimeName,
eventPropertyPrimitiveField.getAsString());
- }
- } catch (NumberFormatException e) {
- LOG.warn("Wrong number format for field {}, ignoring.",
preparedRuntimeName);
+ private void checkEventIsNotNull(Event event) {
+ if (event == null) {
+ throw new SpRuntimeException("event is null");
}
}
- private void handleNonPrimitiveMeasurementProperty(Point.Builder p, Event
event, String preparedRuntimeName) {
- try {
- var json =
rawFieldSerializer.serialize(event.getRaw().get(preparedRuntimeName));
- p.addField(preparedRuntimeName, json);
- } catch (SpRuntimeException e) {
- LOG.warn("Failed to serialize field {}, ignoring.", preparedRuntimeName);
- }
- }
/**
* Shuts down the connection to the InfluxDB server
@@ -246,4 +188,55 @@ public class InfluxStore {
}
influxDb.close();
}
+
+
+ /**
+ * Creates a point object which is later written to the influxDB and adds
the value of the timestamp field
+ */
+ private Point.Builder initializePointWithTimestamp(Event event) {
+ var timestampValue = event.getFieldBySelector(measure.getTimestampField())
+ .getAsPrimitive()
+ .getAsLong();
+ return Point.measurement(measure.getMeasureName())
+ .time((long) timestampValue, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * store sanitized target property runtime names in local variable
+ */
+ private void storeSanitizedRuntimeNames(DataLakeMeasure measure) {
+ measure.getEventSchema()
+ .getEventProperties()
+ .forEach(ep -> sanitizedRuntimeNames.put(
+ ep.getRuntimeName(),
+ InfluxNameSanitizer.renameReservedKeywords(ep.getRuntimeName())
+ ));
+ }
+
+ /**
+ * Returns all measurements properties except the timestamp field
+ */
+ private List<EventProperty>
getAllEventPropertiesExceptTimestamp(DataLakeMeasure dataLakeMeasure) {
+ return dataLakeMeasure.getEventSchema()
+ .getEventProperties()
+ .stream()
+ .filter(ep -> !dataLakeMeasure.getTimestampField()
+
.endsWith(ep.getRuntimeName()))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Iterates over all properties of the event and renames the key if it is a
reserved keywords in InfluxDB
+ */
+ private void sanitizeRuntimeNamesInEvent(Event event) {
+ // sanitize event
+ for (var key : event.getRaw()
+ .keySet()) {
+ if (InfluxDbReservedKeywords.KEYWORD_LIST.stream()
+ .anyMatch(k ->
k.equalsIgnoreCase(key))) {
+ event.renameFieldByRuntimeName(key, key + "_");
+ }
+ }
+ }
+
}
diff --git
a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/PropertyHandler.java
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/PropertyHandler.java
new file mode 100644
index 000000000..9cc4a79d1
--- /dev/null
+++
b/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/PropertyHandler.java
@@ -0,0 +1,193 @@
+/*
+ * 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.dataexplorer.commons.influx;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import
org.apache.streampipes.dataexplorer.commons.influx.serializer.RawFieldSerializer;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.field.PrimitiveField;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.vocabulary.SO;
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.influxdb.dto.Point;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PropertyHandler {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(PropertyHandler.class);
+
+ private RawFieldSerializer rawFieldSerializer;
+
+ public PropertyHandler() {
+ rawFieldSerializer = new RawFieldSerializer();
+ }
+
+ /**
+ * Takes properties of type primitive and stores them as tags or fields to
the given point
+ */
+ public void handlePrimitiveProperty(
+ Point.Builder point,
+ EventPropertyPrimitive eventPropertyPrimitive,
+ PrimitiveField primitiveField,
+ String sanitizedRuntimeName
+ ) {
+ if (primitiveField.getRawValue() != null) {
+ // store property as tag when the field is a dimension property
+ if (PropertyScope.DIMENSION_PROPERTY.name()
+
.equals(eventPropertyPrimitive.getPropertyScope())) {
+ handleDimensionProperty(
+ point,
+ primitiveField,
+ sanitizedRuntimeName
+ );
+ } else {
+ handleMeasurementProperty(
+ point,
+ eventPropertyPrimitive.getRuntimeType(),
+ sanitizedRuntimeName,
+ primitiveField
+ );
+ }
+ }
+ }
+
+
+ /**
+ * Takes non-primitive properties of and stores them as strings to the given
point
+ * Since InfluxDB can't store non-primitive types, store them as string
+ * and deserialize later in downstream processes
+ */
+ public void handleNonPrimitiveProperty(
+ Point.Builder p,
+ Event event,
+ String preparedRuntimeName
+ ) {
+ try {
+ var json = rawFieldSerializer.serialize(event.getRaw()
+ .get(preparedRuntimeName));
+ p.addField(preparedRuntimeName, json);
+ } catch (SpRuntimeException e) {
+ LOG.warn("Failed to serialize field {}, ignoring.", preparedRuntimeName);
+ }
+ }
+
+ /**
+ * Add dimension properties as tags to point
+ */
+ private void handleDimensionProperty(
+ Point.Builder point,
+ PrimitiveField primitiveField,
+ String sanitizedRuntimeName
+ ) {
+ point.tag(sanitizedRuntimeName, primitiveField.getAsString());
+ }
+
+ /**
+ * Transforms the given primitive property to the corresponding InfluxDB
data type and adds it to the point
+ */
+ private void handleMeasurementProperty(
+ Point.Builder p,
+ String runtimeType,
+ String sanitizedRuntimeName,
+ PrimitiveField eventPropertyPrimitiveField
+ ) {
+ // Store property according to property type
+ if (XSD.INTEGER.toString()
+ .equals(runtimeType)) {
+ handleIntegerProperty(p, sanitizedRuntimeName,
eventPropertyPrimitiveField);
+ } else if (XSD.LONG.toString()
+ .equals(runtimeType)) {
+ handleLongProperty(p, sanitizedRuntimeName, eventPropertyPrimitiveField);
+ } else if (XSD.FLOAT.toString()
+ .equals(runtimeType)) {
+ handleFloatProperty(p, sanitizedRuntimeName,
eventPropertyPrimitiveField);
+ } else if (XSD.DOUBLE.toString()
+ .equals(runtimeType)) {
+ handleDoubleProperty(p, sanitizedRuntimeName,
eventPropertyPrimitiveField);
+ } else if (XSD.BOOLEAN.toString()
+ .equals(runtimeType)) {
+ handleBooleanProperty(p, sanitizedRuntimeName,
eventPropertyPrimitiveField);
+ } else if (SO.NUMBER.equals(runtimeType)) {
+ handleDoubleProperty(p, sanitizedRuntimeName,
eventPropertyPrimitiveField);
+ } else {
+ handleStringProperty(p, sanitizedRuntimeName,
eventPropertyPrimitiveField);
+ }
+ }
+
+
+ private void handleStringProperty(
+ Point.Builder p,
+ String sanitizedRuntimeName,
+ PrimitiveField eventPropertyPrimitiveField
+ ) {
+ p.addField(sanitizedRuntimeName,
eventPropertyPrimitiveField.getAsString());
+ }
+
+ private void handleBooleanProperty(
+ Point.Builder p,
+ String sanitizedRuntimeName,
+ PrimitiveField eventPropertyPrimitiveField
+ ) {
+ p.addField(sanitizedRuntimeName,
eventPropertyPrimitiveField.getAsBoolean());
+ }
+
+ private void handleDoubleProperty(
+ Point.Builder p,
+ String sanitizedRuntimeName,
+ PrimitiveField eventPropertyPrimitiveField
+ ) {
+ p.addField(sanitizedRuntimeName,
eventPropertyPrimitiveField.getAsDouble());
+ }
+
+ private void handleFloatProperty(
+ Point.Builder p,
+ String sanitizedRuntimeName,
+ PrimitiveField eventPropertyPrimitiveField
+ ) {
+ p.addField(sanitizedRuntimeName, eventPropertyPrimitiveField.getAsFloat());
+ }
+
+ private void handleLongProperty(
+ Point.Builder p,
+ String sanitizedRuntimeName,
+ PrimitiveField eventPropertyPrimitiveField
+ ) {
+ try {
+ p.addField(sanitizedRuntimeName,
eventPropertyPrimitiveField.getAsLong());
+ } catch (NumberFormatException ef) {
+ handleFloatProperty(p, sanitizedRuntimeName,
eventPropertyPrimitiveField);
+ }
+ }
+
+ private void handleIntegerProperty(
+ Point.Builder p,
+ String sanitizedRuntimeName,
+ PrimitiveField eventPropertyPrimitiveField
+ ) {
+ try {
+ p.addField(sanitizedRuntimeName, eventPropertyPrimitiveField.getAsInt());
+ } catch (NumberFormatException ef) {
+ handleFloatProperty(p, sanitizedRuntimeName,
eventPropertyPrimitiveField);
+ }
+ }
+
+}
diff --git
a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProviderTest.java
b/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProviderTest.java
new file mode 100644
index 000000000..b0470669f
--- /dev/null
+++
b/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProviderTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.dataexplorer.commons.influx;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+
+public class InfluxClientProviderTest {
+
+ private InfluxDB influxDBMock;
+ private InfluxClientProvider influxClientProvider;
+
+ private static final String DATABASE_NAME = "testDB";
+
+ @Before
+ public void setUp() {
+ influxDBMock = mock(InfluxDB.class);
+ influxClientProvider = new InfluxClientProvider();
+ }
+
+ @Test
+ public void createDatabaseSuccess() {
+ var expectedQuery = new Query("CREATE DATABASE test", "");
+
+ influxClientProvider.createDatabase(influxDBMock, "test");
+
+ var pointArgumentCaptor = ArgumentCaptor.forClass(Query.class);
+ verify(influxDBMock).query(pointArgumentCaptor.capture());
+ var actualQuery = pointArgumentCaptor.getValue();
+
+ assertEquals(actualQuery.getDatabase(), expectedQuery.getDatabase());
+ }
+
+ @Test(expected = SpRuntimeException.class)
+ public void createDatabaseIlligalName() {
+ influxClientProvider.createDatabase(influxDBMock, "test%^$");
+ }
+
+ @Test
+ public void dataBaseDoesExist() {
+ var actualResult = testDataBaseExists(List.of(DATABASE_NAME));
+ assertTrue(actualResult);
+ }
+
+ @Test
+ public void dataBaseDoesNotExist() {
+ var actualResult = testDataBaseExists(List.of());
+ assertFalse(actualResult);
+ }
+
+ private boolean testDataBaseExists(List<Object> databaseNames) {
+ var queryResult = getQueryResultWithDatabaseNames(databaseNames);
+
+ when(influxDBMock.query(any())).thenReturn(queryResult);
+
+ return influxClientProvider.databaseExists(influxDBMock, DATABASE_NAME);
+ }
+
+ private QueryResult getQueryResultWithDatabaseNames(List<Object>
databaseNames) {
+ var queryResult = mock(QueryResult.class);
+ var result = mock(QueryResult.Result.class);
+ when(queryResult.getResults()).thenReturn(List.of(result));
+ var series = mock(QueryResult.Series.class);
+ when(result.getSeries()).thenReturn(List.of(series));
+
+ List<List<Object>> values = List.of(databaseNames);
+ when(series.getValues()).thenReturn(values);
+ return queryResult;
+ }
+}
\ No newline at end of file
diff --git
a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStoreTest.java
b/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStoreTest.java
new file mode 100644
index 000000000..55ca38891
--- /dev/null
+++
b/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStoreTest.java
@@ -0,0 +1,395 @@
+/*
+ * 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.dataexplorer.commons.influx;
+
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import
org.apache.streampipes.dataexplorer.commons.influx.serializer.RawFieldSerializer;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.runtime.SchemaInfo;
+import org.apache.streampipes.model.runtime.SourceInfo;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.test.generator.EventPropertyListTestBuilder;
+import org.apache.streampipes.test.generator.EventPropertyNestedTestBuilder;
+import org.apache.streampipes.test.generator.EventPropertyPrimitiveTestBuilder;
+import org.apache.streampipes.test.generator.EventSchemaTestBuilder;
+import org.apache.streampipes.vocabulary.SO;
+import org.apache.streampipes.vocabulary.XSD;
+
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.Point;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class InfluxStoreTest {
+
+ private InfluxDB influxDBMock;
+
+ private RawFieldSerializer serializer;
+
+ private static final String EXPECTED_MEASUREMENT = "testMeasure";
+ private static final String FIELD_NAME = "testId";
+ private static final String TIMESTAMP = "timestamp";
+ private static final long SAMPLE_TIMESTAMP = 1701270890000L;
+
+ @Before
+ public void setUp() {
+ influxDBMock = mock(InfluxDB.class);
+ serializer = new RawFieldSerializer();
+ }
+
+ @Test
+ public void onEventNull() {
+ var influxStore = getInfluxStore(new EventSchema());
+ assertThrows(SpRuntimeException.class, () -> influxStore.onEvent(null));
+ }
+
+
+ @Test
+ public void onEventWithInteger() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, 1)
+ .build();
+
+ var actualPoint = testEventWithOneField(XSD.INTEGER, 1);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithIntegerFloatValue() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, 1.0f)
+ .build();
+
+ var actualPoint = testEventWithOneField(XSD.INTEGER, 1.0f);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithLong() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, 1L)
+ .build();
+
+ var actualPoint = testEventWithOneField(XSD.LONG, 1L);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithLongFloatValue() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, 1.0f)
+ .build();
+
+ var actualPoint = testEventWithOneField(XSD.LONG, 1.0f);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithFloat() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, 1.0f)
+ .build();
+
+ var actualPoint = testEventWithOneField(XSD.FLOAT, 1.0f);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithDouble() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, 1.0)
+ .build();
+
+ var actualPoint = testEventWithOneField(XSD.DOUBLE, 1.0);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithBoolean() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, true)
+ .build();
+
+ var actualPoint = testEventWithOneField(XSD.BOOLEAN, true);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithNumber() throws URISyntaxException {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, 1.0)
+ .build();
+
+ var actualPoint = testEventWithOneField(new URI(SO.NUMBER), 1.0);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithString() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, "testValue")
+ .build();
+
+ var actualPoint = testEventWithOneField(XSD.STRING, "testValue");
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventReservedKeyWord() {
+ // name is a reserved keyword
+ var fieldName = "name";
+ var sanitizedFieldName = "name_";
+ var value = 1;
+
+ var expected = getPointBuilderWithTimestamp()
+ .addField(sanitizedFieldName, value)
+ .build();
+
+ var eventSchema = getEventSchemaBuilderWithTimestamp()
+ .withEventProperty(
+ EventPropertyPrimitiveTestBuilder
+ .create()
+ .withRuntimeName(sanitizedFieldName)
+ .withRuntimeType(XSD.INTEGER)
+ .build())
+ .build();
+
+ var event = getEvent(eventSchema, Map.of(fieldName, value));
+
+ var influxStore = getInfluxStore(eventSchema);
+
+ var actualPoint = executeOnEvent(influxStore, event);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithTag() {
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, "value")
+ .tag("id", "id1")
+ .build();
+
+ var eventSchema = getEventSchemaBuilderWithTimestamp()
+ .withEventProperty(
+ EventPropertyPrimitiveTestBuilder
+ .create()
+ .withRuntimeName(FIELD_NAME)
+ .withRuntimeType(XSD.STRING)
+ .build())
+ .withEventProperty(
+ EventPropertyPrimitiveTestBuilder
+ .create()
+ .withRuntimeName("id")
+ .withRuntimeType(XSD.STRING)
+ .withPropertyScope(PropertyScope.DIMENSION_PROPERTY)
+ .build())
+ .build();
+
+ var event = getEvent(eventSchema, Map.of(FIELD_NAME, "value", "id",
"id1"));
+
+ var influxStore = getInfluxStore(eventSchema);
+
+ var actualPoint = executeOnEvent(influxStore, event);
+
+ assertEquals(expected, actualPoint);
+ }
+
+
+ @Test
+ public void onEventWithNestedProperty() {
+ Map<String, Object> value = new HashMap<>();
+ value.put("one", "two");
+
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, serializer.serialize(value))
+ .build();
+
+ var eventSchema = getEventSchemaBuilderWithTimestamp()
+ .withEventProperty(
+ EventPropertyNestedTestBuilder
+ .create()
+ .withRuntimeName(FIELD_NAME)
+ .build())
+ .build();
+
+ var event = getEvent(eventSchema, Map.of(FIELD_NAME, value));
+
+ var influxStore = getInfluxStore(eventSchema);
+
+ var actualPoint = executeOnEvent(influxStore, event);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ @Test
+ public void onEventWithListProperty() {
+ String[] value = {"one", "two"};
+ var expected = getPointBuilderWithTimestamp()
+ .addField(FIELD_NAME, serializer.serialize(value))
+ .build();
+
+ var eventSchema = getEventSchemaBuilderWithTimestamp()
+ .withEventProperty(
+ EventPropertyListTestBuilder
+ .create()
+ .withRuntimeName(FIELD_NAME)
+ .build())
+ .build();
+
+ var event = getEvent(eventSchema, Map.of(FIELD_NAME, value));
+
+ var influxStore = getInfluxStore(eventSchema);
+
+ var actualPoint = executeOnEvent(influxStore, event);
+
+ assertEquals(expected, actualPoint);
+ }
+
+ /**
+ * Initialize a Point builder with a timestamp
+ */
+ private Point.Builder getPointBuilderWithTimestamp() {
+ return Point.measurement(EXPECTED_MEASUREMENT)
+ .time(SAMPLE_TIMESTAMP, TimeUnit.MILLISECONDS);
+ }
+
+
+ /**
+ * Executes the onEvent method and returns the resuting data point using an
argument captor
+ */
+ private Point executeOnEvent(InfluxStore influxStore, Event event) {
+ influxStore.onEvent(event);
+ var pointArgumentCaptor = ArgumentCaptor.forClass(Point.class);
+
+ verify(influxDBMock).write(pointArgumentCaptor.capture());
+
+ var actualPoint = pointArgumentCaptor.getValue();
+ return actualPoint;
+ }
+
+ /**
+ * Creates an event with the given data and a timestamp
+ */
+ private Event getEvent(EventSchema eventSchema, Map<String, Object>
inputData) {
+ Map<String, Object> data = new HashMap<>();
+ data.put(TIMESTAMP, SAMPLE_TIMESTAMP);
+ data.putAll(inputData);
+
+ return EventFactory.fromMap(
+ data,
+ new SourceInfo("test-topic", "s0"),
+ new SchemaInfo(eventSchema, new ArrayList<>())
+ );
+ }
+
+
+ /**
+ * This convenience method is used to test all cases for the different
supported data types
+ */
+ private Point testEventWithOneField(
+ URI type,
+ Object value
+ ) {
+
+ var eventSchema = getEventSchemaBuilderWithTimestamp()
+ .withEventProperty(
+ EventPropertyPrimitiveTestBuilder
+ .create()
+ .withRuntimeName(FIELD_NAME)
+ .withRuntimeType(type)
+ .build())
+ .build();
+
+
+ var event = EventFactory.fromMap(
+ Map.of(TIMESTAMP, SAMPLE_TIMESTAMP, FIELD_NAME, value),
+ new SourceInfo("test-topic", "s0"),
+ new SchemaInfo(eventSchema, new ArrayList<>())
+ );
+
+ var influxStore = getInfluxStore(eventSchema);
+
+ influxStore.onEvent(event);
+ var pointArgumentCaptor = ArgumentCaptor.forClass(Point.class);
+
+ verify(influxDBMock).write(pointArgumentCaptor.capture());
+ return pointArgumentCaptor.getValue();
+ }
+
+ /**
+ * Returns the event schema builder with a default timestamp property
+ */
+ private EventSchemaTestBuilder getEventSchemaBuilderWithTimestamp() {
+ return EventSchemaTestBuilder
+ .create()
+ .withEventProperty(
+ EventPropertyPrimitiveTestBuilder
+ .create()
+ .withRuntimeName(TIMESTAMP)
+ .build());
+ }
+
+ /**
+ * Initializes an influx store with the given event schema
+ */
+ private InfluxStore getInfluxStore(EventSchema eventSchema) {
+
+ DataLakeMeasure measure = new DataLakeMeasure(
+ EXPECTED_MEASUREMENT,
+ "s0::%s".formatted(TIMESTAMP),
+ eventSchema
+ );
+
+ var influxClientProviderMock = mock(InfluxClientProvider.class);
+ when(influxClientProviderMock.getInitializedInfluxDBClient(any()))
+ .thenReturn(influxDBMock);
+
+ return new InfluxStore(measure, null, influxClientProviderMock);
+ }
+
+}
\ No newline at end of file
diff --git
a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
index 81a9da6cc..b370f415f 100644
---
a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
+++
b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
@@ -20,8 +20,8 @@ package
org.apache.streampipes.extensions.connectors.influx.adapter;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
import
org.apache.streampipes.dataexplorer.commons.influx.InfluxConnectionSettings;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxRequests;
import
org.apache.streampipes.extensions.connectors.influx.shared.SharedInfluxClient;
import org.apache.streampipes.model.connect.guess.GuessSchema;
import org.apache.streampipes.model.schema.EventProperty;
@@ -56,6 +56,8 @@ public class InfluxDbClient extends SharedInfluxClient {
private boolean connected;
+ private final InfluxClientProvider influxClientProvider;
+
public static class Column {
private final String name;
private final Datatypes datatypes;
@@ -79,6 +81,7 @@ public class InfluxDbClient extends SharedInfluxClient {
boolean replaceNullValues) {
super(connectionSettings, measurement);
this.replaceNullValues = replaceNullValues;
+ this.influxClientProvider = new InfluxClientProvider();
this.connected = false;
}
@@ -89,7 +92,7 @@ public class InfluxDbClient extends SharedInfluxClient {
var database = connectionSettings.getDatabaseName();
// Checking whether the database exists
- if (!InfluxRequests.databaseExists(influxDb, database)) {
+ if (!influxClientProvider.databaseExists(influxDb, database)) {
throw new AdapterException("Database " + database + " could not be
found.");
}
diff --git
a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
index d81938839..63b8ce2f6 100644
---
a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
+++
b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
@@ -19,8 +19,8 @@
package org.apache.streampipes.extensions.connectors.influx.sink;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
import
org.apache.streampipes.dataexplorer.commons.influx.InfluxConnectionSettings;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxRequests;
import
org.apache.streampipes.extensions.connectors.influx.shared.SharedInfluxClient;
import org.apache.streampipes.model.runtime.Event;
@@ -41,6 +41,8 @@ public class InfluxDbClient extends SharedInfluxClient {
private final Integer batchSize;
private final Integer flushDuration;
+ private final InfluxClientProvider influxClientProvider;
+
InfluxDbClient(InfluxConnectionSettings connectionSettings,
String measureName,
@@ -52,6 +54,7 @@ public class InfluxDbClient extends SharedInfluxClient {
this.timestampField = timestampField;
this.batchSize = batchSize;
this.flushDuration = flushDuration;
+ this.influxClientProvider = new InfluxClientProvider();
connect();
}
@@ -67,7 +70,7 @@ public class InfluxDbClient extends SharedInfluxClient {
var databaseName = connectionSettings.getDatabaseName();
// Checking whether the database exists
- if (!InfluxRequests.databaseExists(influxDb, databaseName)) {
+ if (!influxClientProvider.databaseExists(influxDb, databaseName)) {
LOG.info("Database '" + databaseName + "' not found. Gets created ...");
createDatabase(databaseName);
}
diff --git
a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyListTestBuilder.java
similarity index 55%
copy from
streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
copy to
streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyListTestBuilder.java
index 8fd69330d..2ac224a05 100644
---
a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
+++
b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyListTestBuilder.java
@@ -15,35 +15,30 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.test.generator;
-import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+package org.apache.streampipes.test.generator;
-import java.net.URI;
-import java.util.List;
+import org.apache.streampipes.model.schema.EventPropertyList;
-public class EventPropertyPrimitiveTestBuilder
- extends EventPropertyTestBuilder<EventPropertyPrimitive,
EventPropertyPrimitiveTestBuilder> {
+public class EventPropertyListTestBuilder
+ extends EventPropertyTestBuilder<EventPropertyList,
EventPropertyListTestBuilder> {
- protected EventPropertyPrimitiveTestBuilder() {
- super(new EventPropertyPrimitive());
+ protected EventPropertyListTestBuilder() {
+ super(new EventPropertyList());
}
- public EventPropertyPrimitiveTestBuilder withSemanticType(String
semanticType) {
- this.eventProperty.setDomainProperties(List.of(URI.create(semanticType)));
- return this;
+ public static EventPropertyNestedTestBuilder create() {
+ return new EventPropertyNestedTestBuilder();
}
- public static EventPropertyPrimitiveTestBuilder create() {
- return new EventPropertyPrimitiveTestBuilder();
+ @Override
+ protected EventPropertyListTestBuilder me() {
+ return this;
}
@Override
- public EventPropertyPrimitive build() {
+ public EventPropertyList build() {
return eventProperty;
}
- protected EventPropertyPrimitiveTestBuilder me() {
- return this;
- }
-}
+}
\ No newline at end of file
diff --git
a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
index 8fd69330d..fc5c1babc 100644
---
a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
+++
b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/EventPropertyPrimitiveTestBuilder.java
@@ -18,6 +18,7 @@
package org.apache.streampipes.test.generator;
import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.PropertyScope;
import java.net.URI;
import java.util.List;
@@ -29,6 +30,16 @@ public class EventPropertyPrimitiveTestBuilder
super(new EventPropertyPrimitive());
}
+ public EventPropertyPrimitiveTestBuilder withRuntimeType(URI runtimeType) {
+ this.eventProperty.setRuntimeType(runtimeType.toString());
+ return this;
+ }
+
+ public EventPropertyPrimitiveTestBuilder withPropertyScope(PropertyScope
propertyScope) {
+ this.eventProperty.setPropertyScope(propertyScope.toString());
+ return this;
+ }
+
public EventPropertyPrimitiveTestBuilder withSemanticType(String
semanticType) {
this.eventProperty.setDomainProperties(List.of(URI.create(semanticType)));
return this;
