Merge pull request #189 from JPVenson/master

[Feature] Added method for fetching a new job in the job queue

Author: frankhommers

src/Hangfire.PostgreSql/PostgreSqlJobQueue.cs

@@ -37,6 +37,14 @@ public class PostgreSqlJobQueue : IPersistentJobQueue
         private readonly PostgreSqlStorageOptions _options;
         private readonly PostgreSqlStorage _storage;
 
+		private AutoResetEvent SignalDequeue { get; }
+
+		public PostgreSqlJobQueue(PostgreSqlStorage storage, PostgreSqlStorageOptions options)
+		{
+			_options = options ?? throw new ArgumentNullException(nameof(options));
+			_storage = storage ?? throw new ArgumentNullException(nameof(storage));
+			SignalDequeue = new AutoResetEvent(false);
+		}
         public PostgreSqlJobQueue(PostgreSqlStorage storage, PostgreSqlStorageOptions options)
         {
             _options = options ?? throw new ArgumentNullException(nameof(options));
@@ -50,8 +58,13 @@ public IFetchedJob Dequeue(string[] queues, CancellationToken cancellationToken)
             if (_options.UseNativeDatabaseTransactions)
                 return Dequeue_Transaction(queues, cancellationToken);
 
-            return Dequeue_UpdateCount(queues, cancellationToken);
-        }
+		/// <summary>
+		///		Signal the waiting Thread to lookup a new Job
+		/// </summary>
+		public void FetchNextJob()
+		{
+			SignalDequeue.Set();
+		}
 
         public void Enqueue(IDbConnection connection, string queue, string jobId)
         {
@@ -149,7 +162,8 @@ LIMIT 1
                     WaitHandle.WaitAny(new[]
                         {
                             cancellationToken.WaitHandle, 
-                            NewItemInQueueEvent
+                            NewItemInQueueEvent,
+	                        SignalDequeue
                         },
                         _options.QueuePollInterval);
 
@@ -253,4 +267,4 @@ private class FetchedJob
             public int UpdateCount { get; set; }
         }
     }
-}
+}

tests/Hangfire.PostgreSql.Tests/PostgreSqlJobQueueFacts.cs

@@ -3,7 +3,9 @@
 using System.Globalization;
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
 using Dapper;
+using Hangfire.Storage;
 using Npgsql;
 using Xunit;
 
@@ -442,6 +444,37 @@ public void Queues_Should_Support_Long_Queue_Names()
 				Assert.Equal(name, record.queue.ToString());
 			});
 		}
+		
+		[Fact, CleanDatabase]
+		public void Queues_Can_Dequeue_On_Signal()
+		{
+			UseConnection((connection, storage) =>
+			{
+				var queue = CreateJobQueue(storage, false);
+				IFetchedJob job = null;
+				//as UseConnection does not support async-await we have to work with Thread.Sleep
+
+				Task.Run(() =>
+				{
+					//dequeue the job asynchronously
+					job = queue.Dequeue(new[] {"default"}, CreateTimingOutCancellationToken());
+				});
+				//all sleeps are possibly way to high but this ensures that any race condition is unlikely
+				//to ensure that the task would run 
+				Thread.Sleep(1000);
+				Assert.Null(job);
+				//enqueue a job that does not trigger the existing queue to reevaluate its state
+				queue.Enqueue(connection, "default", "1");
+				Thread.Sleep(1000);
+				//the job should still be unset
+				Assert.Null(job);
+				//trigger a reevaluation
+				queue.FetchNextJob();
+				//wait for the Dequeue to execute and return the next job
+				Thread.Sleep(1000);
+				Assert.NotNull(job);
+			});
+		}
 
 		private void Enqueue_AddsAJobToTheQueue(bool useNativeDatabaseTransactions)
 		{