Extend request command limits to accept connection activation commands …

…#691 Request queue size checks now consider an additional number of connection activation commands.
redis · Feb 16, 2018 · e38c234 · e38c234
1 parent 07ed1b8
commit e38c234
Show file tree

Hide file tree

Showing 2 changed files with 69 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
@@ -480,17 +480,19 @@ private void validateWrite() {
 
         if (usesBoundedQueues()) {
 
+            boolean connected = isConnected();
+
             if (QUEUE_SIZE.get(this) + 1 > clientOptions.getRequestQueueSize()) {
                 throw new RedisException("Request queue size exceeded: " + clientOptions.getRequestQueueSize()
                         + ". Commands are not accepted until the queue size drops.");
             }
 
-            if (disconnectedBuffer.size() + 1 > clientOptions.getRequestQueueSize()) {
+            if (!connected && disconnectedBuffer.size() + 1 > clientOptions.getRequestQueueSize()) {
                 throw new RedisException("Request queue size exceeded: " + clientOptions.getRequestQueueSize()
                         + ". Commands are not accepted until the queue size drops.");
             }
 
-            if (commandBuffer.size() + 1 > clientOptions.getRequestQueueSize()) {
+            if (connected && commandBuffer.size() + 1 > clientOptions.getRequestQueueSize()) {
                 throw new RedisException("Command buffer size exceeded: " + clientOptions.getRequestQueueSize()
                         + ". Commands are not accepted until the queue size drops.");
             }
@@ -843,7 +845,11 @@ private void validateWrite(int commands) {
 
         if (usesBoundedQueues()) {
 
-            if (stack.size() + commands > clientOptions.getRequestQueueSize())
+            // number of maintenance commands (AUTH, CLIENT SETNAME, SELECT, READONLY) should be allowed on top
+            // of number of user commands to ensure the driver recovers properly from a disconnect
+            int maxMaintenanceCommands = 5;
+            int allowedRequestQueueSize = clientOptions.getRequestQueueSize() + maxMaintenanceCommands;
+            if (stack.size() + commands > allowedRequestQueueSize)
                 throw new RedisException("Internal stack size exceeded: " + clientOptions.getRequestQueueSize()
                         + ". Commands are not accepted until the stack size drops.");
         }

diff --git a/src/test/java/com/lambdaworks/redis/ClientOptionsTest.java b/src/test/java/com/lambdaworks/redis/ClientOptionsTest.java
@@ -119,6 +119,66 @@ public void requestQueueSizeAppliedForReconnect() {
         client.setOptions(ClientOptions.builder().requestQueueSize(10).build());
 
         RedisAsyncCommands<String, String> connection = client.connect().async();
+        testHitRequestQueueLimit(connection);
+    }
+
+    @Test
+    public void testHitRequestQueueLimitReconnectWithAuthCommand() {
+
+        new WithPasswordRequired() {
+
+            @Override
+            protected void run(RedisClient client) {
+
+                client.setOptions(ClientOptions.builder().requestQueueSize(10).build());
+
+                RedisAsyncCommands<String, String> connection = client.connect().async();
+                connection.auth(passwd);
+                testHitRequestQueueLimit(connection);
+            }
+        };
+    }
+
+    @Test
+    public void testHitRequestQueueLimitReconnectWithUriAuth() {
+
+        new WithPasswordRequired() {
+
+            @Override
+            protected void run(RedisClient client) {
+
+                client.setOptions(ClientOptions.builder().requestQueueSize(10).build());
+
+                RedisURI redisURI = RedisURI.create(host, port);
+                redisURI.setPassword(passwd);
+
+                RedisAsyncCommands<String, String> connection = client.connect(redisURI).async();
+                testHitRequestQueueLimit(connection);
+            }
+        };
+    }
+
+    @Test
+    public void testHitRequestQueueLimitReconnectWithUriAuthPingCommand() {
+
+        new WithPasswordRequired() {
+
+            @Override
+            protected void run(RedisClient client) {
+
+                client.setOptions(ClientOptions.builder().requestQueueSize(10).pingBeforeActivateConnection(true).build());
+
+                RedisURI redisURI = RedisURI.create(host, port);
+                redisURI.setPassword(passwd);
+
+                RedisAsyncCommands<String, String> connection = client.connect(redisURI).async();
+                testHitRequestQueueLimit(connection);
+            }
+        };
+    }
+
+    private void testHitRequestQueueLimit(RedisAsyncCommands<String, String> connection) {
+
         ConnectionWatchdog watchdog = getConnectionWatchdog(connection.getStatefulConnection());
 
         watchdog.setListenOnChannelInactive(false);