You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jg...@apache.org on 2023/10/06 19:34:53 UTC
[nifi] branch support/nifi-1.x updated: NIFI-12149 Create nifi-redis-utils and minor improvements to util methods
This is an automated email from the ASF dual-hosted git repository.
jgresock pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new cde6b0196f NIFI-12149 Create nifi-redis-utils and minor improvements to util methods
cde6b0196f is described below
commit cde6b0196f64927f83229c02c130ca2a101b136f
Author: Bryan Bende <bb...@apache.org>
AuthorDate: Fri Sep 29 10:15:22 2023 -0400
NIFI-12149 Create nifi-redis-utils and minor improvements to util methods
Signed-off-by: Joe Gresock <jg...@gmail.com>
This closes #7816.
---
.../nifi-redis-extensions/pom.xml | 5 +
.../redis/service/RedisConnectionPoolService.java | 2 +-
.../nifi/redis/state/RedisStateProvider.java | 2 +-
.../service/TestRedisConnectionPoolService.java | 2 +-
.../pom.xml | 42 +----
.../org/apache/nifi/redis/util/RedisAction.java | 0
.../org/apache/nifi/redis/util/RedisConfig.java | 206 +++++++++++++++++++++
.../org/apache/nifi/redis/util/RedisUtils.java | 137 ++++++++++----
nifi-nar-bundles/nifi-redis-bundle/pom.xml | 1 +
9 files changed, 327 insertions(+), 70 deletions(-)
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
index b533dff4c5..45844271ea 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
@@ -38,6 +38,11 @@
<version>1.24.0-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-redis-utils</artifactId>
+ <version>1.24.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-redis</artifactId>
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisConnectionPoolService.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisConnectionPoolService.java
index 83937fb5d9..b7e4f3beb2 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisConnectionPoolService.java
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/service/RedisConnectionPoolService.java
@@ -89,7 +89,7 @@ public class RedisConnectionPoolService extends AbstractControllerService implem
if (connectionFactory == null) {
synchronized (this) {
if (connectionFactory == null) {
- connectionFactory = RedisUtils.createConnectionFactory(context, getLogger(), sslContext);
+ connectionFactory = RedisUtils.createConnectionFactory(context, sslContext);
}
}
}
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
index ba71dc488b..3739e23cd6 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java
@@ -304,7 +304,7 @@ public class RedisStateProvider extends AbstractConfigurableComponent implements
// visible for testing
synchronized RedisConnection getRedis() {
if (connectionFactory == null) {
- connectionFactory = RedisUtils.createConnectionFactory(context, logger, sslContext);
+ connectionFactory = RedisUtils.createConnectionFactory(context, sslContext);
}
return connectionFactory.getConnection();
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/TestRedisConnectionPoolService.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/TestRedisConnectionPoolService.java
index b1e6fe761f..e3d7dae4c4 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/TestRedisConnectionPoolService.java
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/TestRedisConnectionPoolService.java
@@ -126,7 +126,7 @@ public class TestRedisConnectionPoolService {
final SSLContextService sslContextService = configContext.getProperty(RedisUtils.SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
providedSslContext = sslContextService.createContext();
}
- JedisConnectionFactory connectionFactory = RedisUtils.createConnectionFactory(configContext, testRunner.getLogger(), providedSslContext);
+ JedisConnectionFactory connectionFactory = RedisUtils.createConnectionFactory(configContext, providedSslContext);
return connectionFactory;
}
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/pom.xml
similarity index 69%
copy from nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
copy to nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/pom.xml
index b533dff4c5..9879ac01d8 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/pom.xml
@@ -22,15 +22,13 @@
<version>1.24.0-SNAPSHOT</version>
</parent>
- <artifactId>nifi-redis-extensions</artifactId>
+ <artifactId>nifi-redis-utils</artifactId>
<packaging>jar</packaging>
<dependencies>
- <!-- Provided deps from nifi-redis-service-api -->
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-distributed-cache-client-service-api</artifactId>
- <scope>provided</scope>
+ <artifactId>nifi-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
@@ -38,54 +36,28 @@
<version>1.24.0-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-ssl-context-service-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-redis</artifactId>
<version>${spring.data.redis.version}</version>
<scope>provided</scope>
</dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-beans</artifactId>
- </dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>${jedis.version}</version>
<scope>provided</scope>
</dependency>
- <!-- End Provided deps from nifi-redis-service-api -->
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-databind</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-api</artifactId>
- </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-utils</artifactId>
<version>1.24.0-SNAPSHOT</version>
</dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-ssl-context-service-api</artifactId>
- <scope>provided</scope>
- </dependency>
- <!-- Test dependencies -->
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-mock</artifactId>
- <version>1.24.0-SNAPSHOT</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers</artifactId>
- <version>${testcontainers.version}</version>
- <scope>test</scope>
- </dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisAction.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/src/main/java/org/apache/nifi/redis/util/RedisAction.java
similarity index 100%
rename from nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisAction.java
rename to nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/src/main/java/org/apache/nifi/redis/util/RedisAction.java
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/src/main/java/org/apache/nifi/redis/util/RedisConfig.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/src/main/java/org/apache/nifi/redis/util/RedisConfig.java
new file mode 100644
index 0000000000..c5da310093
--- /dev/null
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/src/main/java/org/apache/nifi/redis/util/RedisConfig.java
@@ -0,0 +1,206 @@
+/*
+ * 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.nifi.redis.util;
+
+import org.apache.nifi.redis.RedisType;
+
+import java.time.Duration;
+import java.util.Objects;
+
+public class RedisConfig {
+
+ private final RedisType redisMode;
+ private final String connectionString;
+
+ private String sentinelMaster;
+ private String sentinelPassword;
+ private String password;
+
+ private int dbIndex = 0;
+ private int timeout = 10000;
+ private int clusterMaxRedirects = 5;
+
+ private int poolMaxTotal = 8;
+ private int poolMaxIdle = 8;
+ private int poolMinIdle = 0;
+ private boolean blockWhenExhausted = true;
+ private Duration maxWaitTime = Duration.ofSeconds(10);
+ private Duration minEvictableIdleTime = Duration.ofSeconds(60);
+ private Duration timeBetweenEvictionRuns = Duration.ofSeconds(30);
+ private int numTestsPerEvictionRun = -1;
+ private boolean testOnCreate = true;
+ private boolean testOnBorrow = true;
+ private boolean testOnReturn = false;
+ private boolean testWhenIdle = true;
+
+ public RedisConfig(final RedisType redisMode, final String connectionString) {
+ this.redisMode = Objects.requireNonNull(redisMode);
+ this.connectionString = Objects.requireNonNull(connectionString);
+ }
+
+ public RedisType getRedisMode() {
+ return redisMode;
+ }
+
+ public String getConnectionString() {
+ return connectionString;
+ }
+
+ public String getSentinelMaster() {
+ return sentinelMaster;
+ }
+
+ public void setSentinelMaster(String sentinelMaster) {
+ this.sentinelMaster = sentinelMaster;
+ }
+
+ public String getSentinelPassword() {
+ return sentinelPassword;
+ }
+
+ public void setSentinelPassword(String sentinelPassword) {
+ this.sentinelPassword = sentinelPassword;
+ }
+
+ public int getDbIndex() {
+ return dbIndex;
+ }
+
+ public void setDbIndex(int dbIndex) {
+ this.dbIndex = dbIndex;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public int getTimeout() {
+ return timeout;
+ }
+
+ public void setTimeout(int timeout) {
+ this.timeout = timeout;
+ }
+
+ public int getClusterMaxRedirects() {
+ return clusterMaxRedirects;
+ }
+
+ public void setClusterMaxRedirects(int clusterMaxRedirects) {
+ this.clusterMaxRedirects = clusterMaxRedirects;
+ }
+
+ public int getPoolMaxTotal() {
+ return poolMaxTotal;
+ }
+
+ public void setPoolMaxTotal(int poolMaxTotal) {
+ this.poolMaxTotal = poolMaxTotal;
+ }
+
+ public int getPoolMaxIdle() {
+ return poolMaxIdle;
+ }
+
+ public void setPoolMaxIdle(int poolMaxIdle) {
+ this.poolMaxIdle = poolMaxIdle;
+ }
+
+ public int getPoolMinIdle() {
+ return poolMinIdle;
+ }
+
+ public void setPoolMinIdle(int poolMinIdle) {
+ this.poolMinIdle = poolMinIdle;
+ }
+
+ public boolean getBlockWhenExhausted() {
+ return blockWhenExhausted;
+ }
+
+ public void setBlockWhenExhausted(boolean blockWhenExhausted) {
+ this.blockWhenExhausted = blockWhenExhausted;
+ }
+
+ public Duration getMaxWaitTime() {
+ return maxWaitTime;
+ }
+
+ public void setMaxWaitTime(Duration maxWaitTime) {
+ this.maxWaitTime = maxWaitTime;
+ }
+
+ public Duration getMinEvictableIdleTime() {
+ return minEvictableIdleTime;
+ }
+
+ public void setMinEvictableIdleTime(Duration minEvictableIdleTime) {
+ this.minEvictableIdleTime = minEvictableIdleTime;
+ }
+
+ public Duration getTimeBetweenEvictionRuns() {
+ return timeBetweenEvictionRuns;
+ }
+
+ public void setTimeBetweenEvictionRuns(Duration timeBetweenEvictionRuns) {
+ this.timeBetweenEvictionRuns = timeBetweenEvictionRuns;
+ }
+
+ public int getNumTestsPerEvictionRun() {
+ return numTestsPerEvictionRun;
+ }
+
+ public void setNumTestsPerEvictionRun(int numTestsPerEvictionRun) {
+ this.numTestsPerEvictionRun = numTestsPerEvictionRun;
+ }
+
+ public boolean getTestOnCreate() {
+ return testOnCreate;
+ }
+
+ public void setTestOnCreate(boolean testOnCreate) {
+ this.testOnCreate = testOnCreate;
+ }
+
+ public boolean getTestOnBorrow() {
+ return testOnBorrow;
+ }
+
+ public void setTestOnBorrow(boolean testOnBorrow) {
+ this.testOnBorrow = testOnBorrow;
+ }
+
+ public boolean getTestOnReturn() {
+ return testOnReturn;
+ }
+
+ public void setTestOnReturn(boolean testOnReturn) {
+ this.testOnReturn = testOnReturn;
+ }
+
+ public boolean getTestWhenIdle() {
+ return testWhenIdle;
+ }
+
+ public void setTestWhenIdle(boolean testWhenIdle) {
+ this.testWhenIdle = testWhenIdle;
+ }
+}
diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
similarity index 76%
rename from nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
rename to nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
index 6792a612f4..b44d6027d0 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
+++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-utils/src/main/java/org/apache/nifi/redis/util/RedisUtils.java
@@ -22,19 +22,26 @@ import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
-import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.redis.RedisConnectionPool;
import org.apache.nifi.redis.RedisType;
import org.apache.nifi.ssl.RestrictedSSLContextService;
import org.apache.nifi.util.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.RedisConfiguration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisPassword;
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.jedis.JedisClientConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.lang.Nullable;
import redis.clients.jedis.JedisPoolConfig;
@@ -48,6 +55,8 @@ import java.util.concurrent.TimeUnit;
public class RedisUtils {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RedisUtils.class);
+
// These properties are shared among the controller service(s) and processor(s) that use a RedisConnectionPool
public static final PropertyDescriptor REDIS_CONNECTION_POOL = new PropertyDescriptor.Builder()
@@ -301,15 +310,44 @@ public class RedisUtils {
REDIS_CONNECTION_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props);
}
+ public static RedisConfig createRedisConfig(final PropertyContext context) {
+ final RedisType redisType = RedisType.fromDisplayName(context.getProperty(RedisUtils.REDIS_MODE).getValue());
+ final String connectString = context.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
+
+ final RedisConfig redisConfig = new RedisConfig(redisType, connectString);
+ redisConfig.setSentinelMaster(context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue());
+ redisConfig.setDbIndex(context.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger());
+ redisConfig.setPassword(context.getProperty(RedisUtils.PASSWORD).evaluateAttributeExpressions().getValue());
+ redisConfig.setSentinelPassword(context.getProperty(RedisUtils.SENTINEL_PASSWORD).evaluateAttributeExpressions().getValue());
+ redisConfig.setTimeout(context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue());
+ redisConfig.setClusterMaxRedirects(context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger());
+ redisConfig.setPoolMaxTotal(context.getProperty(RedisUtils.POOL_MAX_TOTAL).asInteger());
+ redisConfig.setPoolMaxIdle(context.getProperty(RedisUtils.POOL_MAX_IDLE).asInteger());
+ redisConfig.setPoolMinIdle(context.getProperty(RedisUtils.POOL_MIN_IDLE).asInteger());
+ redisConfig.setBlockWhenExhausted(context.getProperty(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED).asBoolean());
+ redisConfig.setMaxWaitTime(Duration.ofMillis(context.getProperty(RedisUtils.POOL_MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS)));
+ redisConfig.setMinEvictableIdleTime(Duration.ofMillis(context.getProperty(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME).asTimePeriod(TimeUnit.MILLISECONDS)));
+ redisConfig.setTimeBetweenEvictionRuns(Duration.ofMillis(context.getProperty(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS).asTimePeriod(TimeUnit.MILLISECONDS)));
+ redisConfig.setNumTestsPerEvictionRun(context.getProperty(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN).asInteger());
+ redisConfig.setTestOnCreate(context.getProperty(RedisUtils.POOL_TEST_ON_CREATE).asBoolean());
+ redisConfig.setTestOnBorrow(context.getProperty(RedisUtils.POOL_TEST_ON_BORROW).asBoolean());
+ redisConfig.setTestOnReturn(context.getProperty(RedisUtils.POOL_TEST_ON_RETURN).asBoolean());
+ redisConfig.setTestWhenIdle(context.getProperty(RedisUtils.POOL_TEST_WHILE_IDLE).asBoolean());
+ return redisConfig;
+ }
+
+ public static JedisConnectionFactory createConnectionFactory(final PropertyContext context, final SSLContext sslContext) {
+ return createConnectionFactory(createRedisConfig(context), sslContext);
+ }
- public static JedisConnectionFactory createConnectionFactory(final PropertyContext context, final ComponentLog logger, final SSLContext sslContext) {
- final String redisMode = context.getProperty(RedisUtils.REDIS_MODE).getValue();
- final String connectionString = context.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue();
- final Integer dbIndex = context.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger();
- final String password = context.getProperty(RedisUtils.PASSWORD).evaluateAttributeExpressions().getValue();
- final String sentinelPassword = context.getProperty(RedisUtils.SENTINEL_PASSWORD).evaluateAttributeExpressions().getValue();
- final Integer timeout = context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue();
- final JedisPoolConfig poolConfig = createJedisPoolConfig(context);
+ public static JedisConnectionFactory createConnectionFactory(final RedisConfig redisConfig, final SSLContext sslContext) {
+ final RedisType redisMode = redisConfig.getRedisMode();
+ final String connectionString = redisConfig.getConnectionString();
+ final Integer dbIndex = redisConfig.getDbIndex();
+ final String password = redisConfig.getPassword();
+ final String sentinelPassword = redisConfig.getSentinelPassword();
+ final Integer timeout = redisConfig.getTimeout();
+ final JedisPoolConfig poolConfig = createJedisPoolConfig(redisConfig);
JedisClientConfiguration.JedisClientConfigurationBuilder builder = JedisClientConfiguration.builder()
.connectTimeout(Duration.ofMillis(timeout))
@@ -328,8 +366,8 @@ public class RedisUtils {
final JedisClientConfiguration jedisClientConfiguration = builder.build();
JedisConnectionFactory connectionFactory;
- if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) {
- logger.info("Connecting to Redis in standalone mode at " + connectionString);
+ if (RedisType.STANDALONE == redisMode) {
+ LOGGER.info("Connecting to Redis in standalone mode at " + connectionString);
final String[] hostAndPortSplit = connectionString.split("[:]");
final String host = hostAndPortSplit[0].trim();
final Integer port = Integer.parseInt(hostAndPortSplit[1].trim());
@@ -338,32 +376,32 @@ public class RedisUtils {
connectionFactory = new JedisConnectionFactory(redisStandaloneConfiguration, jedisClientConfiguration);
- } else if (RedisUtils.REDIS_MODE_SENTINEL.getValue().equals(redisMode)) {
+ } else if (RedisType.SENTINEL == redisMode) {
final String[] sentinels = connectionString.split("[,]");
- final String sentinelMaster = context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue();
+ final String sentinelMaster = redisConfig.getSentinelMaster();
final RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration(sentinelMaster, new HashSet<>(getTrimmedValues(sentinels)));
enrichRedisConfiguration(sentinelConfiguration, dbIndex, password, sentinelPassword);
- logger.info("Connecting to Redis in sentinel mode...");
- logger.info("Redis master = " + sentinelMaster);
+ LOGGER.info("Connecting to Redis in sentinel mode...");
+ LOGGER.info("Redis master = " + sentinelMaster);
for (final String sentinel : sentinels) {
- logger.info("Redis sentinel at " + sentinel);
+ LOGGER.info("Redis sentinel at " + sentinel);
}
connectionFactory = new JedisConnectionFactory(sentinelConfiguration, jedisClientConfiguration);
} else {
final String[] clusterNodes = connectionString.split("[,]");
- final Integer maxRedirects = context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger();
+ final Integer maxRedirects = redisConfig.getClusterMaxRedirects();
final RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(getTrimmedValues(clusterNodes));
enrichRedisConfiguration(clusterConfiguration, dbIndex, password, sentinelPassword);
clusterConfiguration.setMaxRedirects(maxRedirects);
- logger.info("Connecting to Redis in clustered mode...");
+ LOGGER.info("Connecting to Redis in clustered mode...");
for (final String clusterNode : clusterNodes) {
- logger.info("Redis cluster node at " + clusterNode);
+ LOGGER.info("Redis cluster node at " + clusterNode);
}
connectionFactory = new JedisConnectionFactory(clusterConfiguration, jedisClientConfiguration);
@@ -397,20 +435,20 @@ public class RedisUtils {
}
}
- private static JedisPoolConfig createJedisPoolConfig(final PropertyContext context) {
+ private static JedisPoolConfig createJedisPoolConfig(final RedisConfig redisConfig) {
final JedisPoolConfig poolConfig = new JedisPoolConfig();
- poolConfig.setMaxTotal(context.getProperty(RedisUtils.POOL_MAX_TOTAL).asInteger());
- poolConfig.setMaxIdle(context.getProperty(RedisUtils.POOL_MAX_IDLE).asInteger());
- poolConfig.setMinIdle(context.getProperty(RedisUtils.POOL_MIN_IDLE).asInteger());
- poolConfig.setBlockWhenExhausted(context.getProperty(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED).asBoolean());
- poolConfig.setMaxWaitMillis(context.getProperty(RedisUtils.POOL_MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
- poolConfig.setMinEvictableIdleTimeMillis(context.getProperty(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME).asTimePeriod(TimeUnit.MILLISECONDS));
- poolConfig.setTimeBetweenEvictionRunsMillis(context.getProperty(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS).asTimePeriod(TimeUnit.MILLISECONDS));
- poolConfig.setNumTestsPerEvictionRun(context.getProperty(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN).asInteger());
- poolConfig.setTestOnCreate(context.getProperty(RedisUtils.POOL_TEST_ON_CREATE).asBoolean());
- poolConfig.setTestOnBorrow(context.getProperty(RedisUtils.POOL_TEST_ON_BORROW).asBoolean());
- poolConfig.setTestOnReturn(context.getProperty(RedisUtils.POOL_TEST_ON_RETURN).asBoolean());
- poolConfig.setTestWhileIdle(context.getProperty(RedisUtils.POOL_TEST_WHILE_IDLE).asBoolean());
+ poolConfig.setMaxTotal(redisConfig.getPoolMaxTotal());
+ poolConfig.setMaxIdle(redisConfig.getPoolMaxIdle());
+ poolConfig.setMinIdle(redisConfig.getPoolMinIdle());
+ poolConfig.setBlockWhenExhausted(redisConfig.getBlockWhenExhausted());
+ poolConfig.setMaxWait(redisConfig.getMaxWaitTime());
+ poolConfig.setMinEvictableIdleTime(redisConfig.getMinEvictableIdleTime());
+ poolConfig.setTimeBetweenEvictionRuns(redisConfig.getTimeBetweenEvictionRuns());
+ poolConfig.setNumTestsPerEvictionRun(redisConfig.getNumTestsPerEvictionRun());
+ poolConfig.setTestOnCreate(redisConfig.getTestOnCreate());
+ poolConfig.setTestOnBorrow(redisConfig.getTestOnBorrow());
+ poolConfig.setTestOnReturn(redisConfig.getTestOnReturn());
+ poolConfig.setTestWhileIdle(redisConfig.getTestWhenIdle());
return poolConfig;
}
@@ -482,4 +520,39 @@ public class RedisUtils {
}
}
+ public static RedisTemplate<String, String> createRedisTemplateForStreams(final RedisConnectionFactory connectionFactory) {
+ final StreamMessageListenerContainer.StreamMessageListenerContainerOptionsBuilder<String, MapRecord<String, String, byte[]>> builder =
+ StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
+ .hashValueSerializer(RedisSerializer.byteArray());
+
+ final StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, byte[]>> containerOptions = builder.build();
+ return createRedisTemplateForStreams(connectionFactory, containerOptions);
+ }
+
+ public static RedisTemplate<String, String> createRedisTemplateForStreams(
+ final RedisConnectionFactory connectionFactory,
+ final StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, byte[]>> containerOptions) {
+ final RedisTemplate<String,String> redisTemplate = new RedisTemplate<>();
+ redisTemplate.setKeySerializer(containerOptions.getKeySerializer());
+ redisTemplate.setValueSerializer(containerOptions.getKeySerializer());
+ redisTemplate.setHashKeySerializer(containerOptions.getHashKeySerializer());
+ redisTemplate.setHashValueSerializer(containerOptions.getHashValueSerializer());
+ redisTemplate.setConnectionFactory(connectionFactory);
+ redisTemplate.afterPropertiesSet();
+ return redisTemplate;
+ }
+
+ public static RedisTemplate<String, String> createRedisTemplateForKeyValues(final RedisConnectionFactory connectionFactory) {
+ final RedisTemplate<String, String> redisTemplate = new StringRedisTemplate(connectionFactory);
+ redisTemplate.afterPropertiesSet();
+ return redisTemplate;
+ }
+
+ public static RedisTemplate<String, byte[]> createRedisTemplateForPubSub(final RedisConnectionFactory connectionFactory) {
+ final RedisTemplate<String, byte[]> redisTemplate = new RedisTemplate<>();
+ redisTemplate.setConnectionFactory(connectionFactory);
+ redisTemplate.setValueSerializer(RedisSerializer.byteArray());
+ redisTemplate.afterPropertiesSet();
+ return redisTemplate;
+ }
}
diff --git a/nifi-nar-bundles/nifi-redis-bundle/pom.xml b/nifi-nar-bundles/nifi-redis-bundle/pom.xml
index 64cfc0b7c5..f471907cde 100644
--- a/nifi-nar-bundles/nifi-redis-bundle/pom.xml
+++ b/nifi-nar-bundles/nifi-redis-bundle/pom.xml
@@ -31,6 +31,7 @@
</properties>
<modules>
+ <module>nifi-redis-utils</module>
<module>nifi-redis-service-api</module>
<module>nifi-redis-service-api-nar</module>
<module>nifi-redis-extensions</module>