Merge pull request #197 from stanhu/sh-fix-brpop-timeouts

Ensure BLPOP/BRPOP returns nil instead of raising ReadTimeoutError

lib/redis_client/connection_mixin.rb

@@ -28,7 +28,7 @@ def revalidate
     def call(command, timeout)
       @pending_reads += 1
       write(command)
-      result = read(timeout)
+      result = read(connection_timeout(timeout))
       @pending_reads -= 1
       if result.is_a?(Error)
         result._set_command(command)
@@ -49,7 +49,7 @@ def call_pipelined(commands, timeouts, exception: true)
 
       size.times do |index|
         timeout = timeouts && timeouts[index]
-        result = read(timeout)
+        result = read(connection_timeout(timeout))
         @pending_reads -= 1
 
         # A multi/exec command can return an array of results.
@@ -73,5 +73,14 @@ def call_pipelined(commands, timeouts, exception: true)
         results
       end
     end
+
+    def connection_timeout(timeout)
+      return timeout unless timeout && timeout > 0
+
+      # Can't use the command timeout argument as the connection timeout
+      # otherwise it would be very racy. So we add the regular read_timeout on top
+      # to account for the network delay.
+      timeout + config.read_timeout
+    end
   end
 end

test/shared/redis_client_tests.rb

@@ -448,12 +448,30 @@ def test_blocking_call_timeout
     assert_equal "OK", @redis.call("SET", "foo", "bar")
   end
 
+  def test_blocking_call_long_timeout
+    client = new_client(timeout: 0.5)
+    assert_nil client.blocking_call(0.5, "BRPOP", "list", "0.5")
+    assert_nil client.blocking_call_v(0.5, ["BRPOP", "list", "0.5"])
+
+    result = client.pipelined do |pipeline|
+      pipeline.blocking_call(0.5, "BRPOP", "list", "0.5")
+      pipeline.blocking_call_v(0.5, ["BRPOP", "list", "0.5"])
+    end
+
+    assert_equal([nil, nil], result)
+
+    result = client.multi do |pipeline|
+      pipeline.call("BRPOP", "list", "0.5")
+      pipeline.call_v(["BRPOP", "list", "0.5"])
+    end
+
+    assert_equal([nil, nil], result)
+  end
+
   def test_blocking_call_timeout_retries
-    redis = new_client(reconnect_attempts: [3.0])
+    redis = new_client(timeout: 0.5, reconnect_attempts: [3.0])
     start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
-    assert_raises RedisClient::ReadTimeoutError do
-      redis.blocking_call(0.1, "BRPOP", "list", "0.1")
-    end
+    assert_nil redis.blocking_call(0.1, "BRPOP", "list", "0.1")
     duration = Process.clock_gettime(Process::CLOCK_MONOTONIC) - start
     assert duration < 0.5 # if we retried we'd have waited much long
   end