Merge pull request #338 from stephan-hof/corrupted_pipeline_socket

socket gets corrupted if errors in pipeline happen

2 files changed, 24 insertions, 2 deletions

diff --git a/redis/client.py b/redis/client.py
index 0369d30..d9506e6 100644
--- a/redis/client.py
+++ b/redis/client.py
@@ -1794,8 +1794,15 @@ class BasePipeline(object):
             starmap(connection.pack_command,
                     [args for args, options in commands]))
         connection.send_packed_command(all_cmds)
-        response = [self.parse_response(connection, args[0], **options)
-                    for args, options in commands]
+
+        response = []
+        for args, options in commands:
+            try:
+                response.append(
+                    self.parse_response(connection, args[0], **options))
+            except ResponseError:
+                response.append(sys.exc_info()[1])
+
         if raise_on_error:
             self.raise_first_error(response)
         return response
diff --git a/tests/pipeline.py b/tests/pipeline.py
index a21c882..198ee58 100644
--- a/tests/pipeline.py
+++ b/tests/pipeline.py
@@ -185,3 +185,18 @@ class PipelineTestCase(unittest.TestCase):
         result = self.client.transaction(my_transaction, 'a', 'b')
         self.assertEquals(result, [True])
         self.assertEquals(self.client.get('c'), b('4'))
+
+    def test_error_in_simple_pipeline(self):
+        self.client.hmset('x', {'a': 'b'})
+        with self.client.pipeline(transaction=False) as pipe:
+            pipe.llen('x')
+            pipe.expire('x', 100)
+            try:
+                pipe.execute()
+            except redis.ResponseError:
+                pass
+            else:
+                raise
+
+        ret = self.client.hgetall('x')
+        self.assertEqual(ret, {b('a'): b('b')})