Merge pull request #234 from gertgoeman/master

Exception in pipeline callback breaks RedisClient

Author: mythz

src/ServiceStack.Redis/Pipeline/RedisAllPurposePipeline.cs

@@ -35,14 +35,24 @@ public void Flush()
         {
             // flush send buffers
             RedisClient.FlushAndResetSendBuffer();
-
-            //receive expected results
-            foreach (var queuedCommand in QueuedCommands)
+            
+            try
+            {
+                //receive expected results
+                foreach (var queuedCommand in QueuedCommands)
+                {
+                    queuedCommand.ProcessResult();
+                }
+            }
+            catch (Exception)
             {
-                queuedCommand.ProcessResult();
+                // The connection cannot be reused anymore. All queued commands have been sent to redis. Even if a new command is executed, the next response read from the
+                // network stream can be the response of one of the queued commands, depending on when the exception occurred. This response would be invalid for the new command.
+                RedisClient.DisposeConnection();
+                throw;
             }
+            
             ClosePipeline();
-
         }
 
         protected void Execute()

tests/ServiceStack.Redis.Tests/Issues/PipelineIssueTests.cs

@@ -0,0 +1,43 @@
+using System;
+using NUnit.Framework;
+
+namespace ServiceStack.Redis.Tests.Issues
+{
+    [TestFixture]
+    public class PipelineIssueTests
+        : RedisClientTestsBase
+    {
+        [Test]
+        public void Disposing_Client_Clears_Pipeline()
+        {
+            var clientMgr = new PooledRedisClientManager(TestConfig.SingleHost);
+
+            using (var client = clientMgr.GetClient())
+            {
+                client.Set("k1", "v1");
+                client.Set("k2", "v2");
+                client.Set("k3", "v3");
+                
+                using (var pipe = client.CreatePipeline())
+                {
+                    pipe.QueueCommand(c => c.Get<string>("k1"), p => { throw new Exception(); });
+                    pipe.QueueCommand(c => c.Get<string>("k2"));
+
+                    try
+                    {
+                        pipe.Flush();
+                    }
+                    catch (Exception)
+                    {
+                        //The exception is expected. Swallow it.
+                    }
+                }
+            }
+
+            using (var client = clientMgr.GetClient())
+            {
+                Assert.AreEqual("v3", client.Get<string>("k3"));
+            }
+        }
+    }
+}

tests/ServiceStack.Redis.Tests/ServiceStack.Redis.Tests.csproj

@@ -195,6 +195,7 @@
     <Compile Include="ConfigTests.cs" />
     <Compile Include="CustomCommandTests.cs" />
     <Compile Include="Issues\AuthIssue.cs" />
+    <Compile Include="Issues\PipelineIssueTests.cs" />
     <Compile Include="LuaCachedScripts.cs" />
     <Compile Include="Examples\TestData.cs" />
     <Compile Include="Issues\RedisCharacterizationTests.cs" />