Merge pull request #57 from kanshi255/master

Fixed PostgreSqlJobQueue.Dequeue queue prioritization

Author: vytautask

src/Hangfire.PostgreSql.NetCore/PostgreSqlJobQueue.cs

@@ -71,10 +71,15 @@ public IFetchedJob Dequeue_Transaction(string[] queues, CancellationToken cancel
 SET ""fetchedat"" = NOW() AT TIME ZONE 'UTC'
 WHERE ""id"" IN (
     SELECT ""id"" 
-    FROM """ + _options.SchemaName + @""".""jobqueue"" 
+    FROM """ + _options.SchemaName + $@""".""jobqueue"" 
     WHERE ""queue"" = ANY (@queues)
-    AND ""fetchedat"" {0} 
-    ORDER BY ""fetchedat"", ""jobid""
+    AND ""fetchedat"" {{0}}
+    ORDER BY
+    CASE ""queue""
+        {string.Join("\n", queues.Select((q, i) => $"WHEN '{q}' THEN {i}"))}
+        ELSE {queues.Length}
+    END,
+    ""fetchedat"", ""jobid""
     LIMIT 1
 )
 RETURNING ""id"" AS ""Id"", ""jobid"" AS ""JobId"", ""queue"" AS ""Queue"", ""fetchedat"" AS ""FetchedAt"";
@@ -157,10 +162,15 @@ public IFetchedJob Dequeue_UpdateCount(string[] queues, CancellationToken cancel
 
 			string jobToFetchSqlTemplate = @"
 SELECT ""id"" AS ""Id"", ""jobid"" AS ""JobId"", ""queue"" AS ""Queue"", ""fetchedat"" AS ""FetchedAt"", ""updatecount"" AS ""UpdateCount""
-FROM """ + _options.SchemaName + @""".""jobqueue"" 
+FROM """ + _options.SchemaName + $@""".""jobqueue"" 
 WHERE ""queue"" = ANY (@queues)
-AND ""fetchedat"" {0} 
-ORDER BY ""fetchedat"", ""jobid"" 
+AND ""fetchedat"" {{0}} 
+ORDER BY
+CASE ""queue""
+    {string.Join("\n", queues.Select((q, i) => $"WHEN '{q}' THEN {i}"))}
+    ELSE {queues.Length}
+END,
+""fetchedat"", ""jobid"" 
 LIMIT 1;
 ";
 

tests/Hangfire.PostgreSql.NetCore.Tests/PostgreSqlJobQueueFacts.cs

@@ -46,7 +46,25 @@ public void Dequeue_ShouldThrowAnException_WhenQueuesCollectionIsNull()
 			});
 		}
 
-		[Fact, CleanDatabase]
+        [Fact, CleanDatabase]
+        public void Dequeue_ShouldFetchAJob_FromQueueWithHigherPriority()
+        {
+            UseConnection(connection =>
+            {
+                var queue = CreateJobQueue(connection, false);
+                var token = CreateTimingOutCancellationToken();
+
+                queue.Enqueue("1", "1");
+                queue.Enqueue("2", "2");
+                queue.Enqueue("3", "3");
+
+                Assert.Equal("1", queue.Dequeue(new[] { "1", "2", "3" }, token).JobId);
+                Assert.Equal("2", queue.Dequeue(new[] { "2", "3", "1" }, token).JobId);
+                Assert.Equal("3", queue.Dequeue(new[] { "3", "1", "2" }, token).JobId);
+            });
+        }
+
+        [Fact, CleanDatabase]
 		private void Dequeue_ShouldThrowAnException_WhenQueuesCollectionIsEmpty_WithUseNativeDatabaseTransactions()
 		{
 			Dequeue_ShouldThrowAnException_WhenQueuesCollectionIsEmpty(true);