This is an automated email from the ASF dual-hosted git repository.
riemer 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 2e29e3770 Migrate Redis Sink in databases-jvm (#1622)
2e29e3770 is described below
commit 2e29e3770c9cb7eb011ff97eda1e03e3cce4d2a6
Author: Liu Xiao <[email protected]>
AuthorDate: Sat May 27 18:18:57 2023 +0800
Migrate Redis Sink in databases-jvm (#1622)
* Migrate Redis Sink in databases-jvm
* remove blank lines
---
.../sinks/databases/jvm/DatabasesJvmInit.java | 4 +-
.../sinks/databases/jvm/redis/Redis.java | 264 +++++++++++----------
.../sinks/databases/jvm/redis/RedisParameters.java | 234 +++++++++---------
.../redis/{RedisController.java => RedisSink.java} | 232 ++++++++++--------
4 files changed, 384 insertions(+), 350 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 313551c26..2c8cddb88 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
@@ -33,7 +33,7 @@ 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;
+import org.apache.streampipes.sinks.databases.jvm.redis.RedisSink;
public class DatabasesJvmInit extends ExtensionsModelSubmitter {
@@ -53,7 +53,7 @@ public class DatabasesJvmInit extends
ExtensionsModelSubmitter {
new PostgreSqlSink(),
new IotDbSink(),
new DittoSink(),
- new RedisController())
+ new RedisSink())
.registerMessagingFormats(
new JsonDataFormatFactory(),
new CborDataFormatFactory(),
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/Redis.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/Redis.java
index c64e4b434..ff4b8ce52 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/Redis.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/Redis.java
@@ -1,131 +1,133 @@
-/*
- * 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.redis;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.commons.lang3.StringUtils;
-import redis.clients.jedis.Jedis;
-import redis.clients.jedis.JedisPool;
-import redis.clients.jedis.JedisPoolConfig;
-import redis.clients.jedis.exceptions.JedisException;
-
-public class Redis implements EventSink<RedisParameters> {
-
- private static final String EVENT_PREFIX = "sp:event:";
- private static final String EVENT_COUNT = "sp:events";
- private static JedisPool jedisPool = null;
- private String primaryKey;
- private Boolean autoIncrement;
- private String password;
- private String clientName;
- private Integer index;
- private Integer ttl;
-
- @Override
- public void onInvocation(RedisParameters parameters, EventSinkRuntimeContext
runtimeContext) {
- if (jedisPool == null) {
- initialPool(parameters);
- }
- primaryKey = parameters.getPrimaryKey();
- autoIncrement = parameters.isAutoIncrement();
- password = parameters.getRedisPassword();
- clientName = parameters.getRedisClient();
- index = parameters.getRedisIndex();
- ttl = parameters.getTTL();
- }
-
- @Override
- public void onEvent(Event inputEvent) throws SpRuntimeException {
- try (Jedis jedis = getJedis()) {
- final String eventKey = getEventKey(inputEvent, autoIncrement ?
jedis.incr(EVENT_COUNT) : 0L);
- jedis.set(eventKey, getEventValue(inputEvent));
- if (ttl > -1) {
- jedis.expire(eventKey, ttl);
- }
- } catch (SpRuntimeException e) {
- throw e;
- } catch (Exception ex) {
- throw new SpRuntimeException("Could not persist event to redis", ex);
- }
- }
-
- @Override
- public void onDetach() {
- if (jedisPool != null && !jedisPool.isClosed()) {
- jedisPool.close();
- }
- jedisPool = null;
- }
-
- private void initialPool(RedisParameters parameters) {
- JedisPoolConfig config = new JedisPoolConfig();
- config.setMaxTotal(parameters.getRedisPoolMaxActive());
- config.setMaxIdle(parameters.getRedisPoolMaxIdle());
- config.setMaxWaitMillis(parameters.getRedisPoolMaxWait());
- config.setTestWhileIdle(false);
- config.setTestOnBorrow(false);
- config.setTestOnReturn(false);
- jedisPool = new JedisPool(config, parameters.getRedisHost(),
- parameters.getRedisPort(), parameters.getRedisPoolTimeout());
- }
-
- private Jedis getJedis() throws SpRuntimeException {
- Jedis jedis = jedisPool.getResource();
- try {
- jedis.connect();
- if (StringUtils.isNotBlank(password)) {
- jedis.auth(password);
- }
- if (StringUtils.isNotBlank(clientName)) {
- jedis.clientSetname(clientName);
- }
- if (index > -1) {
- jedis.select(index);
- }
- } catch (JedisException e) {
- jedis.close();
- throw new SpRuntimeException("Could not connect to redis", e);
- }
- return jedis;
- }
-
- private String getEventKey(Event event, Long count) {
- if (autoIncrement) {
- return EVENT_PREFIX + count;
- } else {
- String value =
event.getFieldBySelector(primaryKey).getAsPrimitive().getAsString();
- return EVENT_PREFIX + value;
- }
- }
-
- private String getEventValue(Event event) throws SpRuntimeException {
- try {
- return new ObjectMapper().writeValueAsString(event.getRaw());
- } catch (JsonProcessingException e) {
- throw new SpRuntimeException("Could not convert event to JSON", e);
- }
- }
-
-}
+/*
+ * 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.redis;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.runtime.Event;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.exceptions.JedisException;
+
+public class Redis {
+
+ private static final String EVENT_PREFIX = "sp:event:";
+
+ private static final String EVENT_COUNT = "sp:events";
+
+ private static JedisPool jedisPool = null;
+
+ private String primaryKey;
+
+ private Boolean autoIncrement;
+
+ private String password;
+
+ private String clientName;
+
+ private Integer index;
+
+ private Integer ttl;
+
+ public void onInvocation(RedisParameters parameters) {
+ if (jedisPool == null) {
+ initialPool(parameters);
+ }
+ primaryKey = parameters.getPrimaryKey();
+ autoIncrement = parameters.isAutoIncrement();
+ password = parameters.getRedisPassword();
+ clientName = parameters.getRedisClient();
+ index = parameters.getRedisIndex();
+ ttl = parameters.getTTL();
+ }
+
+ public void onEvent(Event event) throws SpRuntimeException {
+ try (Jedis jedis = getJedis()) {
+ final String eventKey = getEventKey(event, autoIncrement ?
jedis.incr(EVENT_COUNT) : 0L);
+ jedis.set(eventKey, getEventValue(event));
+ if (ttl > -1) {
+ jedis.expire(eventKey, ttl);
+ }
+ } catch (SpRuntimeException e) {
+ throw e;
+ } catch (Exception ex) {
+ throw new SpRuntimeException("Could not persist event to redis", ex);
+ }
+ }
+
+ public void onDetach() throws SpRuntimeException {
+ if (jedisPool != null && !jedisPool.isClosed()) {
+ jedisPool.close();
+ }
+ jedisPool = null;
+ }
+
+ private void initialPool(RedisParameters parameters) {
+ JedisPoolConfig config = new JedisPoolConfig();
+ config.setMaxTotal(parameters.getRedisPoolMaxActive());
+ config.setMaxIdle(parameters.getRedisPoolMaxIdle());
+ config.setMaxWaitMillis(parameters.getRedisPoolMaxWait());
+ config.setTestWhileIdle(false);
+ config.setTestOnBorrow(false);
+ config.setTestOnReturn(false);
+ jedisPool = new JedisPool(config, parameters.getRedisHost(),
+ parameters.getRedisPort(), parameters.getRedisPoolTimeout());
+ }
+
+ private Jedis getJedis() throws SpRuntimeException {
+ Jedis jedis = jedisPool.getResource();
+ try {
+ jedis.connect();
+ if (StringUtils.isNotBlank(password)) {
+ jedis.auth(password);
+ }
+ if (StringUtils.isNotBlank(clientName)) {
+ jedis.clientSetname(clientName);
+ }
+ if (index > -1) {
+ jedis.select(index);
+ }
+ } catch (JedisException e) {
+ jedis.close();
+ throw new SpRuntimeException("Could not connect to redis", e);
+ }
+ return jedis;
+ }
+
+ private String getEventKey(Event event, Long count) {
+ if (autoIncrement) {
+ return EVENT_PREFIX + count;
+ } else {
+ String value =
event.getFieldBySelector(primaryKey).getAsPrimitive().getAsString();
+ return EVENT_PREFIX + value;
+ }
+ }
+
+ private String getEventValue(Event event) throws SpRuntimeException {
+ try {
+ return new ObjectMapper().writeValueAsString(event.getRaw());
+ } catch (JsonProcessingException e) {
+ throw new SpRuntimeException("Could not convert event to JSON", e);
+ }
+ }
+}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisParameters.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisParameters.java
index 0279afc28..09d2c1659 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisParameters.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisParameters.java
@@ -1,114 +1,120 @@
-/*
- * 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.redis;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class RedisParameters extends EventSinkBindingParams {
-
- private String primaryKey;
- private Boolean autoIncrement;
- private Integer ttl;
- private String redisHost;
- private Integer redisPort;
- private String redisPassword;
- private String redisClient;
- private Integer redisIndex;
- private Integer redisPoolMaxActive;
- private Integer redisPoolMaxIdle;
- private Integer redisPoolMaxWait;
- private Integer redisPoolTimeout;
-
- public RedisParameters(DataSinkInvocation graph,
- String primaryKey,
- Boolean autoIncrement,
- Integer ttl,
- String redisHost,
- Integer redisPort,
- String redisPassword,
- String redisClient,
- Integer redisIndex,
- Integer redisPoolMaxActive,
- Integer redisPoolMaxIdle,
- Integer redisPoolMaxWait,
- Integer redisPoolTimeout) {
- super(graph);
- this.primaryKey = primaryKey;
- this.autoIncrement = autoIncrement;
- this.ttl = ttl;
- this.redisHost = redisHost;
- this.redisPort = redisPort;
- this.redisPassword = redisPassword;
- this.redisClient = redisClient;
- this.redisIndex = redisIndex;
- this.redisPoolMaxActive = redisPoolMaxActive;
- this.redisPoolMaxIdle = redisPoolMaxIdle;
- this.redisPoolMaxWait = redisPoolMaxWait;
- this.redisPoolTimeout = redisPoolTimeout;
- }
-
- public String getPrimaryKey() {
- return primaryKey;
- }
-
- public Boolean isAutoIncrement() {
- return autoIncrement;
- }
-
- public Integer getTTL() {
- return ttl;
- }
-
- public String getRedisHost() {
- return redisHost;
- }
-
- public Integer getRedisPort() {
- return redisPort;
- }
-
- public String getRedisPassword() {
- return redisPassword;
- }
-
- public String getRedisClient() {
- return redisClient;
- }
-
- public Integer getRedisIndex() {
- return redisIndex;
- }
-
- public Integer getRedisPoolMaxActive() {
- return redisPoolMaxActive;
- }
-
- public Integer getRedisPoolMaxIdle() {
- return redisPoolMaxIdle;
- }
-
- public Integer getRedisPoolMaxWait() {
- return redisPoolMaxWait;
- }
-
- public Integer getRedisPoolTimeout() {
- return redisPoolTimeout;
- }
-}
+/*
+ * 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.redis;
+
+public class RedisParameters {
+
+ private final String primaryKey;
+
+ private final Boolean autoIncrement;
+
+ private final Integer ttl;
+
+ private final String redisHost;
+
+ private final Integer redisPort;
+
+ private final String redisPassword;
+
+ private final String redisClient;
+
+ private final Integer redisIndex;
+
+ private final Integer redisPoolMaxActive;
+
+ private final Integer redisPoolMaxIdle;
+
+ private final Integer redisPoolMaxWait;
+
+ private final Integer redisPoolTimeout;
+
+ public RedisParameters(String primaryKey,
+ Boolean autoIncrement,
+ Integer ttl,
+ String redisHost,
+ Integer redisPort,
+ String redisPassword,
+ String redisClient,
+ Integer redisIndex,
+ Integer redisPoolMaxActive,
+ Integer redisPoolMaxIdle,
+ Integer redisPoolMaxWait,
+ Integer redisPoolTimeout) {
+ this.primaryKey = primaryKey;
+ this.autoIncrement = autoIncrement;
+ this.ttl = ttl;
+ this.redisHost = redisHost;
+ this.redisPort = redisPort;
+ this.redisPassword = redisPassword;
+ this.redisClient = redisClient;
+ this.redisIndex = redisIndex;
+ this.redisPoolMaxActive = redisPoolMaxActive;
+ this.redisPoolMaxIdle = redisPoolMaxIdle;
+ this.redisPoolMaxWait = redisPoolMaxWait;
+ this.redisPoolTimeout = redisPoolTimeout;
+ }
+
+ public String getPrimaryKey() {
+ return primaryKey;
+ }
+
+ public Boolean isAutoIncrement() {
+ return autoIncrement;
+ }
+
+ public Integer getTTL() {
+ return ttl;
+ }
+
+ public String getRedisHost() {
+ return redisHost;
+ }
+
+ public Integer getRedisPort() {
+ return redisPort;
+ }
+
+ public String getRedisPassword() {
+ return redisPassword;
+ }
+
+ public String getRedisClient() {
+ return redisClient;
+ }
+
+ public Integer getRedisIndex() {
+ return redisIndex;
+ }
+
+ public Integer getRedisPoolMaxActive() {
+ return redisPoolMaxActive;
+ }
+
+ public Integer getRedisPoolMaxIdle() {
+ return redisPoolMaxIdle;
+ }
+
+ public Integer getRedisPoolMaxWait() {
+ return redisPoolMaxWait;
+ }
+
+ public Integer getRedisPoolTimeout() {
+ return redisPoolTimeout;
+ }
+}
diff --git
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisController.java
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisSink.java
similarity index 82%
rename from
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisController.java
rename to
streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisSink.java
index 95c9a9221..e68a5c525 100644
---
a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisController.java
+++
b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/redis/RedisSink.java
@@ -1,103 +1,129 @@
-/*
- * 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.redis;
-
-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.helpers.Options;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
-
-public class RedisController extends
StandaloneEventSinkDeclarer<RedisParameters> {
- private static final String EVENT_PRIMARY_KEY = "event_pk";
- private static final String EVENT_KEY_AUTO_INCREMENT = "event_key_ai";
- private static final String EVENT_TTL_KEY = "event_ttl";
- private static final String REDIS_HOST_KEY = "redis_host";
- private static final String REDIS_PORT_KEY = "redis_port";
- private static final String REDIS_PASSWORD_KEY = "redis_pw";
- private static final String REDIS_CLIENT_KEY = "redis_client";
- private static final String REDIS_INDEX_KEY = "redis_index";
- private static final String REDIS_POOL_MAX_ACTIVE_KEY =
"redis_pool_max_active";
- private static final String REDIS_POOL_MAX_IDLE_KEY = "redis_pool_max_idle";
- private static final String REDIS_POOL_MAX_WAIT_KEY = "redis_pool_max_wait";
- private static final String REDIS_POOL_TIMEOUT_KEY = "redis_pool_timeout";
-
- @Override
- public DataSinkDescription declareModel() {
- return
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.redis")
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .category(DataSinkType.DATABASE)
- .requiredStream(StreamRequirementsBuilder
- .create()
- .requiredPropertyWithUnaryMapping(EpRequirements.anyProperty(),
- Labels.withId(EVENT_PRIMARY_KEY),
- PropertyScope.NONE).build())
- .requiredSingleValueSelection(Labels.withId(EVENT_KEY_AUTO_INCREMENT),
Options.from("False", "True"))
- .requiredTextParameter(Labels.withId(REDIS_HOST_KEY))
- .requiredIntegerParameter(Labels.withId(REDIS_PORT_KEY), 6379)
- .requiredIntegerParameter(Labels.withId(EVENT_TTL_KEY), -1)
- // TODO: Use this after optional parameters implementation
- // .requiredSecret(Labels.withId(REDIS_PASSWORD_KEY))
- // .requiredTextParameter(Labels.withId(REDIS_CLIENT_KEY))
- .requiredIntegerParameter(Labels.withId(REDIS_INDEX_KEY), -1)
- .requiredIntegerParameter(Labels.withId(REDIS_POOL_MAX_ACTIVE_KEY), 8)
- .requiredIntegerParameter(Labels.withId(REDIS_POOL_MAX_IDLE_KEY), 8)
- .requiredIntegerParameter(Labels.withId(REDIS_POOL_MAX_WAIT_KEY), -1)
- .requiredIntegerParameter(Labels.withId(REDIS_POOL_TIMEOUT_KEY), 2000)
- .build();
- }
-
-
- @Override
- public ConfiguredEventSink<RedisParameters> onInvocation(DataSinkInvocation
graph,
-
DataSinkParameterExtractor extractor) {
- String redisHost = extractor.singleValueParameter(REDIS_HOST_KEY,
String.class);
- Integer redisPort = extractor.singleValueParameter(REDIS_PORT_KEY,
Integer.class);
- String primaryKey = extractor.mappingPropertyValue(EVENT_PRIMARY_KEY);
- Boolean autoIncrement =
Boolean.valueOf(extractor.selectedSingleValue(EVENT_KEY_AUTO_INCREMENT,
String.class));
- Integer ttl = extractor.singleValueParameter(EVENT_TTL_KEY, Integer.class);
- // TODO: Use this after optional parameters implementation
- // String redisPassword = extractor.secretValue(REDIS_PASSWORD_KEY);
- // String redisClient =
extractor.singleValueParameter(REDIS_CLIENT_KEY, String.class);
- Integer redisIndex = extractor.singleValueParameter(REDIS_INDEX_KEY,
Integer.class);
- Integer redisPoolMaxActive =
extractor.singleValueParameter(REDIS_POOL_MAX_ACTIVE_KEY, Integer.class);
- Integer redisPoolMaxIdle =
extractor.singleValueParameter(REDIS_POOL_MAX_IDLE_KEY, Integer.class);
- Integer redisPoolMaxWait =
extractor.singleValueParameter(REDIS_POOL_MAX_WAIT_KEY, Integer.class);
- Integer redisPoolTimeout =
extractor.singleValueParameter(REDIS_POOL_TIMEOUT_KEY, Integer.class);
- String redisPassword = "";
- String redisClient = "";
-
- RedisParameters params = new RedisParameters(graph, primaryKey,
autoIncrement, ttl, redisHost,
- redisPort, redisPassword, redisClient, redisIndex, redisPoolMaxActive,
redisPoolMaxIdle,
- redisPoolMaxWait, redisPoolTimeout);
-
- return new ConfiguredEventSink<>(params, Redis::new);
- }
-
-}
+/*
+ * 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.redis;
+
+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.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.utils.Assets;
+import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
+
+public class RedisSink extends StreamPipesDataSink {
+
+ private static final String EVENT_PRIMARY_KEY = "event_pk";
+
+ private static final String EVENT_KEY_AUTO_INCREMENT = "event_key_ai";
+
+ private static final String EVENT_TTL_KEY = "event_ttl";
+
+ private static final String REDIS_HOST_KEY = "redis_host";
+
+ private static final String REDIS_PORT_KEY = "redis_port";
+
+ private static final String REDIS_PASSWORD_KEY = "redis_pw";
+
+ private static final String REDIS_CLIENT_KEY = "redis_client";
+
+ private static final String REDIS_INDEX_KEY = "redis_index";
+
+ private static final String REDIS_POOL_MAX_ACTIVE_KEY =
"redis_pool_max_active";
+
+ private static final String REDIS_POOL_MAX_IDLE_KEY = "redis_pool_max_idle";
+
+ private static final String REDIS_POOL_MAX_WAIT_KEY = "redis_pool_max_wait";
+
+ private static final String REDIS_POOL_TIMEOUT_KEY = "redis_pool_timeout";
+
+ private Redis redis;
+
+ @Override
+ public DataSinkDescription declareModel() {
+ return
DataSinkBuilder.create("org.apache.streampipes.sinks.databases.jvm.redis")
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .category(DataSinkType.DATABASE)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+ .requiredPropertyWithUnaryMapping(EpRequirements.anyProperty(),
+ Labels.withId(EVENT_PRIMARY_KEY),
+ PropertyScope.NONE).build())
+ .requiredSingleValueSelection(Labels.withId(EVENT_KEY_AUTO_INCREMENT),
Options.from("False", "True"))
+ .requiredTextParameter(Labels.withId(REDIS_HOST_KEY))
+ .requiredIntegerParameter(Labels.withId(REDIS_PORT_KEY), 6379)
+ .requiredIntegerParameter(Labels.withId(EVENT_TTL_KEY), -1)
+ // TODO: Use this after optional parameters implementation
+ // .requiredSecret(Labels.withId(REDIS_PASSWORD_KEY))
+ // .requiredTextParameter(Labels.withId(REDIS_CLIENT_KEY))
+ .requiredIntegerParameter(Labels.withId(REDIS_INDEX_KEY), -1)
+ .requiredIntegerParameter(Labels.withId(REDIS_POOL_MAX_ACTIVE_KEY), 8)
+ .requiredIntegerParameter(Labels.withId(REDIS_POOL_MAX_IDLE_KEY), 8)
+ .requiredIntegerParameter(Labels.withId(REDIS_POOL_MAX_WAIT_KEY), -1)
+ .requiredIntegerParameter(Labels.withId(REDIS_POOL_TIMEOUT_KEY), 2000)
+ .build();
+ }
+
+ @Override
+ public void onInvocation(SinkParams parameters, EventSinkRuntimeContext
runtimeContext) throws SpRuntimeException {
+ DataSinkParameterExtractor extractor = parameters.extractor();
+ String redisHost = extractor.singleValueParameter(REDIS_HOST_KEY,
String.class);
+ Integer redisPort = extractor.singleValueParameter(REDIS_PORT_KEY,
Integer.class);
+ String primaryKey = extractor.mappingPropertyValue(EVENT_PRIMARY_KEY);
+ Boolean autoIncrement =
Boolean.valueOf(extractor.selectedSingleValue(EVENT_KEY_AUTO_INCREMENT,
String.class));
+ Integer ttl = extractor.singleValueParameter(EVENT_TTL_KEY, Integer.class);
+ // TODO: Use this after optional parameters implementation
+ // String redisPassword = extractor.secretValue(REDIS_PASSWORD_KEY);
+ // String redisClient =
extractor.singleValueParameter(REDIS_CLIENT_KEY, String.class);
+ Integer redisIndex = extractor.singleValueParameter(REDIS_INDEX_KEY,
Integer.class);
+ Integer redisPoolMaxActive =
extractor.singleValueParameter(REDIS_POOL_MAX_ACTIVE_KEY, Integer.class);
+ Integer redisPoolMaxIdle =
extractor.singleValueParameter(REDIS_POOL_MAX_IDLE_KEY, Integer.class);
+ Integer redisPoolMaxWait =
extractor.singleValueParameter(REDIS_POOL_MAX_WAIT_KEY, Integer.class);
+ Integer redisPoolTimeout =
extractor.singleValueParameter(REDIS_POOL_TIMEOUT_KEY, Integer.class);
+ String redisPassword = "";
+ String redisClient = "";
+
+ RedisParameters
+ params = new RedisParameters(primaryKey, autoIncrement, ttl, redisHost,
+ redisPort, redisPassword, redisClient, redisIndex, redisPoolMaxActive,
redisPoolMaxIdle,
+ redisPoolMaxWait, redisPoolTimeout);
+
+ this.redis = new Redis();
+ redis.onInvocation(params);
+ }
+
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
+ redis.onEvent(event);
+ }
+
+ @Override
+ public void onDetach() throws SpRuntimeException {
+ redis.onDetach();
+ }
+}
