This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch 1618-migrate-data-sinks-in-databases-jvm-module
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/1618-migrate-data-sinks-in-databases-jvm-module by this push:
new b03d09193 Migrate data sinks in databases-jvm module (#1618)
b03d09193 is described below
commit b03d0919399a848551d348022b70dd77d24f517e
Author: Dominik Riemer <[email protected]>
AuthorDate: Fri May 26 22:51:27 2023 +0200
Migrate data sinks in databases-jvm module (#1618)
---
.../sinks/databases/jvm/DatabasesJvmInit.java | 20 ++---
.../sinks/databases/jvm/couchdb/CouchDb.java | 59 ---------------
.../databases/jvm/couchdb/CouchDbParameters.java | 61 ---------------
.../{CouchDbController.java => CouchDbSink.java} | 43 ++++++++---
.../sinks/databases/jvm/ditto/DittoController.java | 81 --------------------
.../sinks/databases/jvm/ditto/DittoParameters.java | 75 -------------------
.../jvm/ditto/{Ditto.java => DittoSink.java} | 62 ++++++++++++---
.../sinks/databases/jvm/iotdb/IotDbController.java | 80 --------------------
.../sinks/databases/jvm/iotdb/IotDbParameters.java | 87 ----------------------
.../jvm/iotdb/{IotDb.java => IotDbSink.java} | 76 +++++++++++++++----
.../sinks/databases/jvm/opcua/OpcUa.java | 47 ++++++------
.../sinks/databases/jvm/opcua/OpcUaParameters.java | 67 ++++-------------
.../opcua/{UpcUaController.java => OpcUaSink.java} | 33 +++++---
.../sinks/databases/jvm/postgresql/PostgreSql.java | 9 +--
.../jvm/postgresql/PostgreSqlParameters.java | 3 -
...stgreSqlController.java => PostgreSqlSink.java} | 82 +++++++++++---------
.../databases/jvm/iotdb/TestIotDbParameters.java | 52 -------------
17 files changed, 264 insertions(+), 673 deletions(-)
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabasesJvmInit.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabasesJvmInit.java
index 23b2da117..313551c26 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabasesJvmInit.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabasesJvmInit.java
@@ -28,11 +28,11 @@ import
org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
-import org.apache.streampipes.sinks.databases.jvm.couchdb.CouchDbController;
-import org.apache.streampipes.sinks.databases.jvm.ditto.DittoController;
-import org.apache.streampipes.sinks.databases.jvm.iotdb.IotDbController;
-import org.apache.streampipes.sinks.databases.jvm.opcua.UpcUaController;
-import
org.apache.streampipes.sinks.databases.jvm.postgresql.PostgreSqlController;
+import org.apache.streampipes.sinks.databases.jvm.couchdb.CouchDbSink;
+import org.apache.streampipes.sinks.databases.jvm.ditto.DittoSink;
+import org.apache.streampipes.sinks.databases.jvm.iotdb.IotDbSink;
+import org.apache.streampipes.sinks.databases.jvm.opcua.OpcUaSink;
+import org.apache.streampipes.sinks.databases.jvm.postgresql.PostgreSqlSink;
import org.apache.streampipes.sinks.databases.jvm.redis.RedisController;
public class DatabasesJvmInit extends ExtensionsModelSubmitter {
@@ -48,11 +48,11 @@ public class DatabasesJvmInit extends
ExtensionsModelSubmitter {
"",
8090)
.registerPipelineElements(
- new CouchDbController(),
- new UpcUaController(),
- new PostgreSqlController(),
- new IotDbController(),
- new DittoController(),
+ new CouchDbSink(),
+ new OpcUaSink(),
+ new PostgreSqlSink(),
+ new IotDbSink(),
+ new DittoSink(),
new RedisController())
.registerMessagingFormats(
new JsonDataFormatFactory(),
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDb.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDb.java
deleted file mode 100644
index c25c71154..000000000
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDb.java
+++ /dev/null
@@ -1,59 +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.sinks.databases.jvm.couchdb;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.model.runtime.EventConverter;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
-
-import org.lightcouch.CouchDbClient;
-import org.lightcouch.CouchDbProperties;
-
-public class CouchDb implements EventSink<CouchDbParameters> {
-
- private CouchDbClient couchDbClient;
-
- @Override
- public void onInvocation(CouchDbParameters parameters,
EventSinkRuntimeContext runtimeContext) throws
- SpRuntimeException {
- this.couchDbClient = new CouchDbClient(new CouchDbProperties(
- parameters.getDatabaseName(),
- true,
- "http",
- parameters.getCouchDbHost(),
- parameters.getCouchDbPort(),
- parameters.getUser(),
- parameters.getPassword()
- ));
- }
-
- @Override
- public void onEvent(Event inputEvent) {
- couchDbClient.save(new EventConverter(inputEvent).toInputEventMap());
- }
-
- @Override
- public void onDetach() throws SpRuntimeException {
- this.couchDbClient.shutdown();
- }
-
-
-}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbParameters.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbParameters.java
deleted file mode 100644
index ee2321fa7..000000000
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbParameters.java
+++ /dev/null
@@ -1,61 +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.sinks.databases.jvm.couchdb;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class CouchDbParameters extends EventSinkBindingParams {
-
- private String couchDbHost;
- private Integer couchDbPort;
- private String databaseName;
- private String user;
- private String password;
-
- public CouchDbParameters(DataSinkInvocation graph, String couchDbHost,
Integer couchDbPort, String databaseName,
- String user, String password) {
- super(graph);
- this.couchDbHost = couchDbHost;
- this.couchDbPort = couchDbPort;
- this.databaseName = databaseName;
- this.user = user;
- this.password = password;
- }
-
- public String getCouchDbHost() {
- return couchDbHost;
- }
-
- public Integer getCouchDbPort() {
- return couchDbPort;
- }
-
- public String getDatabaseName() {
- return databaseName;
- }
-
- public String getUser() {
- return user;
- }
-
- public String getPassword() {
- return password;
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbController.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSink.java
similarity index 70%
rename from
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbController.java
rename to
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSink.java
index 9adf5c7f4..775ddf131 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbController.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSink.java
@@ -18,20 +18,25 @@
package org.apache.streampipes.sinks.databases.jvm.couchdb;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.model.DataSinkType;
import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventConverter;
import org.apache.streampipes.sdk.builder.DataSinkBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
import org.apache.streampipes.sdk.helpers.EpRequirements;
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.wrapper.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
+import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
-public class CouchDbController extends
StandaloneEventSinkDeclarer<CouchDbParameters> {
+import org.lightcouch.CouchDbClient;
+import org.lightcouch.CouchDbProperties;
+
+public class CouchDbSink extends StreamPipesDataSink {
private static final String DATABASE_HOST_KEY = "db_host";
private static final String DATABASE_PORT_KEY = "db_port";
@@ -39,6 +44,8 @@ public class CouchDbController extends
StandaloneEventSinkDeclarer<CouchDbParame
private static final String DATABASE_USER_KEY = "db_user";
private static final String DATABASE_PASSORD_KEY = "db_password";
+ private CouchDbClient couchDbClient;
+
@Override
public DataSinkDescription declareModel() {
return
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.couchdb")
@@ -55,11 +62,10 @@ public class CouchDbController extends
StandaloneEventSinkDeclarer<CouchDbParame
.build();
}
-
@Override
- public ConfiguredEventSink<CouchDbParameters>
onInvocation(DataSinkInvocation graph,
-
DataSinkParameterExtractor extractor) {
-
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
String hostname = extractor.singleValueParameter(DATABASE_HOST_KEY,
String.class);
Integer port = extractor.singleValueParameter(DATABASE_PORT_KEY,
Integer.class);
String dbName = extractor.singleValueParameter(DATABASE_NAME_KEY,
String.class);
@@ -71,9 +77,24 @@ public class CouchDbController extends
StandaloneEventSinkDeclarer<CouchDbParame
String user = null;
String password = null;
- CouchDbParameters params = new CouchDbParameters(graph, hostname, port,
dbName, user, password);
+ this.couchDbClient = new CouchDbClient(new CouchDbProperties(
+ dbName,
+ true,
+ "http",
+ hostname,
+ port,
+ user,
+ password
+ ));
+ }
- return new ConfiguredEventSink<>(params, CouchDb::new);
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
+ couchDbClient.save(new EventConverter(event).toInputEventMap());
}
+ @Override
+ public void onDetach() throws SpRuntimeException {
+ this.couchDbClient.shutdown();
+ }
}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoController.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoController.java
deleted file mode 100644
index ece4ee68e..000000000
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoController.java
+++ /dev/null
@@ -1,81 +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.sinks.databases.jvm.ditto;
-
-import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.sdk.builder.DataSinkBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
-import org.apache.streampipes.sdk.helpers.EpRequirements;
-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.wrapper.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
-
-import java.util.List;
-
-public class DittoController extends
StandaloneEventSinkDeclarer<DittoParameters> {
-
- private static final String DITTO_API_ENDPOINT_KEY = "dittoApiEndpointKey";
- private static final String DITTO_USER_KEY = "dittoUserKey";
- private static final String DITTO_PASSWORD_KEY = "dittoPasswordKey";
-
- private static final String DITTO_THING_ID_KEY = "dittoThingIdKey";
- private static final String DITTO_FEATURE_ID_KEY = "dittoFeatureIdKey";
-
- private static final String SELECTED_FIELDS_KEY = "selectedFieldsKey";
-
- @Override
- public DataSinkDescription declareModel() {
- return
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.ditto")
- .category(DataSinkType.FORWARD)
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
-
.requiredStream(StreamRequirementsBuilder.create().requiredPropertyWithNaryMapping(
- EpRequirements.anyProperty(),
- Labels.withId(SELECTED_FIELDS_KEY),
- PropertyScope.NONE).build())
- .requiredTextParameter(Labels.withId(DITTO_API_ENDPOINT_KEY))
- .requiredTextParameter(Labels.withId(DITTO_USER_KEY))
- .requiredSecret(Labels.withId(DITTO_PASSWORD_KEY))
- .requiredTextParameter(Labels.withId(DITTO_THING_ID_KEY))
- .requiredTextParameter(Labels.withId(DITTO_FEATURE_ID_KEY))
- .build();
- }
-
- @Override
- public ConfiguredEventSink<DittoParameters> onInvocation(DataSinkInvocation
graph,
-
DataSinkParameterExtractor extractor) {
- String dittoApiEndpoint = extractor.textParameter(DITTO_API_ENDPOINT_KEY);
- String dittoUser = extractor.textParameter(DITTO_USER_KEY);
- String dittoPassword = extractor.secretValue(DITTO_PASSWORD_KEY);
- String dittoThingId = extractor.textParameter(DITTO_THING_ID_KEY);
- String dittoFeatureId = extractor.textParameter(DITTO_FEATURE_ID_KEY);
-
- List<String> selectedFieldSelectors =
extractor.mappingPropertyValues(SELECTED_FIELDS_KEY);
- DittoParameters params = new DittoParameters(graph, dittoApiEndpoint,
dittoUser,
- dittoPassword, dittoThingId, dittoFeatureId, selectedFieldSelectors);
-
- return new ConfiguredEventSink<>(params, Ditto::new);
- }
-
-}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoParameters.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoParameters.java
deleted file mode 100644
index 5fd7dcc01..000000000
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoParameters.java
+++ /dev/null
@@ -1,75 +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.sinks.databases.jvm.ditto;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-import java.util.List;
-
-public class DittoParameters extends EventSinkBindingParams {
-
- private String dittoApiEndpoint;
- private String dittoUser;
- private String dittoPassword;
-
- private String thingId;
- private String featureId;
-
- private List<String> selectedFields;
-
- public DittoParameters(DataSinkInvocation graph,
- String dittoApiEndpoint,
- String dittoUser,
- String dittoPassword,
- String thingId,
- String featureId,
- List<String> selectedFields) {
- super(graph);
- this.dittoApiEndpoint = dittoApiEndpoint;
- this.dittoUser = dittoUser;
- this.dittoPassword = dittoPassword;
- this.thingId = thingId;
- this.featureId = featureId;
- this.selectedFields = selectedFields;
- }
-
- public String getDittoApiEndpoint() {
- return dittoApiEndpoint;
- }
-
- public String getDittoUser() {
- return dittoUser;
- }
-
- public String getDittoPassword() {
- return dittoPassword;
- }
-
- public String getThingId() {
- return thingId;
- }
-
- public String getFeatureId() {
- return featureId;
- }
-
- public List<String> getSelectedFields() {
- return selectedFields;
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/Ditto.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoSink.java
similarity index 60%
rename from
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/Ditto.java
rename to
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoSink.java
index 13d9d9f85..d4b8086cd 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/Ditto.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/DittoSink.java
@@ -18,9 +18,19 @@
package org.apache.streampipes.sinks.databases.jvm.ditto;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.graph.DataSinkDescription;
import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.sdk.builder.DataSinkBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+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.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
import org.eclipse.ditto.client.DittoClient;
import org.eclipse.ditto.client.DittoClients;
@@ -41,7 +51,16 @@ import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
-public class Ditto implements EventSink<DittoParameters> {
+public class DittoSink extends StreamPipesDataSink {
+
+ private static final String DITTO_API_ENDPOINT_KEY = "dittoApiEndpointKey";
+ private static final String DITTO_USER_KEY = "dittoUserKey";
+ private static final String DITTO_PASSWORD_KEY = "dittoPasswordKey";
+
+ private static final String DITTO_THING_ID_KEY = "dittoThingIdKey";
+ private static final String DITTO_FEATURE_ID_KEY = "dittoFeatureIdKey";
+
+ private static final String SELECTED_FIELDS_KEY = "selectedFieldsKey";
private DittoClient client;
private String thingId;
@@ -49,22 +68,45 @@ public class Ditto implements EventSink<DittoParameters> {
private List<String> selectedFields;
+
+ @Override
+ public DataSinkDescription declareModel() {
+ return
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.ditto")
+ .category(DataSinkType.FORWARD)
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+
.requiredStream(StreamRequirementsBuilder.create().requiredPropertyWithNaryMapping(
+ EpRequirements.anyProperty(),
+ Labels.withId(SELECTED_FIELDS_KEY),
+ PropertyScope.NONE).build())
+ .requiredTextParameter(Labels.withId(DITTO_API_ENDPOINT_KEY))
+ .requiredTextParameter(Labels.withId(DITTO_USER_KEY))
+ .requiredSecret(Labels.withId(DITTO_PASSWORD_KEY))
+ .requiredTextParameter(Labels.withId(DITTO_THING_ID_KEY))
+ .requiredTextParameter(Labels.withId(DITTO_FEATURE_ID_KEY))
+ .build();
+ }
+
@Override
- public void onInvocation(DittoParameters parameters, EventSinkRuntimeContext
runtimeContext)
- throws SpRuntimeException {
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
+ String dittoApiEndpoint = extractor.textParameter(DITTO_API_ENDPOINT_KEY);
+ String dittoUser = extractor.textParameter(DITTO_USER_KEY);
+ String dittoPassword = extractor.secretValue(DITTO_PASSWORD_KEY);
- this.thingId = parameters.getThingId();
- this.featureId = parameters.getFeatureId();
- this.selectedFields = parameters.getSelectedFields();
+ this.thingId = extractor.textParameter(DITTO_THING_ID_KEY);
+ this.featureId = extractor.textParameter(DITTO_FEATURE_ID_KEY);
+ this.selectedFields = extractor.mappingPropertyValues(SELECTED_FIELDS_KEY);
MessagingConfiguration configuration =
WebSocketMessagingConfiguration.newBuilder()
- .endpoint(parameters.getDittoApiEndpoint())
+ .endpoint(dittoApiEndpoint)
.build();
WebSocketMessagingProvider provider =
WebSocketMessagingProvider.newInstance(configuration,
AuthenticationProviders
.basic(BasicAuthenticationConfiguration.newBuilder()
- .username(parameters.getDittoUser())
- .password(parameters.getDittoPassword())
+ .username(dittoUser)
+ .password(dittoPassword)
.build()),
Executors.newFixedThreadPool(4));
this.client = DittoClients.newInstance(provider);
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java
deleted file mode 100644
index 7bc974bba..000000000
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbController.java
+++ /dev/null
@@ -1,80 +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.sinks.databases.jvm.iotdb;
-
-import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.sdk.builder.DataSinkBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
-import org.apache.streampipes.sdk.helpers.EpRequirements;
-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.wrapper.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
-
-public class IotDbController extends
StandaloneEventSinkDeclarer<IotDbParameters> {
-
- private static final String HOST_KEY = "db_host";
- private static final String PORT_KEY = "db_port";
-
- private static final String DATABASE_KEY = "db_database";
- private static final String DEVICE_KEY = "db_device";
-
- private static final String USER_KEY = "db_user";
- private static final String PASSWORD_KEY = "db_password";
-
- private static final String TIMESTAMP_MAPPING_KEY = "timestamp_mapping";
-
- @Override
- public DataSinkDescription declareModel() {
- return
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.iotdb").withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION,
Assets.ICON).category(DataSinkType.DATABASE).requiredStream(
- StreamRequirementsBuilder.create()
-
.requiredPropertyWithUnaryMapping(EpRequirements.timestampReq(),
Labels.withId(TIMESTAMP_MAPPING_KEY),
-
PropertyScope.NONE).build()).requiredTextParameter(Labels.withId(HOST_KEY))
- .requiredIntegerParameter(Labels.withId(PORT_KEY),
6667).requiredTextParameter(Labels.withId(USER_KEY), "root")
-
.requiredSecret(Labels.withId(PASSWORD_KEY)).requiredTextParameter(Labels.withId(DATABASE_KEY))
- .requiredTextParameter(Labels.withId(DEVICE_KEY)).build();
- }
-
- @Override
- public ConfiguredEventSink<IotDbParameters> onInvocation(DataSinkInvocation
graph,
-
DataSinkParameterExtractor extractor) {
-
- final String host = extractor.singleValueParameter(HOST_KEY, String.class);
- final Integer port = extractor.singleValueParameter(PORT_KEY,
Integer.class);
-
- final String database = extractor.singleValueParameter(DATABASE_KEY,
String.class);
- final String device = extractor.singleValueParameter(DEVICE_KEY,
String.class);
-
- final String user = extractor.singleValueParameter(USER_KEY, String.class);
- final String password = extractor.secretValue(PASSWORD_KEY);
-
- final String timestampField =
extractor.mappingPropertyValue(TIMESTAMP_MAPPING_KEY);
-
- final IotDbParameters params =
- new IotDbParameters(graph, host, port, user, password, database,
device, timestampField);
-
- return new ConfiguredEventSink<>(params, IotDb::new);
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java
deleted file mode 100644
index 4f41610c8..000000000
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbParameters.java
+++ /dev/null
@@ -1,87 +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.sinks.databases.jvm.iotdb;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class IotDbParameters extends EventSinkBindingParams {
-
- private final String host;
- private final Integer port;
-
- private final String user;
- private final String password;
-
- private final String database;
- // entity
- private final String device;
-
- private final String timestampField;
-
- public IotDbParameters(DataSinkInvocation graph,
- String host,
- Integer port,
- String user,
- String password,
- String database,
- String device,
- String timestampField) {
- super(graph);
-
- this.host = host;
- this.port = port;
-
- this.user = user;
- this.password = password;
-
- this.database = database;
- this.device = device;
-
- this.timestampField = timestampField;
- }
-
- public String getHost() {
- return host;
- }
-
- public Integer getPort() {
- return port;
- }
-
- public String getUser() {
- return user;
- }
-
- public String getPassword() {
- return password;
- }
-
- public String getDatabase() {
- return database;
- }
-
- public String getDevice() {
- return device;
- }
-
- public String getTimestampField() {
- return timestampField;
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
similarity index 55%
rename from
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
rename to
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
index df8300f71..2af93034b 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDbSink.java
@@ -18,25 +18,48 @@
package org.apache.streampipes.sinks.databases.jvm.iotdb;
-import org.apache.streampipes.logging.api.Logger;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.graph.DataSinkDescription;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.runtime.field.AbstractField;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.sdk.builder.DataSinkBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+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.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-public class IotDb implements EventSink<IotDbParameters> {
+public class IotDbSink extends StreamPipesDataSink {
- private static Logger logger;
+ private static final Logger LOG = LoggerFactory.getLogger(IotDbSink.class);
+
+ private static final String HOST_KEY = "db_host";
+ private static final String PORT_KEY = "db_port";
+
+ private static final String DATABASE_KEY = "db_database";
+ private static final String DEVICE_KEY = "db_device";
+
+ private static final String USER_KEY = "db_user";
+ private static final String PASSWORD_KEY = "db_password";
+
+ private static final String TIMESTAMP_MAPPING_KEY = "timestamp_mapping";
private String timestampFieldId;
private String deviceId;
@@ -50,20 +73,47 @@ public class IotDb implements EventSink<IotDbParameters> {
private SessionPool sessionPool;
@Override
- public void onInvocation(IotDbParameters parameters, EventSinkRuntimeContext
runtimeContext) {
- logger = parameters.getGraph().getLogger(IotDb.class);
+ public DataSinkDescription declareModel() {
+ return
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.iotdb").withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION,
Assets.ICON).category(DataSinkType.DATABASE).requiredStream(
+ StreamRequirementsBuilder.create()
+
.requiredPropertyWithUnaryMapping(EpRequirements.timestampReq(),
Labels.withId(TIMESTAMP_MAPPING_KEY),
+
PropertyScope.NONE).build()).requiredTextParameter(Labels.withId(HOST_KEY))
+ .requiredIntegerParameter(Labels.withId(PORT_KEY),
6667).requiredTextParameter(Labels.withId(USER_KEY), "root")
+
.requiredSecret(Labels.withId(PASSWORD_KEY)).requiredTextParameter(Labels.withId(DATABASE_KEY))
+ .requiredTextParameter(Labels.withId(DEVICE_KEY)).build();
+ }
+
+ @Override
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
+ final String host = extractor.singleValueParameter(HOST_KEY, String.class);
+ final Integer port = extractor.singleValueParameter(PORT_KEY,
Integer.class);
+
+ final String database = extractor.singleValueParameter(DATABASE_KEY,
String.class);
+ final String device = extractor.singleValueParameter(DEVICE_KEY,
String.class);
+
+ final String user = extractor.singleValueParameter(USER_KEY, String.class);
+ final String password = extractor.secretValue(PASSWORD_KEY);
- deviceId = "root." + parameters.getDatabase() + "." +
parameters.getDevice();
- timestampFieldId = parameters.getTimestampField();
+ timestampFieldId = extractor.mappingPropertyValue(TIMESTAMP_MAPPING_KEY);
+ deviceId = "root." + database + "." + device;
// In our case, the pool size is set to 2.
// One connection is for current requests, and the other is a backup for
fast-recovery when connection dies.
- sessionPool = new
SessionPool.Builder().maxSize(2).enableCompression(false).host(parameters.getHost())
-
.port(parameters.getPort()).user(parameters.getUser()).password(parameters.getPassword()).build();
+ sessionPool = new SessionPool.Builder()
+ .maxSize(2)
+ .enableCompression(false)
+ .host(host)
+ .port(port)
+ .user(user)
+ .password(password)
+ .build();
}
@Override
- public void onEvent(Event event) {
+ public void onEvent(Event event) throws SpRuntimeException {
if (event == null) {
return;
}
@@ -117,13 +167,13 @@ public class IotDb implements EventSink<IotDbParameters> {
try {
sessionPool.insertRecord(deviceId, timestamp, measurements, types,
values);
} catch (IoTDBConnectionException | StatementExecutionException e) {
- logger.error("Failed to save event to IoTDB, because: " +
e.getMessage());
+ LOG.error("Failed to save event to IoTDB, because: " + e.getMessage());
e.printStackTrace();
}
}
@Override
- public void onDetach() {
+ public void onDetach() throws SpRuntimeException {
sessionPool.close();
}
}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java
index 2004bb908..5f2b3d25c 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java
@@ -19,12 +19,9 @@
package org.apache.streampipes.sinks.databases.jvm.opcua;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.logging.api.Logger;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.runtime.field.PrimitiveField;
import org.apache.streampipes.vocabulary.XSD;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.api.config.OpcUaClientConfig;
@@ -37,6 +34,8 @@ import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.HashMap;
@@ -44,9 +43,9 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-public class OpcUa implements EventSink<OpcUaParameters> {
+public class OpcUa {
- private static Logger log;
+ private static final Logger LOG = LoggerFactory.getLogger(OpcUa.class);
private OpcUaClient opcUaClient;
private OpcUaParameters params;
@@ -78,21 +77,19 @@ public class OpcUa implements EventSink<OpcUaParameters> {
compatibleDataTypes.put(String.class, new Class[]{String.class});
}
- @Override
- public void onInvocation(OpcUaParameters parameters, EventSinkRuntimeContext
runtimeContext) throws
+ public void onInvocation(OpcUaParameters parameters) throws
SpRuntimeException {
- log = parameters.getGraph().getLogger(OpcUa.class);
- if (!parameters.getHostName().startsWith("opc.tcp://")) {
- serverUrl = "opc.tcp://" + parameters.getHostName() + ":" +
parameters.getPort();
+ if (!parameters.hostname().startsWith("opc.tcp://")) {
+ serverUrl = "opc.tcp://" + parameters.hostname() + ":" +
parameters.port();
} else {
- serverUrl = parameters.getHostName() + ":" + parameters.getPort();
+ serverUrl = parameters.hostname() + ":" + parameters.port();
}
- if (isInteger(parameters.getNodeId())) {
- int integerNodeId = Integer.parseInt(parameters.getNodeId());
- node = new NodeId(parameters.getNameSpaceIndex(), integerNodeId);
+ if (isInteger(parameters.nodeId())) {
+ int integerNodeId = Integer.parseInt(parameters.nodeId());
+ node = new NodeId(parameters.nameSpaceIndex(), integerNodeId);
} else {
- node = new NodeId(parameters.getNameSpaceIndex(),
parameters.getNodeId());
+ node = new NodeId(parameters.nameSpaceIndex(), parameters.nodeId());
}
@@ -126,7 +123,7 @@ public class OpcUa implements EventSink<OpcUaParameters> {
try {
Variant value =
opcUaClient.getAddressSpace().getVariableNode(node).readValue().getValue();
targetDataType = value.getValue().getClass();
- sourceDataType = XSDMatchings.get(params.getMappingPropertyType());
+ sourceDataType = XSDMatchings.get(params.mappingPropertyType());
if (!sourceDataType.equals(targetDataType)) {
if
(!Arrays.stream(compatibleDataTypes.get(sourceDataType)).anyMatch(dt ->
dt.equals(targetDataType))) {
throw new SpRuntimeException("Data Type of event of target node are
not compatible");
@@ -138,13 +135,12 @@ public class OpcUa implements EventSink<OpcUaParameters> {
}
- @Override
public void onEvent(Event inputEvent) {
Variant v = getValue(inputEvent);
if (v == null) {
- log.error("Mapping property type: " +
this.params.getMappingPropertyType() + " is not supported");
+ LOG.error("Mapping property type: " + this.params.mappingPropertyType()
+ " is not supported");
} else {
DataValue value = new DataValue(v);
@@ -154,23 +150,22 @@ public class OpcUa implements EventSink<OpcUaParameters> {
StatusCode status = f.get();
if (status.isBad()) {
if (status.getValue() == 0x80740000L) {
- log.error("Type missmatch! Tried to write value of type: " +
this.params.getMappingPropertyType()
+ LOG.error("Type missmatch! Tried to write value of type: " +
this.params.mappingPropertyType()
+ " but server did not accept this");
} else if (status.getValue() == 0x803B0000L) {
- log.error("Wrong access level. Not allowed to write to nodes");
+ LOG.error("Wrong access level. Not allowed to write to nodes");
}
- log.error(
+ LOG.error(
"Value: " + value.getValue().toString() + " could not be written
to node Id: "
- + this.params.getNodeId() + " on " + "OPC-UA server: " +
this.serverUrl);
+ + this.params.nodeId() + " on " + "OPC-UA server: " +
this.serverUrl);
}
} catch (InterruptedException | ExecutionException e) {
- log.error("Exception: Value: " + value.getValue().toString() + " could
not be written to node Id: "
- + this.params.getNodeId() + " on " + "OPC-UA server: " +
this.serverUrl);
+ LOG.error("Exception: Value: " + value.getValue().toString() + " could
not be written to node Id: "
+ + this.params.nodeId() + " on " + "OPC-UA server: " +
this.serverUrl);
}
}
}
- @Override
public void onDetach() throws SpRuntimeException {
opcUaClient.disconnect();
}
@@ -178,7 +173,7 @@ public class OpcUa implements EventSink<OpcUaParameters> {
private Variant getValue(Event inputEvent) {
Variant result = null;
PrimitiveField propertyPrimitive =
-
inputEvent.getFieldBySelector(this.params.getMappingPropertySelector()).getAsPrimitive();
+
inputEvent.getFieldBySelector(this.params.mappingPropertySelector()).getAsPrimitive();
if (targetDataType.equals(Integer.class)) {
result = new Variant(propertyPrimitive.getAsInt());
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUaParameters.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUaParameters.java
index 639aa2ba4..ce831afb2 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUaParameters.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUaParameters.java
@@ -18,21 +18,16 @@
package org.apache.streampipes.sinks.databases.jvm.opcua;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class OpcUaParameters extends EventSinkBindingParams {
-
- private String hostname;
- private Integer port;
- private String nodeId;
- private Integer nameSpaceIndex;
- private String mappingPropertySelector;
- private String mappingPropertyType;
-
- public OpcUaParameters(DataSinkInvocation graph, String hostname, Integer
port, String nodeId, Integer nameSpaceIndex,
+public final class OpcUaParameters {
+ private final String hostname;
+ private final Integer port;
+ private final String nodeId;
+ private final Integer nameSpaceIndex;
+ private final String mappingPropertySelector;
+ private final String mappingPropertyType;
+
+ public OpcUaParameters(String hostname, Integer port, String nodeId, Integer
nameSpaceIndex,
String mappingPropertySelector, String
mappingPropertyType) {
- super(graph);
this.hostname = hostname;
this.port = port;
this.nodeId = nodeId;
@@ -41,59 +36,27 @@ public class OpcUaParameters extends EventSinkBindingParams
{
this.mappingPropertyType = mappingPropertyType;
}
- public String getHostName() {
+ public String hostname() {
return hostname;
}
- public void setHostName(String hostname) {
- this.hostname = hostname;
- }
-
- public Integer getPort() {
+ public Integer port() {
return port;
}
- public void setPort(Integer port) {
- this.port = port;
- }
-
- public String getNodeId() {
+ public String nodeId() {
return nodeId;
}
- public void setNodeId(String nodeId) {
- this.nodeId = nodeId;
- }
-
- public Integer getNameSpaceIndex() {
+ public Integer nameSpaceIndex() {
return nameSpaceIndex;
}
- public void setNameSpaceIndex(Integer nameSpaceIndex) {
- this.nameSpaceIndex = nameSpaceIndex;
- }
-
- public String getHostname() {
- return hostname;
- }
-
- public void setHostname(String hostname) {
- this.hostname = hostname;
- }
-
- public String getMappingPropertySelector() {
+ public String mappingPropertySelector() {
return mappingPropertySelector;
}
- public void setMappingPropertySelector(String mappingPropertySelector) {
- this.mappingPropertySelector = mappingPropertySelector;
- }
-
- public String getMappingPropertyType() {
+ public String mappingPropertyType() {
return mappingPropertyType;
}
-
- public void setMappingPropertyType(String mappingPropertyType) {
- this.mappingPropertyType = mappingPropertyType;
- }
}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/UpcUaController.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUaSink.java
similarity index 79%
rename from
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/UpcUaController.java
rename to
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUaSink.java
index 336bf66b9..a66d89f97 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/UpcUaController.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUaSink.java
@@ -21,19 +21,19 @@ package org.apache.streampipes.sinks.databases.jvm.opcua;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.model.DataSinkType;
import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.schema.PropertyScope;
import org.apache.streampipes.sdk.builder.DataSinkBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
import org.apache.streampipes.sdk.helpers.EpRequirements;
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.wrapper.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
+import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
-public class UpcUaController extends
StandaloneEventSinkDeclarer<OpcUaParameters> {
+public class OpcUaSink extends StreamPipesDataSink {
private static final String OPC_SERVER_KEY = "opc_host";
private static final String OPC_PORT_KEY = "opc_port";
@@ -41,6 +41,7 @@ public class UpcUaController extends
StandaloneEventSinkDeclarer<OpcUaParameters
private static final String OPC_NODE_ID_KEY = "opc_node_id_index";
private static final String MAPPING_PROPERTY_KEY = "mapping_property_key";
+ private OpcUa opcUa;
@Override
public DataSinkDescription declareModel() {
@@ -60,11 +61,10 @@ public class UpcUaController extends
StandaloneEventSinkDeclarer<OpcUaParameters
.build();
}
-
@Override
- public ConfiguredEventSink<OpcUaParameters> onInvocation(DataSinkInvocation
graph,
-
DataSinkParameterExtractor extractor) {
-
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
String hostname = extractor.singleValueParameter(OPC_SERVER_KEY,
String.class);
Integer port = extractor.singleValueParameter(OPC_PORT_KEY, Integer.class);
@@ -80,11 +80,20 @@ public class UpcUaController extends
StandaloneEventSinkDeclarer<OpcUaParameters
e.printStackTrace();
}
-
- OpcUaParameters params = new OpcUaParameters(graph, hostname, port,
nodeId, nameSpaceIndex,
+ OpcUaParameters params = new OpcUaParameters(hostname, port, nodeId,
nameSpaceIndex,
mappingPropertySelector, mappingPropertyType);
- return new ConfiguredEventSink<>(params, OpcUa::new);
+ this.opcUa = new OpcUa();
+ this.opcUa.onInvocation(params);
+ }
+
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
+ this.opcUa.onEvent(event);
}
+ @Override
+ public void onDetach() throws SpRuntimeException {
+ this.opcUa.onDetach();
+ }
}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
index bef2f4ec5..268342d6d 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
@@ -23,17 +23,14 @@ import org.apache.streampipes.logging.api.Logger;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.sinks.databases.jvm.jdbcclient.JdbcClient;
import
org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.SupportedDbEngines;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
-public class PostgreSql extends JdbcClient implements
EventSink<PostgreSqlParameters> {
+public class PostgreSql extends JdbcClient {
private PostgreSqlParameters params;
private Logger log;
- @Override
- public void onInvocation(PostgreSqlParameters parameters,
EventSinkRuntimeContext runtimeContext)
+ public void onInvocation(PostgreSqlParameters parameters)
throws SpRuntimeException {
this.params = parameters;
@@ -60,7 +57,6 @@ public class PostgreSql extends JdbcClient implements
EventSink<PostgreSqlParame
queryParameter);
}
- @Override
public void onEvent(Event event) {
try {
save(event);
@@ -71,7 +67,6 @@ public class PostgreSql extends JdbcClient implements
EventSink<PostgreSqlParame
}
}
- @Override
public void onDetach() throws SpRuntimeException {
closeAll();
}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParameters.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParameters.java
index 5e1596e9a..d00af3ffd 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParameters.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlParameters.java
@@ -23,9 +23,6 @@ import
org.apache.streampipes.sinks.databases.jvm.jdbcclient.model.JdbcConnectio
public class PostgreSqlParameters extends JdbcConnectionParameters {
-
- private boolean sslEnabled;
-
public PostgreSqlParameters(DataSinkInvocation graph, String postgreSqlHost,
Integer postgreSqlPort,
String databaseName, String tableName, String
user, String password, Boolean sslEnabled) {
super(
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlController.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSink.java
similarity index 58%
rename from
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlController.java
rename to
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSink.java
index 9378c1874..6638a7f99 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlController.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSqlSink.java
@@ -18,22 +18,23 @@
package org.apache.streampipes.sinks.databases.jvm.postgresql;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.model.DataSinkType;
import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.sdk.builder.DataSinkBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
import org.apache.streampipes.sdk.helpers.EpRequirements;
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.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
+import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
-public class PostgreSqlController extends
StandaloneEventSinkDeclarer<PostgreSqlParameters> {
+public class PostgreSqlSink extends StreamPipesDataSink {
private static final String DATABASE_HOST_KEY = "db_host";
private static final String DATABASE_PORT_KEY = "db_port";
@@ -45,32 +46,34 @@ public class PostgreSqlController extends
StandaloneEventSinkDeclarer<PostgreSql
private static final String SSL_ENABLED = "ssl_enabled";
private static final String SSL_DISABLED = "ssl_disabled";
+ private PostgreSql postgreSql;
+
@Override
public DataSinkDescription declareModel() {
return
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.postgresql")
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .category(DataSinkType.DATABASE)
- .requiredStream(StreamRequirementsBuilder.create()
- .requiredProperty(EpRequirements.anyProperty())
- .build())
- .requiredTextParameter(Labels.withId(DATABASE_HOST_KEY))
- .requiredIntegerParameter(Labels.withId(DATABASE_PORT_KEY), 5432)
- .requiredTextParameter(Labels.withId(DATABASE_NAME_KEY))
- .requiredTextParameter(Labels.withId(DATABASE_TABLE_KEY))
- .requiredTextParameter(Labels.withId(DATABASE_USER_KEY))
- .requiredSecret(Labels.withId(DATABASE_PASSWORD_KEY))
- .requiredSingleValueSelection(Labels.withId(SSL_MODE),
- Options.from(
- new Tuple2<>("Yes", SSL_ENABLED),
- new Tuple2<>("No", SSL_DISABLED)))
- .build();
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .category(DataSinkType.DATABASE)
+ .requiredStream(StreamRequirementsBuilder.create()
+ .requiredProperty(EpRequirements.anyProperty())
+ .build())
+ .requiredTextParameter(Labels.withId(DATABASE_HOST_KEY))
+ .requiredIntegerParameter(Labels.withId(DATABASE_PORT_KEY), 5432)
+ .requiredTextParameter(Labels.withId(DATABASE_NAME_KEY))
+ .requiredTextParameter(Labels.withId(DATABASE_TABLE_KEY))
+ .requiredTextParameter(Labels.withId(DATABASE_USER_KEY))
+ .requiredSecret(Labels.withId(DATABASE_PASSWORD_KEY))
+ .requiredSingleValueSelection(Labels.withId(SSL_MODE),
+ Options.from(
+ new Tuple2<>("Yes", SSL_ENABLED),
+ new Tuple2<>("No", SSL_DISABLED)))
+ .build();
}
@Override
- public ConfiguredEventSink<PostgreSqlParameters>
onInvocation(DataSinkInvocation graph,
-
DataSinkParameterExtractor extractor) {
-
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
String hostname = extractor.singleValueParameter(DATABASE_HOST_KEY,
String.class);
Integer port = extractor.singleValueParameter(DATABASE_PORT_KEY,
Integer.class);
String dbName = extractor.singleValueParameter(DATABASE_NAME_KEY,
String.class);
@@ -79,15 +82,26 @@ public class PostgreSqlController extends
StandaloneEventSinkDeclarer<PostgreSql
String password = extractor.secretValue(DATABASE_PASSWORD_KEY);
String sslSelection = extractor.selectedSingleValueInternalName(SSL_MODE,
String.class);
- PostgreSqlParameters params = new PostgreSqlParameters(graph,
- hostname,
- port,
- dbName,
- tableName,
- user,
- password,
- sslSelection.equals(SSL_ENABLED));
+ PostgreSqlParameters params = new
PostgreSqlParameters(parameters.getGraph(),
+ hostname,
+ port,
+ dbName,
+ tableName,
+ user,
+ password,
+ sslSelection.equals(SSL_ENABLED));
+
+ this.postgreSql = new PostgreSql();
+ postgreSql.onInvocation(params);
+ }
+
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
+ postgreSql.onEvent(event);
+ }
- return new ConfiguredEventSink<>(params, PostgreSql::new);
+ @Override
+ public void onDetach() throws SpRuntimeException {
+ postgreSql.onDetach();
}
}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java
deleted file mode 100644
index 1bf309252..000000000
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/iotdb/TestIotDbParameters.java
+++ /dev/null
@@ -1,52 +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.sinks.databases.jvm.iotdb;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-import static org.mockito.Mockito.mock;
-
-public class TestIotDbParameters {
-
- @Test
- public void testIotDbParameters() {
- final String host = "localhost";
- final Integer port = 6667;
- final String user = "user";
- final String password = "password";
- final String database = "database";
- final String device = "device";
- final String timestampField = "timestampField";
-
- final IotDbParameters parameters =
- new IotDbParameters(mock(DataSinkInvocation.class), host, port, user,
password, database, device,
- timestampField);
-
- Assert.assertEquals(host, parameters.getHost());
- Assert.assertEquals(port, parameters.getPort());
- Assert.assertEquals(user, parameters.getUser());
- Assert.assertEquals(password, parameters.getPassword());
- Assert.assertEquals(database, parameters.getDatabase());
- Assert.assertEquals(device, parameters.getDevice());
- Assert.assertEquals(timestampField, parameters.getTimestampField());
- }
-}
\ No newline at end of file
