allow for configuring underlying library

Author: Crim

src/main/java/org/sourcelab/storm/spout/redis/RedisStreamSpout.java

@@ -8,6 +8,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.sourcelab.storm.spout.redis.client.Client;
+import org.sourcelab.storm.spout.redis.client.ClientFactory;
 import org.sourcelab.storm.spout.redis.client.Consumer;
 import org.sourcelab.storm.spout.redis.client.lettuce.LettuceClient;
 import org.sourcelab.storm.spout.redis.funnel.ConsumerFunnel;
@@ -85,7 +86,7 @@ public void open(
 
  // Create consumer and client
  final int taskIndex = topologyContext.getThisTaskIndex();
- final Client client = new LettuceClient(config, taskIndex);
+ final Client client = new ClientFactory().createClient(config, taskIndex);
  final Consumer consumer = new Consumer(config, client, (ConsumerFunnel) funnel);
 
  // Create background consuming thread.

src/main/java/org/sourcelab/storm/spout/redis/RedisStreamSpoutConfig.java

@@ -1,5 +1,6 @@
 package org.sourcelab.storm.spout.redis;
 
+import org.sourcelab.storm.spout.redis.client.ClientType;
 import org.sourcelab.storm.spout.redis.failhandler.NoRetryHandler;
 
 import java.io.Serializable;
@@ -70,6 +71,11 @@ public class RedisStreamSpoutConfig implements Serializable {
  */
  private final boolean metricsEnabled;
 
+ /**
+ * Defines which underlying client library/implementation to use.
+ */
+ private final ClientType clientType;
+
  /**
  * Constructor.
  * Use Builder instance.
@@ -85,7 +91,7 @@ private RedisStreamSpoutConfig(
 
  // Other settings
  final int maxConsumePerRead, final int maxTupleQueueSize, final int maxAckQueueSize, final long consumerDelayMillis,
- final boolean metricsEnabled
+ final boolean metricsEnabled, final ClientType clientType
  ) {
  // Connection
  if (redisCluster != null && redisServer != null) {
@@ -117,6 +123,9 @@ private RedisStreamSpoutConfig(
  this.maxAckQueueSize = maxAckQueueSize;
  this.consumerDelayMillis = consumerDelayMillis;
  this.metricsEnabled = metricsEnabled;
+
+ // Client type implementation
+ this.clientType = Objects.requireNonNull(clientType);
  }
 
  public String getStreamKey() {
@@ -185,6 +194,10 @@ public boolean isMetricsEnabled() {
  return metricsEnabled;
  }
 
+ public ClientType getClientType() {
+ return clientType;
+ }
+
  /**
  * Create a new Builder instance.
  * @return Builder for Configuration instance.
@@ -229,6 +242,12 @@ public static final class Builder {
  private long consumerDelayMillis = 1000L;
  private boolean metricsEnabled = true;
 
+ /**
+ * Underlying library to use.
+ * Defaults to using Lettuce.
+ */
+ private ClientType clientType = ClientType.LETTUCE;
+
  private Builder() {
  }
 
@@ -396,6 +415,19 @@ public Builder withMetricsEnabled(final boolean enabled) {
  return this;
  }
 
+ public Builder withLettuceClientLibrary() {
+ return withClientType(ClientType.LETTUCE);
+ }
+
+ public Builder withJedisClientLibrary() {
+ return withClientType(ClientType.JEDIS);
+ }
+
+ public Builder withClientType(final ClientType clientType) {
+ this.clientType = Objects.requireNonNull(clientType);
+ return this;
+ }
+
  /**
  * Creates new Configuration instance.
  * @return Configuration instance.
@@ -416,7 +448,10 @@ public RedisStreamSpoutConfig build() {
  tupleConverter, failureHandler,
  // Other settings
  maxConsumePerRead, maxTupleQueueSize, maxAckQueueSize, consumerDelayMillis,
- metricsEnabled
+ metricsEnabled,
+
+ // Underlying client type
+ clientType
  );
  }
  }

src/main/java/org/sourcelab/storm/spout/redis/client/ClientFactory.java

@@ -0,0 +1,37 @@
+package org.sourcelab.storm.spout.redis.client;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.sourcelab.storm.spout.redis.RedisStreamSpoutConfig;
+import org.sourcelab.storm.spout.redis.client.jedis.JedisClient;
+import org.sourcelab.storm.spout.redis.client.lettuce.LettuceClient;
+
+import java.util.Objects;
+
+/**
+ * Factory for creating the appropriate Client instance based on config.
+ */
+public class ClientFactory {
+ private static final Logger logger = LoggerFactory.getLogger(ClientFactory.class);
+
+ /**
+ * Create the appropriate client intance based on configuration.
+ * @param config Spout configuration.
+ * @param instanceId Instance id of spout.
+ * @return Client.
+ */
+ public Client createClient(final RedisStreamSpoutConfig config, final int instanceId) {
+ Objects.requireNonNull(config);
+
+ switch (config.getClientType()) {
+ case JEDIS:
+ logger.info("Using Jedis client library.");
+ return new JedisClient(config, instanceId);
+ case LETTUCE:
+ logger.info("Using Lettuce client library.");
+ return new LettuceClient(config, instanceId);
+ default:
+ throw new IllegalStateException("Unknown/Unhandled Client Type");
+ }
+ }
+}

src/main/java/org/sourcelab/storm/spout/redis/client/ClientType.java

@@ -0,0 +1,9 @@
+package org.sourcelab.storm.spout.redis.client;
+
+/**
+ * Defines allowed implementations.
+ */
+public enum ClientType {
+ LETTUCE,
+ JEDIS;
+}

src/main/java/org/sourcelab/storm/spout/redis/client/jedis/JedisClient.java

@@ -45,17 +45,6 @@ public JedisClient(final RedisStreamSpoutConfig config, final int instanceId) {
  );
  }
 
- private static JedisAdapter createAdapter(final RedisStreamSpoutConfig config, final int instanceId) {
- final String connectStr = config.getConnectString().replaceAll("redis://", "");
- if (config.isConnectingToCluster()) {
- logger.info("Connecting to RedisCluster at {}", config.getConnectStringMasked());
- return new JedisClusterAdapter(new JedisCluster(HostAndPort.parseString(connectStr)), config, instanceId);
- } else {
- logger.info("Connecting to RedisCluster at {}", config.getConnectStringMasked());
- return new JedisRedisAdapter(new Jedis(HostAndPort.parseString(connectStr)), config, instanceId);
- }
- }
-
  /**
  * Protected constructor for injecting a RedisClient instance, typically for tests.
  * @param adapter JedisAdapter instance.
@@ -111,4 +100,20 @@ public void commitMessage(final String msgId) {
  public void disconnect() {
  adapter.close();
  }
+
+ /**
+ * Factory method for creating the appropriate adapter based on configuration.
+ * @param config Spout configuration.
+ * @return Appropriate Adapter.
+ */
+ private static JedisAdapter createAdapter(final RedisStreamSpoutConfig config, final int instanceId) {
+ final String connectStr = config.getConnectString().replaceAll("redis://", "");
+ if (config.isConnectingToCluster()) {
+ logger.info("Connecting to RedisCluster at {}", config.getConnectStringMasked());
+ return new JedisClusterAdapter(new JedisCluster(HostAndPort.parseString(connectStr)), config, instanceId);
+ } else {
+ logger.info("Connecting to RedisCluster at {}", config.getConnectStringMasked());
+ return new JedisRedisAdapter(new Jedis(HostAndPort.parseString(connectStr)), config, instanceId);
+ }
+ }
 }

src/main/java/org/sourcelab/storm/spout/redis/client/lettuce/LettuceClient.java

@@ -67,16 +67,6 @@ public LettuceClient(final RedisStreamSpoutConfig config, final int instanceId)
  );
  }
 
- private static LettuceAdapter createAdapter(final RedisStreamSpoutConfig config) {
- if (config.isConnectingToCluster()) {
- logger.info("Connecting to RedisCluster at {}", config.getConnectStringMasked());
- return new LettuceClusterAdapter(RedisClusterClient.create(config.getConnectString()));
- } else {
- logger.info("Connecting to Redis server at {}", config.getConnectStringMasked());
- return new LettuceRedisAdapter(RedisClient.create(config.getConnectString()));
- }
- }
-
  /**
  * Protected constructor for injecting a RedisClient instance, typically for tests.
  * @param config Configuration.
@@ -188,4 +178,19 @@ public void commitMessage(final String msgId) {
  public void disconnect() {
  adapter.shutdown();
  }
+
+ /**
+ * Factory method for creating the appropriate adapter based on configuration.
+ * @param config Spout configuration.
+ * @return Appropriate Adapter.
+ */
+ private static LettuceAdapter createAdapter(final RedisStreamSpoutConfig config) {
+ if (config.isConnectingToCluster()) {
+ logger.info("Connecting to RedisCluster at {}", config.getConnectStringMasked());
+ return new LettuceClusterAdapter(RedisClusterClient.create(config.getConnectString()));
+ } else {
+ logger.info("Connecting to Redis server at {}", config.getConnectStringMasked());
+ return new LettuceRedisAdapter(RedisClient.create(config.getConnectString()));
+ }
+ }
 }

src/test/java/org/sourcelab/storm/spout/redis/AbstractRedisStreamSpoutIntegrationTest.java

@@ -9,6 +9,9 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.sourcelab.storm.spout.redis.client.ClientType;
 import org.sourcelab.storm.spout.redis.example.TestTupleConverter;
 import org.sourcelab.storm.spout.redis.failhandler.RetryFailedTuples;
 import org.sourcelab.storm.spout.redis.util.outputcollector.EmittedTuple;
@@ -101,8 +104,12 @@ void cleanup() {
  /**
  * Most basic lifecycle smoke test.
  */
- @Test
- void smokeTest_openAndClose() {
+ @ParameterizedTest
+ @EnumSource(ClientType.class)
+ void smokeTest_openAndClose(final ClientType clientType) {
+ // Inject client type into config
+ configBuilder.withClientType(clientType);
+
  // Create spout
  try (final RedisStreamSpout spout = new RedisStreamSpout(configBuilder.build())) {
 
@@ -121,8 +128,12 @@ void smokeTest_openAndClose() {
  /**
  * Basic lifecycle smoke test.
  */
- @Test
- void smokeTest_openActivateDeactivateAndClose() throws InterruptedException {
+ @ParameterizedTest
+ @EnumSource(ClientType.class)
+ void smokeTest_openActivateDeactivateAndClose(final ClientType clientType) throws InterruptedException {
+ // Inject client type into config
+ configBuilder.withClientType(clientType);
+
  // Create spout
  try (final RedisStreamSpout spout = new RedisStreamSpout(configBuilder.build())) {
  final StubSpoutCollector collector = new StubSpoutCollector();
@@ -150,7 +161,12 @@ void smokeTest_openActivateDeactivateAndClose() throws InterruptedException {
  *
  * Disabled for now.
  */
- void smokeTest_configureInvalidRedisHost() throws InterruptedException {
+ @ParameterizedTest
+ @EnumSource(ClientType.class)
+ void smokeTest_configureInvalidRedisHost(final ClientType clientType) throws InterruptedException {
+ // Inject client type into config
+ configBuilder.withClientType(clientType);
+
  // Lets override the redis host with something invalid
  configBuilder
  .withServer(getTestContainer().getHost(), 124);
@@ -186,8 +202,12 @@ void smokeTest_configureInvalidRedisHost() throws InterruptedException {
  /**
  * Basic usage test.
  */
- @Test
- void smokeTest_consumeAndAckMessages() throws InterruptedException {
+ @ParameterizedTest
+ @EnumSource(ClientType.class)
+ void smokeTest_consumeAndAckMessages(final ClientType clientType) throws InterruptedException {
+ // Inject client type into config
+ configBuilder.withClientType(clientType);
+
  // Create spout
  try (final RedisStreamSpout spout = new RedisStreamSpout(configBuilder.build())) {
  final StubSpoutCollector collector = new StubSpoutCollector();
@@ -274,8 +294,12 @@ void smokeTest_consumeAndAckMessages() throws InterruptedException {
  /**
  * Basic usage with retry failure handler.
  */
- @Test
- void smokeTest_consumeFailAndAckMessages() throws InterruptedException {
+ @ParameterizedTest
+ @EnumSource(ClientType.class)
+ void smokeTest_consumeFailAndAckMessages(final ClientType clientType) throws InterruptedException {
+ // Inject client type into config
+ configBuilder.withClientType(clientType);
+
  // Swap out failure handler
  configBuilder.withFailureHandler(new RetryFailedTuples(2));
 
@@ -397,8 +421,12 @@ void smokeTest_consumeFailAndAckMessages() throws InterruptedException {
  /**
  * Verify declareOutputFields using TestTupleConverter.
  */
- @Test
- void test_declareOutputFields() {
+ @ParameterizedTest
+ @EnumSource(ClientType.class)
+ void test_declareOutputFields(final ClientType clientType) {
+ // Inject client type into config
+ configBuilder.withClientType(clientType);
+
  // Create a test implementation
  final TupleConverter converter = new DummyTupleConverter() ;
 
@@ -444,8 +472,12 @@ void test_declareOutputFields() {
  /**
  * Verify spout emits tuples down the correct stream.
  */
- @Test
- void test_EmitDownSeparateStreams() {
+ @ParameterizedTest
+ @EnumSource(ClientType.class)
+ void test_EmitDownSeparateStreams(final ClientType clientType) {
+ // Inject client type into config
+ configBuilder.withClientType(clientType);
+
  // Create a test implementation
  final TupleConverter converter = new DummyTupleConverter() ;
 
@@ -503,8 +535,12 @@ void test_EmitDownSeparateStreams() {
  * Verify if tuple converter instance returns null, then the message
  * is simply acked and nothing is emitted.
  */
- @Test
- void test_NullConversionJustGetsAckedNothingEmitted() {
+ @ParameterizedTest
+ @EnumSource(ClientType.class)
+ void test_NullConversionJustGetsAckedNothingEmitted(final ClientType clientType) {
+ // Inject client type into config
+ configBuilder.withClientType(clientType);
+
  // Create a test implementation
  final TupleConverter converter = new NullTupleConverter() ;
 

src/test/java/org/sourcelab/storm/spout/redis/RedisStreamSpout_ClusterIntegrationTest.java

@@ -6,7 +6,7 @@
 import org.testcontainers.junit.jupiter.Testcontainers;
 
 /**
- *
+ * Runs Spout integration tests against a RedisCluster.
  */
 @Testcontainers
 @Tag("Integration")