added ability to re-queue items at front of queue. Useful

for when processing hits an error condition on one of the
queue items, and needs to pause the processing

Author: Aaron Boxer

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

@@ -5,10 +5,34 @@ namespace ServiceStack.Redis.Support.Queue
 {
 	public interface ISequentialWorkQueue<T> : IDisposable where T : class
 	{
+        /// <summary>
+        /// Enqueue item in queue corresponding to workItemId identifier
+        /// </summary>
+        /// <param name="workItemId"></param>
+        /// <param name="workItem"></param>
 		void Enqueue(string workItemId, T workItem);
 
+        /// <summary>
+        /// Dequeue up to maxBatchSize items from queue corresponding to workItemId identifier.
+        /// Once this method is called, no items for workItemId queue can be dequeued until
+        /// <see cref="PostDequeue"/> is called
+        /// </summary>
+        /// <param name="maxBatchSize"></param>
+        /// <returns></returns>
 		KeyValuePair<string, IList<T>> Dequeue(int maxBatchSize);
 
+        /// <summary>
+        /// Push items to front of queue corresponding to workItemId identifier
+        /// </summary>
+        /// <param name="workItemId"></param>
+        /// <param name="workItems"></param>
+	    void PushFront(string workItemId, IList<T> workItems);
+
+        /// <summary>
+        ///  Must be called after dequeued items have been processed, in order for other callers
+        /// to dequeue items from workItemId queue
+        /// </summary>
+        /// <param name="workItemId"></param>
 		void PostDequeue(string workItemId);
 	}
 }

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

@@ -6,10 +6,23 @@ namespace ServiceStack.Redis.Support.Queue
 {
 	public interface ISimpleWorkQueue<T> : IDisposable where T : class
 	{
+        /// <summary>
+        /// Enqueue item
+        /// </summary>
+        /// <param name="workItem"></param>
 		void Enqueue(T workItem);
 
-        void UnDequeue(IList<T> workItems);
+        /// <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>
+        /// <param name="maxBatchSize"></param>
+        /// <returns></returns>
 		IList<T> Dequeue(int maxBatchSize);
 	}
 }

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

@@ -51,7 +51,33 @@ 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>

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

@@ -35,30 +35,13 @@ public void Enqueue(T msg)
                 client.RPush(key, client.Serialize(msg));
             }
         }
-
         /// <summary>
-        /// Return dequeued items to front of list
+        /// Return dequeued items to front of queue 
         /// </summary>
         /// <param name="workItems"></param>
-        public void UnDequeue(IList<T> workItems)
+        public void PushFront(IList<T> workItems)
         {
-            if (workItems == null || workItems.Count == 0)
-                return;
-            using (var disposableClient = clientManager.GetDisposableClient<SerializingRedisClient>())
-            {
-                var client = disposableClient.Client;
-                var key = queueNamespace.GlobalCacheKey(pendingWorkItemIdQueue);
-                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();
-
-                }
-            }
+            UnDequeueImpl(workItems, pendingWorkItemIdQueue);
         }
 
 

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

@@ -35,6 +35,34 @@ 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

@@ -48,11 +48,19 @@ public void TestSequentialWorkQueue()
                 Assert.AreEqual(batch.Value.Count, numMessages/2);
                 queue.PostDequeue(batch.Key);
 
+                // test that PushFront works
+                queue.PushFront(batch.Key, batch.Value);
+                var undequeuedBatch = queue.Dequeue(numMessages * 2);
+                Assert.AreEqual(undequeuedBatch.Value, batch.Value);
+                Assert.AreEqual(undequeuedBatch.Key, batch.Key);
+                queue.PostDequeue(batch.Key);
+
+
                 // check that there are no more messages in the queue
                 batch = queue.Dequeue(numMessages);
                 Assert.IsNull(batch.Key);
                 Assert.AreEqual(batch.Value.Count, 0);
-             
+ 
             }
         }
 
@@ -120,7 +128,7 @@ public void TestSimpleWorkQueue()
                 Assert.AreEqual(batch.Count, 0);
 
                 // test that UnDequeue works
-                queue.UnDequeue(batch);
+                queue.PushFront(batch);
                 var undequeuedBatch = queue.Dequeue(numMessages*2);
                 Assert.AreEqual(undequeuedBatch, batch);