Synchronize access to transport buffer #324

redis · Jul 27, 2016 · 41c7727 · 41c7727
1 parent 207ca0d
commit 41c7727
Showing 1 changed file with 12 additions and 3 deletions.
diff --git a/src/main/java/com/lambdaworks/redis/protocol/CommandHandler.java b/src/main/java/com/lambdaworks/redis/protocol/CommandHandler.java
@@ -405,7 +405,7 @@ private void writeSingleCommand(ChannelHandlerContext ctx, RedisCommand<K, V, ?>
             throws Exception {
 
         if (command.isCancelled()) {
-            transportBuffer.remove(command);
+            removeFromTransportBuffer(command);
             return;
         }
 
@@ -434,7 +434,7 @@ private void writeBatch(ChannelHandlerContext ctx, Collection<RedisCommand<K, V,
             for (RedisCommand<K, V, ?> command : commands) {
 
                 if (command.isCancelled()) {
-                    transportBuffer.remove(command);
+                    removeFromTransportBuffer(command);
                     continue;
                 }
 
@@ -465,7 +465,7 @@ private void queueCommand(RedisCommand<K, V, ?> command, ChannelPromise promise)
                 sentTimes.put(command, new SentReceived(nanoTime()));
                 queue.add(command);
             }
-            transportBuffer.remove(command);
+            removeFromTransportBuffer(command);
         } catch (Exception e) {
             command.setException(e);
             command.cancel(true);
@@ -474,6 +474,15 @@ private void queueCommand(RedisCommand<K, V, ?> command, ChannelPromise promise)
         }
     }
 
+    private void removeFromTransportBuffer(RedisCommand<K, V, ?> command) {
+        try {
+            writeLock.lock();
+            transportBuffer.remove(command);
+        } finally {
+            writeLock.unlock();
+        }
+    }
+
     private long nanoTime() {
         return System.nanoTime();
     }