Fix connection inconsistencies on all drivers.

Author: djanowski

lib/redis/client.rb

@@ -77,6 +77,8 @@ def initialize(options = {})
       @connection = nil
       @command_map = {}
 
+      @pending_reads = 0
+
       if options.include?(:sentinels)
         @connector = Connector::Sentinel.new(@options)
       else
@@ -243,12 +245,15 @@ def io
 
     def read
       io do
-        connection.read
+        value = connection.read
+        @pending_reads -= 1
+        value
       end
     end
 
     def write(command)
       io do
+        @pending_reads += 1
         connection.write(command)
       end
     end
@@ -315,6 +320,7 @@ def establish_connection
       @options[:port] = server[:port]
 
       @connection = @options[:driver].connect(server)
+      @pending_reads = 0
     rescue TimeoutError,
            Errno::ECONNREFUSED,
            Errno::EHOSTDOWN,
@@ -326,6 +332,8 @@ def establish_connection
     end
 
     def ensure_connected
+      disconnect if @pending_reads > 0
+
       attempts = 0
 
       begin
@@ -342,10 +350,7 @@ def ensure_connected
           connect
         end
 
-        connection.use { yield }
-      rescue ConnectionCorruptedError
-        disconnect
-        retry
+        yield
       rescue BaseConnectionError
         disconnect
 

lib/redis/connection/hiredis.rb

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

lib/redis/connection/ruby.rb

@@ -246,7 +246,6 @@ def get_tcp_keepalive
 
       def initialize(sock)
         @sock = sock
-        @pending_reads = 0
       end
 
       def connected?
@@ -267,7 +266,6 @@ def timeout=(timeout)
       end
 
       def write(command)
-        @pending_reads += 1
         @sock.write(build_command(command))
       end
 
@@ -292,22 +290,18 @@ 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))
@@ -317,20 +311,10 @@ 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,10 +117,6 @@ def read
           raise "Unknown type #{type.inspect}"
         end
       end
-
-      def use
-        yield
-      end
     end
   end
 end

lib/redis/errors.rb

@@ -37,11 +37,4 @@ 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