better error handling with hiredis. fully fixes #456

3 files changed, 47 insertions, 27 deletions

diff --git a/redis/connection.py b/redis/connection.py
index eae8619..92a5c16 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -1,10 +1,12 @@
 from __future__ import with_statement
+from distutils.version import StrictVersion
 from itertools import chain
 from select import select
 import os
 import socket
 import sys
 import threading
+import warnings
 
 from redis._compat import (b, xrange, imap, byte_to_chr, unicode, bytes, long,
                            BytesIO, nativestr, basestring,
@@ -23,6 +25,14 @@ from redis.utils import HIREDIS_AVAILABLE
 if HIREDIS_AVAILABLE:
     import hiredis
 
+    hiredis_version = StrictVersion(hiredis.__version__)
+    HIREDIS_SUPPORTS_CALLABLE_ERRORS = \
+        hiredis_version >= StrictVersion('0.1.3')
+
+    if not HIREDIS_SUPPORTS_CALLABLE_ERRORS:
+        msg = ('redis-py works best with hiredis >= 0.1.3. Your hiredis '
+               'is %s. Please consider upgrading.' % hiredis.__version__)
+        warnings.warn(msg)
 
 SYM_STAR = b('*')
 SYM_DOLLAR = b('$')
@@ -48,8 +58,6 @@ class BaseParser(object):
 
 
 class SocketBuffer(object):
-    MAX_READ_LENGTH = 1000000
-
     def __init__(self, socket):
         self._sock = socket
         self._buffer = BytesIO()
@@ -222,8 +230,12 @@ class HiredisParser(BaseParser):
         self._sock = connection._sock
         kwargs = {
             'protocolError': InvalidResponse,
-            'replyError': ResponseError,
+            'replyError': self.parse_error,
         }
+
+        if not HIREDIS_SUPPORTS_CALLABLE_ERRORS:
+            kwargs['replyError'] = ResponseError
+
         if connection.decode_responses:
             kwargs['encoding'] = connection.encoding
         self._reader = hiredis.Reader(**kwargs)
@@ -268,13 +280,22 @@ class HiredisParser(BaseParser):
             if not buffer.endswith(SYM_CRLF):
                 continue
             response = self._reader.gets()
-        if isinstance(response, ResponseError):
-            response = self.parse_error(response.args[0])
-        # hiredis only knows about ResponseErrors.
-        # self.parse_error() might turn the exception into a ConnectionError
-        # which needs raising.
+        # if an older version of hiredis is installed, we need to attempt
+        # to convert ResponseErrors to their appropriate types.
+        if not HIREDIS_SUPPORTS_CALLABLE_ERRORS:
+            if isinstance(response, ResponseError):
+                response = self.parse_error(response.args[0])
+            elif isinstance(response, list) and response and \
+                    isinstance(response[0], ResponseError):
+                response[0] = self.parse_error(response[0].args[0])
+        # if the response is a ConnectionError or the response is a list and
+        # the first item is a ConnectionError, raise it as something bad
+        # happened
         if isinstance(response, ConnectionError):
             raise response
+        elif isinstance(response, list) and response and \
+                isinstance(response[0], ConnectionError):
+            raise response[0]
         return response
 
 if HIREDIS_AVAILABLE:
diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py
index b4b6268..c678b8e 100644
--- a/tests/test_connection_pool.py
+++ b/tests/test_connection_pool.py
@@ -203,18 +203,17 @@ class TestConnection(object):
         assert len(pool._available_connections) == 1
         assert not pool._available_connections[0]._sock
 
-    # TODO: This fails on hiredis. need to think about this
-    # @skip_if_server_version_lt('2.8.8')
-    # def test_busy_loading_from_pipeline(self, r):
-    #     """
-    #     BusyLoadingErrors should be raised from a pipeline execution
-    #     regardless of the raise_on_error flag.
-    #     """
-    #     pipe = r.pipeline()
-    #     pipe.execute_command('DEBUG', 'ERROR', 'LOADING fake message')
-    #     with pytest.raises(redis.BusyLoadingError):
-    #         pipe.execute()
-    #     pool = r.connection_pool
-    #     assert not pipe.connection
-    #     assert len(pool._available_connections) == 1
-    #     assert not pool._available_connections[0]._sock
+    @skip_if_server_version_lt('2.8.8')
+    def test_busy_loading_from_pipeline(self, r):
+        """
+        BusyLoadingErrors should be raised from a pipeline execution
+        regardless of the raise_on_error flag.
+        """
+        pipe = r.pipeline()
+        pipe.execute_command('DEBUG', 'ERROR', 'LOADING fake message')
+        with pytest.raises(redis.BusyLoadingError):
+            pipe.execute()
+        pool = r.connection_pool
+        assert not pipe.connection
+        assert len(pool._available_connections) == 1
+        assert not pool._available_connections[0]._sock
diff --git a/tox.ini b/tox.ini
index 5dd93df..0b6f4ce 100644
--- a/tox.ini
+++ b/tox.ini
@@ -8,28 +8,28 @@ commands = py.test []
 [testenv:hi26]
 basepython = python2.6
 deps =
-    hiredis
+    hiredis>=0.1.3
     pytest>=2.5.0
 commands = py.test []
 
 [testenv:hi27]
 basepython = python2.7
 deps =
-    hiredis
+    hiredis>=0.1.3
     pytest>=2.5.0
 commands = py.test []
 
 [testenv:hi32]
 basepython = python3.2
 deps =
-    hiredis
+    hiredis>=0.1.3
     pytest>=2.5.0
 commands = py.test []
 
 [testenv:hi33]
 basepython = python3.3
 deps =
-    hiredis
+    hiredis>=0.1.3
     pytest>=2.5.0
 commands = py.test []