better error handling (#18)

examples/callbacks.py

@@ -5,7 +5,7 @@
 
 
 def ping_callback(result):
-    if not isinstance(result, tornadis.ConnectionError):
+    if not isinstance(result, tornadis.TornadisException):
         print result
 
 

examples/context_manager.py

@@ -9,7 +9,7 @@ def ping_redis(num):
         # it will be automatically released to the pool thanks to the
         # "with" keyword
         reply = yield client.call("PING")
-        if not isinstance(reply, tornadis.ConnectionError):
+        if not isinstance(reply, tornadis.TornadisException):
             print("reply #%i : %s" % (num, reply))
 
 

examples/coroutines.py

@@ -7,7 +7,8 @@
 @tornado.gen.coroutine
 def talk_to_redis():
     result = yield client.call("PING")
-    print "Result: %s" % result
+    if not isinstance(result, tornadis.TornadisException):
+        print "Result: %s" % result
 
 
 loop = tornado.ioloop.IOLoop.instance()

examples/web.py

@@ -14,7 +14,7 @@ class HelloHandler(RequestHandler):
     def get(self):
         with (yield POOL.connected_client()) as client:
             reply = yield client.call("PING")
-            if not isinstance(reply, tornadis.ConnectionError):
+            if not isinstance(reply, tornadis.TornadisException):
                 self.write("Hello, %s" % reply)
         self.finish()
 

tests/test_client.py

@@ -8,7 +8,7 @@
 import functools
 
 from tornadis.client import Client
-from tornadis.exceptions import ConnectionError
+from tornadis.exceptions import ConnectionError, ClientError
 from support import test_redis_or_raise_skiptest
 
 
@@ -35,6 +35,14 @@ def test_ping(self):
         self.assertEquals(res, b"PONG")
         c.disconnect()
 
+    @tornado.testing.gen_test
+    def test_reply_error(self):
+        c = Client()
+        yield c.connect()
+        res = yield c.call('BADCOMMAND')
+        self.assertTrue(isinstance(res, ClientError))
+        c.disconnect()
+
     @tornado.testing.gen_test
     def test_autoconnect_future(self):
         c = Client(autoconnect=True)

tornadis/__init__.py

@@ -15,7 +15,8 @@
 from tornadis.pubsub import PubSubClient
 from tornadis.pool import ClientPool
 from tornadis.pipeline import Pipeline
-from tornadis.exceptions import ConnectionError, ClientError
+from tornadis.exceptions import ConnectionError, ClientError, TornadisException
 
 __all__ = ['Client', 'ClientPool', 'Pipeline',
-           'ConnectionError', 'ClientError', 'PubSubClient', 'WriteBuffer']
+           'ConnectionError', 'ClientError', 'TornadisException',
+           'PubSubClient', 'WriteBuffer']

tornadis/client.py

@@ -9,6 +9,7 @@
 import collections
 import functools
 import toro
+import logging
 
 from tornadis.connection import Connection
 from tornadis.pipeline import Pipeline
@@ -17,6 +18,9 @@
 from tornadis.exceptions import ConnectionError, ClientError
 
 
+LOG = logging.getLogger(__name__)
+
+
 def discard_reply_cb(reply):
     pass
 
@@ -82,7 +86,7 @@ def connect(self):
         cb2 = self._close_callback
         self.__callback_queue = collections.deque()
         self._reply_list = []
-        self.__reader = hiredis.Reader()
+        self.__reader = hiredis.Reader(replyError=ClientError)
         kwargs = self.connection_kwargs
         self.__connection = Connection(cb1, cb2, **kwargs)
         return self.__connection.connect()
@@ -119,21 +123,26 @@ def _read_callback(self, data=None):
         Args:
             data (str): string (buffer) read on the socket.
         """
-        if data is not None:
-            self.__reader.feed(data)
-            while True:
-                reply = self.__reader.gets()
-                if reply is not False:
-                    try:
-                        callback = self.__callback_queue.popleft()
-                        # normal client (1 reply = 1 callback)
-                        callback(reply)
-                    except IndexError:
-                        # pubsub clients
-                        self._reply_list.append(reply)
-                        self._condition.notify_all()
-                else:
-                    break
+        try:
+            if data is not None:
+                self.__reader.feed(data)
+                while True:
+                    reply = self.__reader.gets()
+                    if reply is not False:
+                        try:
+                            callback = self.__callback_queue.popleft()
+                            # normal client (1 reply = 1 callback)
+                            callback(reply)
+                        except IndexError:
+                            # pubsub clients
+                            self._reply_list.append(reply)
+                            self._condition.notify_all()
+                    else:
+                        break
+        except hiredis.ProtocolError:
+            # something nasty occured (corrupt stream => no way to recover)
+            LOG.warning("corrupted stream => disconnect")
+            self.disconnect()
 
     def call(self, *args, **kwargs):
         """Calls a redis command and returns a Future of the reply.