1. when completing queued operations, simply return if there is no current queued op,

Aaron Boxer · Aaron Boxer · commit bb939072e944 · 2011-01-03T10:49:41.000-05:00
rather than throwing exception: this situation can occur when replaying pipeline/transaction
2. now re-throwing exception when completing queued op
3. added some tests for above fixes
diff --git a/src/ServiceStack.Redis/Pipeline/RedisQueueCompletableOperation.cs b/src/ServiceStack.Redis/Pipeline/RedisQueueCompletableOperation.cs
@@ -35,63 +35,79 @@ protected virtual void AddCurrentQueuedOperation()
 
         public virtual void CompleteVoidQueuedCommand(Action voidReadCommand)
         {
-            AssertCurrentOperation();
+            //AssertCurrentOperation();
+            // this can happen when replaying pipeline/transaction
+            if (CurrentQueuedOperation == null) return;
 
             CurrentQueuedOperation.VoidReadCommand = voidReadCommand;
             AddCurrentQueuedOperation();
         }
 
         public virtual  void CompleteIntQueuedCommand(Func<int> intReadCommand)
         {
-            AssertCurrentOperation();
+            //AssertCurrentOperation();
+            // this can happen when replaying pipeline/transaction
+            if (CurrentQueuedOperation == null) return;
 
             CurrentQueuedOperation.IntReadCommand = intReadCommand;
             AddCurrentQueuedOperation();
         }
 
         public virtual void CompleteLongQueuedCommand(Func<long> longReadCommand)
         {
-            AssertCurrentOperation();
+            //AssertCurrentOperation();
+            // this can happen when replaying pipeline/transaction
+            if (CurrentQueuedOperation == null) return;
 
             CurrentQueuedOperation.LongReadCommand = longReadCommand;
             AddCurrentQueuedOperation();
         }
 
         public virtual void CompleteBytesQueuedCommand(Func<byte[]> bytesReadCommand)
         {
-            AssertCurrentOperation();
+            //AssertCurrentOperation();
+            // this can happen when replaying pipeline/transaction
+            if (CurrentQueuedOperation == null) return;
 
             CurrentQueuedOperation.BytesReadCommand = bytesReadCommand;
             AddCurrentQueuedOperation();
         }
 
         public virtual void CompleteMultiBytesQueuedCommand(Func<byte[][]> multiBytesReadCommand)
         {
-            AssertCurrentOperation();
+            //AssertCurrentOperation();
+            // this can happen when replaying pipeline/transaction
+            if (CurrentQueuedOperation == null) return;
 
             CurrentQueuedOperation.MultiBytesReadCommand = multiBytesReadCommand;
             AddCurrentQueuedOperation();
         }
 
         public virtual void CompleteStringQueuedCommand(Func<string> stringReadCommand)
         {
-            AssertCurrentOperation();
+            //AssertCurrentOperation();
+            // this can happen when replaying pipeline/transaction
+            if (CurrentQueuedOperation == null) return;
 
             CurrentQueuedOperation.StringReadCommand = stringReadCommand;
             AddCurrentQueuedOperation();
         }
 
         public virtual void CompleteMultiStringQueuedCommand(Func<List<string>> multiStringReadCommand)
         {
-            AssertCurrentOperation();
+            //AssertCurrentOperation();
+            // this can happen when replaying pipeline/transaction
+            if (CurrentQueuedOperation == null) return;
 
             CurrentQueuedOperation.MultiStringReadCommand = multiStringReadCommand;
             AddCurrentQueuedOperation();
         }
 
         public virtual void CompleteDoubleQueuedCommand(Func<double> doubleReadCommand)
         {
-            AssertCurrentOperation();
+            //AssertCurrentOperation();
+            // this can happen when replaying pipeline/transaction
+            if (CurrentQueuedOperation == null) return;
 
             CurrentQueuedOperation.DoubleReadCommand = doubleReadCommand;
             AddCurrentQueuedOperation();
diff --git a/tests/ServiceStack.Redis.Tests/RedisPipelineTests.cs b/tests/ServiceStack.Redis.Tests/RedisPipelineTests.cs
@@ -73,6 +73,29 @@ public void Can_call_single_operation_3_Times_in_pipeline()
 
 			Assert.That(Redis.GetValue(Key), Is.EqualTo("3"));
 		}
+        [Test]
+        public void Can_call_multiple_setexs_in_pipeline()
+        {
+            Assert.That(Redis.GetValue(Key), Is.Null);
+            var keys = new[] {"key1", "key2", "key3"};
+            var values = new[] { "1","2","3" };
+            var pipeline = Redis.CreatePipeline();
+          
+            for (int i = 0; i < 3; ++i )
+            {
+                int index0 = i;
+                pipeline.QueueCommand(r => ((RedisNativeClient)r).SetEx(keys[index0], 100, GetBytes(values[index0])));
+            }
+
+            pipeline.Flush();
+            pipeline.Replay();
+        
+            
+            for (int i = 0; i < 3; ++i )
+                Assert.AreEqual(Redis.GetValue(keys[i]), values[i]);
+
+            pipeline.Dispose();
+        }
 
 		[Test]
 		public void Can_call_single_operation_with_callback_3_Times_in_pipeline()
diff --git a/tests/ServiceStack.Redis.Tests/RedisTransactionTests.cs b/tests/ServiceStack.Redis.Tests/RedisTransactionTests.cs
@@ -40,6 +40,30 @@ public void No_commit_of_atomic_transactions_discards_all_commands()
 			Assert.That(Redis.GetValue(Key), Is.Null);
 		}
 
+        [Test]
+        public void Watch_aborts_transaction()
+        {
+            Assert.That(Redis.GetValue(Key), Is.Null);
+            const string value1 = "value1";
+            const string value2 = "value2";
+            try
+            {
+                Redis.Watch(Key);
+                Redis.Set(Key, value1);
+                using (var trans = Redis.CreateTransaction())
+                {
+                    trans.QueueCommand(r => r.Set(Key,value1));
+                    var success = trans.Commit();
+                    Assert.False(success);
+                    Assert.AreEqual(Redis.GetValue(Key), value1);
+                }
+            }
+            catch (NotSupportedException ignore)
+            {
+                Assert.That(Redis.GetValue(Key), Is.Null);
+            }
+        }
+
 		[Test]
 		public void Exception_in_atomic_transactions_discards_all_commands()
 		{
@@ -153,6 +177,29 @@ public void Can_call_multi_string_operations_in_transaction()
 			Assert.That(item4, Is.Null);
 		}
         [Test]
+        public void Can_call_multiple_setexs_in_transaction()
+        {
+            Assert.That(Redis.GetValue(Key), Is.Null);
+            var keys = new[] { "key1", "key2", "key3" };
+            var values = new[] { "1", "2", "3" };
+            var trans = Redis.CreateTransaction();
+
+            for (int i = 0; i < 3; ++i)
+            {
+                int index0 = i;
+                trans.QueueCommand(r => ((RedisNativeClient)r).SetEx(keys[index0], 100, GetBytes(values[index0])));
+            }
+
+            trans.Commit();
+            trans.Replay();
+
+
+            for (int i = 0; i < 3; ++i)
+                Assert.AreEqual(Redis.GetValue(keys[i]), values[i]);
+
+            trans.Dispose();
+        }
+        [Test]
         // Operations that are not supported in older versions will look at server info to determine what to do.
         // If server info is fetched each time, then it will interfer with transaction
         public void Can_call_operation_not_supported_on_older_servers_in_transaction()

Original file line number	Diff line number	Diff line change
`@@ -35,63 +35,79 @@ protected virtual void AddCurrentQueuedOperation()`
`35`	`35`
`36`	`36`	`public virtual void CompleteVoidQueuedCommand(Action voidReadCommand)`
`37`	`37`	`{`
`38`		`- AssertCurrentOperation();`
	`38`	`+ //AssertCurrentOperation();`
	`39`	`+ // this can happen when replaying pipeline/transaction`
	`40`	`+ if (CurrentQueuedOperation == null) return;`
`39`	`41`
`40`	`42`	`CurrentQueuedOperation.VoidReadCommand = voidReadCommand;`
`41`	`43`	`AddCurrentQueuedOperation();`
`42`	`44`	`}`
`43`	`45`
`44`	`46`	`public virtual void CompleteIntQueuedCommand(Func<int> intReadCommand)`
`45`	`47`	`{`
`46`		`- AssertCurrentOperation();`
	`48`	`+ //AssertCurrentOperation();`
	`49`	`+ // this can happen when replaying pipeline/transaction`
	`50`	`+ if (CurrentQueuedOperation == null) return;`
`47`	`51`
`48`	`52`	`CurrentQueuedOperation.IntReadCommand = intReadCommand;`
`49`	`53`	`AddCurrentQueuedOperation();`
`50`	`54`	`}`
`51`	`55`
`52`	`56`	`public virtual void CompleteLongQueuedCommand(Func<long> longReadCommand)`
`53`	`57`	`{`
`54`		`- AssertCurrentOperation();`
	`58`	`+ //AssertCurrentOperation();`
	`59`	`+ // this can happen when replaying pipeline/transaction`
	`60`	`+ if (CurrentQueuedOperation == null) return;`
`55`	`61`
`56`	`62`	`CurrentQueuedOperation.LongReadCommand = longReadCommand;`
`57`	`63`	`AddCurrentQueuedOperation();`
`58`	`64`	`}`
`59`	`65`
`60`	`66`	`public virtual void CompleteBytesQueuedCommand(Func<byte[]> bytesReadCommand)`
`61`	`67`	`{`
`62`		`- AssertCurrentOperation();`
	`68`	`+ //AssertCurrentOperation();`
	`69`	`+ // this can happen when replaying pipeline/transaction`
	`70`	`+ if (CurrentQueuedOperation == null) return;`
`63`	`71`
`64`	`72`	`CurrentQueuedOperation.BytesReadCommand = bytesReadCommand;`
`65`	`73`	`AddCurrentQueuedOperation();`
`66`	`74`	`}`
`67`	`75`
`68`	`76`	`public virtual void CompleteMultiBytesQueuedCommand(Func<byte[][]> multiBytesReadCommand)`
`69`	`77`	`{`
`70`		`- AssertCurrentOperation();`
	`78`	`+ //AssertCurrentOperation();`
	`79`	`+ // this can happen when replaying pipeline/transaction`
	`80`	`+ if (CurrentQueuedOperation == null) return;`
`71`	`81`
`72`	`82`	`CurrentQueuedOperation.MultiBytesReadCommand = multiBytesReadCommand;`
`73`	`83`	`AddCurrentQueuedOperation();`
`74`	`84`	`}`
`75`	`85`
`76`	`86`	`public virtual void CompleteStringQueuedCommand(Func<string> stringReadCommand)`
`77`	`87`	`{`
`78`		`- AssertCurrentOperation();`
	`88`	`+ //AssertCurrentOperation();`
	`89`	`+ // this can happen when replaying pipeline/transaction`
	`90`	`+ if (CurrentQueuedOperation == null) return;`
`79`	`91`
`80`	`92`	`CurrentQueuedOperation.StringReadCommand = stringReadCommand;`
`81`	`93`	`AddCurrentQueuedOperation();`
`82`	`94`	`}`
`83`	`95`
`84`	`96`	`public virtual void CompleteMultiStringQueuedCommand(Func<List<string>> multiStringReadCommand)`
`85`	`97`	`{`
`86`		`- AssertCurrentOperation();`
	`98`	`+ //AssertCurrentOperation();`
	`99`	`+ // this can happen when replaying pipeline/transaction`
	`100`	`+ if (CurrentQueuedOperation == null) return;`
`87`	`101`
`88`	`102`	`CurrentQueuedOperation.MultiStringReadCommand = multiStringReadCommand;`
`89`	`103`	`AddCurrentQueuedOperation();`
`90`	`104`	`}`
`91`	`105`
`92`	`106`	`public virtual void CompleteDoubleQueuedCommand(Func<double> doubleReadCommand)`
`93`	`107`	`{`
`94`		`- AssertCurrentOperation();`
	`108`	`+ //AssertCurrentOperation();`
	`109`	`+ // this can happen when replaying pipeline/transaction`
	`110`	`+ if (CurrentQueuedOperation == null) return;`
`95`	`111`
`96`	`112`	`CurrentQueuedOperation.DoubleReadCommand = doubleReadCommand;`
`97`	`113`	`AddCurrentQueuedOperation();`