cleanup

Author: Crim

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

@@ -21,7 +21,7 @@
 /**
  * Redis Stream based Spout for Apache Storm 2.2.x.
  */
-public class RedisStreamSpout implements IRichSpout {
+public class RedisStreamSpout implements IRichSpout, AutoCloseable {
     private static final Logger logger = LoggerFactory.getLogger(RedisStreamSpout.class);
 
     /**

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

@@ -37,7 +37,7 @@ public class LettuceClient implements Client {
     /**
      * The underlying Redis Client.
      */
-    private final LettuceAdapter client;
+    private final LettuceAdapter adapter;
 
     /**
      * Re-usable instance to prevent unnecessary garbage creation.
@@ -58,20 +58,20 @@ public LettuceClient(final RedisStreamSpoutConfig config, final int instanceId)
             // Determine which adapter to use based on what type of redis instance we are
             // communicating with.
             config.isConnectingToCluster()
-                ? new LettuceClusterClient(RedisClusterClient.create(config.getConnectString()))
-                : new LettuceRedisClient(RedisClient.create(config.getConnectString()))
+                ? new LettuceClusterAdapter(RedisClusterClient.create(config.getConnectString()))
+                : new LettuceRedisAdapter(RedisClient.create(config.getConnectString()))
         );
     }
 
     /**
      * Protected constructor for injecting a RedisClient instance, typically for tests.
      * @param config Configuration.
      * @param instanceId Which instance number is this running under.
-     * @param client RedisClient instance.
+     * @param adapter RedisClient instance.
      */
-    LettuceClient(final RedisStreamSpoutConfig config, final int instanceId, final LettuceAdapter client) {
+    LettuceClient(final RedisStreamSpoutConfig config, final int instanceId, final LettuceAdapter adapter) {
         this.config = Objects.requireNonNull(config);
-        this.client = Objects.requireNonNull(client);
+        this.adapter = Objects.requireNonNull(adapter);
 
         // Calculate consumerId
         this.consumerId = config.getConsumerIdPrefix() + instanceId;
@@ -92,15 +92,15 @@ public LettuceClient(final RedisStreamSpoutConfig config, final int instanceId)
 
     @Override
     public void connect() {
-        if (client.isConnected()) {
+        if (adapter.isConnected()) {
             throw new IllegalStateException("Cannot call connect more than once!");
         }
 
-        client.connect();
+        adapter.connect();
 
         try {
             // Attempt to create consumer group
-            client.getSyncCommands().xgroupCreate(
+            adapter.getSyncCommands().xgroupCreate(
                 // Start the group at first offset for our key.
                 XReadArgs.StreamOffset.from(config.getStreamKey(), "0-0"),
                 // Define the group name
@@ -128,7 +128,7 @@ public void connect() {
     @Override
     public List<Message> nextMessages() {
         // Get next batch of messages.
-        final List<StreamMessage<String, String>> messages = client.getSyncCommands().xreadgroup(
+        final List<StreamMessage<String, String>> messages = adapter.getSyncCommands().xreadgroup(
             consumerFrom,
             xreadArgs,
             lastConsumed
@@ -144,7 +144,7 @@ public List<Message> nextMessages() {
     @Override
     public void commitMessage(final String msgId) {
         // Confirm that the message has been processed using XACK
-        client.getSyncCommands().xack(
+        adapter.getSyncCommands().xack(
             config.getStreamKey(),
             config.getGroupName(),
             msgId
@@ -153,6 +153,6 @@ public void commitMessage(final String msgId) {
 
     @Override
     public void disconnect() {
-        client.shutdown();
+        adapter.shutdown();
     }
 }

src/main/java/org/sourcelab/storm/spout/redis/client/LettuceClusterClient.java renamed to src/main/java/org/sourcelab/storm/spout/redis/client/LettuceClusterAdapter.java

@@ -8,9 +8,9 @@
 
 /**
  * Adapter for talking to a RedisCluster.
- * If you need to talk to a single Redis instance {@link LettuceRedisClient}.
+ * If you need to talk to a single Redis instance {@link LettuceRedisAdapter}.
  */
-public class LettuceClusterClient implements LettuceAdapter {
+public class LettuceClusterAdapter implements LettuceAdapter {
     /**
      * The underlying Redis Client.
      */
@@ -22,7 +22,7 @@ public class LettuceClusterClient implements LettuceAdapter {
     private StatefulRedisClusterConnection<String, String> connection;
     private RedisStreamCommands<String, String> syncCommands;
 
-    public LettuceClusterClient(final RedisClusterClient redisClient) {
+    public LettuceClusterAdapter(final RedisClusterClient redisClient) {
         this.redisClient = Objects.requireNonNull(redisClient);
     }
 
@@ -51,6 +51,7 @@ public RedisStreamCommands<String, String> getSyncCommands() {
     public void shutdown() {
         // Close our connection and shutdown.
         if (connection != null) {
+            syncCommands = null;
             connection.close();
             connection = null;
         }

src/main/java/org/sourcelab/storm/spout/redis/client/LettuceRedisClient.java renamed to src/main/java/org/sourcelab/storm/spout/redis/client/LettuceRedisAdapter.java

@@ -8,9 +8,9 @@
 
 /**
  * Adapter for talking to a single Redis instance.
- * If you need to talk to a RedisCluster {@link LettuceClusterClient}.
+ * If you need to talk to a RedisCluster {@link LettuceClusterAdapter}.
  */
-public class LettuceRedisClient implements LettuceAdapter {
+public class LettuceRedisAdapter implements LettuceAdapter {
 
     /**
      * The underlying Redis Client.
@@ -23,7 +23,7 @@ public class LettuceRedisClient implements LettuceAdapter {
     private StatefulRedisConnection<String, String> connection;
     private RedisStreamCommands<String, String> syncCommands;
 
-    public LettuceRedisClient(final RedisClient redisClient) {
+    public LettuceRedisAdapter(final RedisClient redisClient) {
         this.redisClient = Objects.requireNonNull(redisClient);
     }
 
@@ -52,6 +52,7 @@ public RedisStreamCommands<String, String> getSyncCommands() {
     public void shutdown() {
         // Close our connection and shutdown.
         if (connection != null) {
+            syncCommands = null;
             connection.close();
             connection = null;
         }