small tweaks to work queues

Author: Aaron Boxer

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

@@ -12,12 +12,6 @@ public interface ISimpleWorkQueue<T> : IDisposable where T : class
         /// <param name="workItem"></param>
 		void Enqueue(T workItem);
 
-        /// <summary>
-        /// Push items to front of queue
-        /// </summary>
-        /// <param name="workItems"></param>
-        void PushFront(IList<T> workItems);
-
         /// <summary>
         /// Dequeue up to maxBatchSize items from queue
         /// </summary>

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

@@ -15,7 +15,7 @@ public class RedisSequentialWorkQueue<T> : RedisWorkQueue<T>, ISequentialWorkQue
     {
         private int lockAcquisitionTimeout = 2;
         private int lockTimeout = 2;
-        private double  CONVENIENTLY_SIZED_FLOAT = 18014398509481984.0;
+        protected const double  CONVENIENTLY_SIZED_FLOAT = 18014398509481984.0;
 
 
 
@@ -24,7 +24,7 @@ public RedisSequentialWorkQueue(int maxReadPoolSize, int maxWritePoolSize, strin
         {
         }
 
-        public RedisSequentialWorkQueue(int maxReadPoolSize, int maxWritePoolSize, string host, int port, string queueName ) 
+        public RedisSequentialWorkQueue(int maxReadPoolSize, int maxWritePoolSize, string host, int port, string queueName) 
             : base(maxReadPoolSize, maxWritePoolSize, host, port, queueName)
         {
         }
@@ -53,33 +53,6 @@ public void Enqueue(string workItemId, T workItem)
             }
         }
 
-        /// <summary>
-        /// Return dequeued items to front of queue 
-        /// </summary>
-        /// <param name="workItems"></param>
-        /// <param name="workItemId"></param>
-        public void PushFront(string workItemId, IList<T> workItems)
-        {
-            if (workItems == null || workItems.Count == 0)
-                return;
-            using (var disposableClient = clientManager.GetDisposableClient<SerializingRedisClient>())
-            {
-                var client = disposableClient.Client;
-                var key = queueNamespace.GlobalCacheKey(workItemId);
-                using (var pipe = client.CreatePipeline())
-                {
-                    for (int i = workItems.Count - 1; i >= 0; i--)
-                    {
-                        int index = i;
-                        pipe.QueueCommand(r => ((RedisNativeClient)r).LPush(key, client.Serialize(workItems[index])));
-                    }
-                    pipe.Flush();
-
-                }
-                PostDequeue(workItemId);
-            }
-        }
-        
         /// <summary>
         /// Dequeue next batch of messages for processing. After this method is called,
         /// no other messages with same work item id will be available for
@@ -123,6 +96,37 @@ public KeyValuePair<string, IList<T>> Dequeue(int maxBatchSize)
             }
         }
 
+        /// <summary>
+        /// Return dequeued items to front of queue 
+        /// </summary>
+        /// <param name="workItems"></param>
+        /// <param name="workItemId"></param>
+        public void PushFront(string workItemId, IList<T> workItems)
+        {
+            if (workItems == null || workItems.Count == 0)
+                return;
+            using (var disposableClient = clientManager.GetDisposableClient<SerializingRedisClient>())
+            {
+                var client = disposableClient.Client;
+                var key = queueNamespace.GlobalCacheKey(workItemId);
+                var lockKey = queueNamespace.GlobalKey(workItemId, RedisNamespace.NumTagsForLockKey);
+                using (var disposableLock = new DisposableDistributedLock(client, lockKey, lockAcquisitionTimeout, lockTimeout))
+                {
+                    using (var pipe = client.CreatePipeline())
+                    {
+                        for (int i = workItems.Count - 1; i >= 0; i--)
+                        {
+                            int index = i;
+                            pipe.QueueCommand(
+                                r => ((RedisNativeClient) r).LPush(key, client.Serialize(workItems[index])));
+                        }
+                        pipe.QueueCommand(r => ((RedisNativeClient)r).ZIncrBy(pendingWorkItemIdQueue, -1, client.Serialize(workItemId)));
+                        pipe.Flush();
+                    }
+                }
+            }
+        }
+        
         /// <summary>
         /// Unlock message id, so other servers can process messages for this message id
         /// </summary>

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

@@ -35,14 +35,6 @@ public void Enqueue(T msg)
                 client.RPush(key, client.Serialize(msg));
             }
         }
-        /// <summary>
-        /// Return dequeued items to front of queue 
-        /// </summary>
-        /// <param name="workItems"></param>
-        public void PushFront(IList<T> workItems)
-        {
-            UnDequeueImpl(workItems, pendingWorkItemIdQueue);
-        }
 
 
         /// <summary>

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

@@ -35,34 +35,6 @@ public RedisWorkQueue(int maxReadPoolSize, int maxWritePoolSize, string host, in
                                 };
         }
 
-
-        /// <summary>
-        /// Return dequeued items to front of queue 
-        /// </summary>
-        /// <param name="workItems"></param>
-        /// <param name="listId"></param>
-        protected void UnDequeueImpl(IList<T> workItems, string listId)
-        {
-            if (workItems == null || workItems.Count == 0)
-                return;
-            using (var disposableClient = clientManager.GetDisposableClient<SerializingRedisClient>())
-            {
-                var client = disposableClient.Client;
-                var key = queueNamespace.GlobalCacheKey(listId);
-                using (var pipe = client.CreatePipeline())
-                {
-                    for (int i = workItems.Count - 1; i >= 0; i--)
-                    {
-                        int index = i;
-                        pipe.QueueCommand(r => ((RedisNativeClient)r).LPush(key, client.Serialize(workItems[index])));
-                    }
-                    pipe.Flush();
-
-                }
-            }
-        }
-
-
         public void Dispose()
         {
             clientManager.Dispose();

tests/ServiceStack.Redis.Tests/QueueTests.cs

@@ -8,6 +8,7 @@ namespace ServiceStack.Redis.Tests
     [TestFixture]
     public class QueueTests
     {
+    
         [Test]
         public void TestSequentialWorkQueue()
         {
@@ -127,11 +128,6 @@ public void TestSimpleWorkQueue()
                 batch = queue.Dequeue(numMessages * 2);
                 Assert.AreEqual(batch.Count, 0);
 
-                // test that UnDequeue works
-                queue.PushFront(batch);
-                var undequeuedBatch = queue.Dequeue(numMessages*2);
-                Assert.AreEqual(undequeuedBatch, batch);
-
             }
         }