Fix inconsistent connections when using Timeout under MRI 2.1+.

Author: djanowski

lib/redis/client.rb

@@ -342,7 +342,10 @@ def ensure_connected
           connect
         end
 
-        yield
+        connection.use { yield }
+      rescue ConnectionCorruptedError
+        disconnect
+        retry
       rescue BaseConnectionError
         disconnect
 

lib/redis/connection/hiredis.rb

@@ -56,6 +56,10 @@ def read
       rescue RuntimeError => err
         raise ProtocolError.new(err.message)
       end
+
+      def use
+        yield
+      end
     end
   end
 end

lib/redis/connection/ruby.rb

@@ -246,6 +246,7 @@ def get_tcp_keepalive
 
       def initialize(sock)
         @sock = sock
+        @pending_reads = 0
       end
 
       def connected?
@@ -266,6 +267,7 @@ def timeout=(timeout)
       end
 
       def write(command)
+        @pending_reads += 1
         @sock.write(build_command(command))
       end
 
@@ -290,18 +292,22 @@ def format_reply(reply_type, line)
       end
 
       def format_error_reply(line)
+        @pending_reads -= 1
         CommandError.new(line.strip)
       end
 
       def format_status_reply(line)
+        @pending_reads -= 1
         line.strip
       end
 
       def format_integer_reply(line)
+        @pending_reads -= 1
         line.to_i
       end
 
       def format_bulk_reply(line)
+        @pending_reads -= 1
         bulklen = line.to_i
         return if bulklen == -1
         reply = encode(@sock.read(bulklen))
@@ -311,10 +317,20 @@ def format_bulk_reply(line)
 
       def format_multi_bulk_reply(line)
         n = line.to_i
+
+        @pending_reads += n
+
         return if n == -1
 
+        @pending_reads -= 1
+
         Array.new(n) { read }
       end
+
+      def use
+        raise ConnectionCorruptedError if @pending_reads > 0
+        yield
+      end
     end
   end
 end

lib/redis/connection/synchrony.rb

@@ -117,6 +117,10 @@ def read
           raise "Unknown type #{type.inspect}"
         end
       end
+
+      def use
+        yield
+      end
     end
   end
 end

lib/redis/errors.rb

@@ -37,4 +37,11 @@ class TimeoutError < BaseConnectionError
   # Raised when the connection was inherited by a child process.
   class InheritedError < BaseConnectionError
   end
+
+  # Raised when the connection is left in an inconsistent state
+  # due to timeout issues. See issue #501.
+  #
+  # Only used internally.
+  class ConnectionCorruptedError < BaseConnectionError
+  end
 end