add a kind of reference count for sequential queue lock: when all items have marked as processed,

call Unlock() method.

Author: Aaron Boxer

src/ServiceStack.Redis/Support/Queue/Implementation/RedisSequentialWorkQueue.Locks.cs

@@ -11,38 +11,24 @@ namespace ServiceStack.Redis.Support.Queue.Implementation
     /// 
     /// 
     /// </summary>
-    public partial class RedisSequentialWorkQueue<T> : RedisWorkQueue<T>, ISequentialWorkQueue<T> where T : class
+    public partial class RedisSequentialWorkQueue<T> 
     {
-        public class DequeueLockFactory : DistributedLockFactory
-        {
-            protected readonly RedisSequentialWorkQueue<T> workQueue;
-            protected readonly string workItemId;
-            protected readonly PooledRedisClientManager clientManager;
-            public DequeueLockFactory(IRedisClient client, PooledRedisClientManager clientManager, RedisSequentialWorkQueue<T> workQueue, string workItemId)
-                : base(client)
-            {
-                this.clientManager = clientManager;
-                this.workQueue = workQueue;
-                this.workItemId = workItemId;
-            }
-            public override IDistributedLock CreateLock()
-            {
-                return new DequeueLock(client, clientManager, workQueue, workItemId);
-            }
-        }
 
         public class DequeueLock : DistributedLock
         {
             private bool ownsClient;
             protected readonly RedisSequentialWorkQueue<T> workQueue;
             protected readonly string workItemId;
             protected readonly PooledRedisClientManager clientManager;
-            public DequeueLock(IRedisClient client,    PooledRedisClientManager clientManager, RedisSequentialWorkQueue<T> workQueue, string workItemId) : base(client)
+            protected readonly int numberOfDequeuedItems;
+            protected int numberOfProcessedItems;
+            public DequeueLock(IRedisClient client,    PooledRedisClientManager clientManager, RedisSequentialWorkQueue<T> workQueue, string workItemId, int numberOfDequeuedItems) : base(client)
             {
                 this.workQueue = workQueue;
                 this.workItemId = workItemId;
                 this.clientManager = clientManager;
                 ownsClient = false;
+                this.numberOfDequeuedItems = numberOfDequeuedItems;
             }
 
             public override long Lock(string key, int acquisitionTimeout, int lockTimeout)
@@ -53,6 +39,13 @@ public override long Lock(string key, int acquisitionTimeout, int lockTimeout)
                 return rc;
             }
 
+            public void DoneProcessedWorkItem()
+            {
+                numberOfProcessedItems++;
+                if (numberOfProcessedItems == numberOfDequeuedItems)
+                    Unlock();
+            }
+
             public override bool Unlock()
             {
                 workQueue.Unlock(workItemId);
@@ -80,16 +73,15 @@ protected override RedisClient AcquireClient()
         }
         public class DeferredDequeueLock : DequeueLock
         {
-            private readonly int numberofPeekedItems;
-            public DeferredDequeueLock(IRedisClient client, PooledRedisClientManager clientManager, RedisSequentialWorkQueue<T> workQueue, string workItemId, int numberofPeekedItems)
-                                                                       :base(client, clientManager, workQueue, workItemId)
+            public DeferredDequeueLock(IRedisClient client, PooledRedisClientManager clientManager, RedisSequentialWorkQueue<T> workQueue, string workItemId, int numberOfDequeuedItems) 
+                : base(client, clientManager, workQueue, workItemId, numberOfDequeuedItems)
             {
-                this.numberofPeekedItems = numberofPeekedItems;
             }
+
             public override bool Unlock()
             {
                 //remove items from queue
-                workQueue.Pop(workItemId, numberofPeekedItems);
+                workQueue.Pop(workItemId, numberOfDequeuedItems);
 
                 // unlock work queue id
                 workQueue.Unlock(workItemId);

src/ServiceStack.Redis/Support/Queue/Implementation/RedisSequentialWorkQueue.cs

@@ -74,7 +74,7 @@ private SequentialData<T> DequeueImpl(int maxBatchSize, bool defer)
                 string workItemId = null;
                 var dequeueItems = new List<T>();
                 var smallest = client.ZRangeWithScores(pendingWorkItemIdQueue, 0, 0);
-                IDistributedLock workItemIdLock = null;
+                DequeueLock workItemIdLock = null;
                 try
                 {
                     if (smallest != null && smallest.Length > 1 && RedisNativeClient.ParseDouble(smallest[1]) != CONVENIENTLY_SIZED_FLOAT)
@@ -118,7 +118,7 @@ private SequentialData<T> DequeueImpl(int maxBatchSize, bool defer)
                         }
                         workItemIdLock = defer ?   
                             new DeferredDequeueLock(client, clientManager, this, workItemId, dequeueItems.Count) :
-                                new DequeueLock(client, clientManager, this, workItemId);
+                                new DequeueLock(client, clientManager, this, workItemId, dequeueItems.Count);
                         var dequeueLockKey = queueNamespace.GlobalKey(workItemId, numTagsForDequeueLock);
                         workItemIdLock.Lock(dequeueLockKey, lockAcquisitionTimeout, dequeueLockTimeout);
 

src/ServiceStack.Redis/Support/Queue/SequentialData.cs

@@ -1,5 +1,6 @@
 using System.Collections.Generic;
 using ServiceStack.Redis.Support.Locking;
+using ServiceStack.Redis.Support.Queue.Implementation;
 
 namespace ServiceStack.Redis.Support.Queue
 {
@@ -13,7 +14,7 @@ public IList<T> WorkItems
         {
             get; set;
         }
-        public IDistributedLock WorkItemIdLock
+        public RedisSequentialWorkQueue<T>.DequeueLock WorkItemIdLock
         {
             get;set;
         }

tests/ServiceStack.Redis.Tests/QueueTests.cs

@@ -48,9 +48,11 @@ public void TestSequentialWorkQueue()
                     // check that batch size is respected
                     Assert.AreEqual(batch.WorkItems.Count, numMessages);
                     for (int i = 0; i < numMessages; ++i)
+                    {
                         Assert.AreEqual(batch.WorkItems[i], messages1[i]);
-                    batch.WorkItemIdLock.Unlock();
-
+                        batch.WorkItemIdLock.DoneProcessedWorkItem();
+                    }
+               
 
                     // check that there are numMessages/2 messages in the queue
                     batch = queue.Dequeue(numMessages, defer);