Fixed queue to utilize fetchToken and fixes #6

arnoldasgudas · Oct 27, 2016 · 7031c05 · 7031c05
1 parent 31e1c58
commit 7031c05
Show file tree

Hide file tree

Showing 2 changed files with 41 additions and 19 deletions.
diff --git a/Hangfire.MySql/JobQueue/MySqlFetchedJob.cs b/Hangfire.MySql/JobQueue/MySqlFetchedJob.cs
@@ -14,6 +14,9 @@ internal class MySqlFetchedJob : IFetchedJob
         private readonly MySqlStorage _storage;
         private readonly IDbConnection _connection;
         private readonly int _id;
+        private bool _removedFromQueue;
+        private bool _requeued;
+        private bool _disposed;
 
         public MySqlFetchedJob(
             MySqlStorage storage, 
@@ -33,7 +36,17 @@ internal class MySqlFetchedJob : IFetchedJob
 
         public void Dispose()
         {
+
+            if (_disposed) return;
+
+            if (!_removedFromQueue && !_requeued)
+            {
+                Requeue();
+            }
+
             _storage.ReleaseConnection(_connection);
+
+            _disposed = true;
         }
 
         public void RemoveFromQueue()
@@ -48,6 +61,8 @@ public void RemoveFromQueue()
                 {
                     id = _id
                 });
+
+            _removedFromQueue = true;
         }
 
         public void Requeue()
@@ -62,6 +77,7 @@ public void Requeue()
                 {
                     id = _id
                 });
+            _requeued = true;
         }
 
         public string JobId { get; private set; }

diff --git a/Hangfire.MySql/JobQueue/MySqlJobQueue.cs b/Hangfire.MySql/JobQueue/MySqlJobQueue.cs
@@ -43,27 +43,33 @@ public IFetchedJob Dequeue(string[] queues, CancellationToken cancellationToken)
                 {
                     using (new MySqlDistributedLock(_storage, "JobQueue", TimeSpan.FromSeconds(30)))
                     {
-                        fetchedJob =
-                        connection
-                            .Query<FetchedJob>(
-                                "select Id, JobId, Queue " +
-                                "from JobQueue " +
-                                "where (FetchedAt is null or FetchedAt < DATE_ADD(UTC_TIMESTAMP(), INTERVAL @timeout SECOND)) " +
-                                "   and Queue in @queues " +
-                                "limit 1;",
-                                new
-                                {
-                                    queues = queues,
-                                    timeout = _options.InvisibilityTimeout.Negate().TotalSeconds
-                                })
-                            .SingleOrDefault();
+                        string token = Guid.NewGuid().ToString();
 
-                        if (fetchedJob != null)
+                        int nUpdated = connection.Execute(
+                            "update JobQueue set FetchedAt = UTC_TIMESTAMP(), FetchToken = @fetchToken " +
+                            "where (FetchedAt is null or FetchedAt < DATE_ADD(UTC_TIMESTAMP(), INTERVAL @timeout SECOND)) " +
+                            "   and Queue in @queues " +
+                            "LIMIT 1;",
+                            new
+                            {
+                                queues = queues,
+                                timeout = _options.InvisibilityTimeout.Negate().TotalSeconds,
+                                fetchToken = token
+                            });
+
+                        if(nUpdated != 0)
                         {
-                            connection
-                                .Execute(
-                                    "update JobQueue set FetchedAt = @fetchedAt where Id = @id",
-                                    new {fetchedAt = DateTime.UtcNow, id = fetchedJob.Id});
+                            fetchedJob =
+                                connection
+                                    .Query<FetchedJob>(
+                                        "select Id, JobId, Queue " +
+                                        "from JobQueue " +
+                                        "where FetchToken = @fetchToken;",
+                                        new
+                                        {
+                                            fetchToken = token
+                                        })
+                                    .SingleOrDefault();
                         }
                     }
                 }