TPL migration

Author: danielgerlag

src/WorkflowCore/Interface/IDistributedLockProvider.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace WorkflowCore.Interface
@@ -11,7 +12,7 @@ namespace WorkflowCore.Interface
     /// </remarks>
     public interface IDistributedLockProvider
     {
-        Task<bool> AcquireLock(string Id);
+        Task<bool> AcquireLock(string Id, CancellationToken cancellationToken);
 
         Task ReleaseLock(string Id);
 

src/WorkflowCore/Interface/IQueueProvider.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using WorkflowCore.Models;
 
@@ -25,7 +26,9 @@ public interface IQueueProvider : IDisposable
         /// If the queue is empty, NULL is returned
         /// </summary>
         /// <returns></returns>
-        Task<string> DequeueWork(QueueType queue);                
+        Task<string> DequeueWork(QueueType queue, CancellationToken cancellationToken);
+
+        bool IsDequeueBlocking { get; }
 
         Task Start();
         Task Stop();

src/WorkflowCore/Models/WorkflowOptions.cs

@@ -20,7 +20,7 @@ public WorkflowOptions()
         {
             //set defaults
             PollInterval = TimeSpan.FromSeconds(10);
-            IdleTime = TimeSpan.FromMilliseconds(500);
+            IdleTime = TimeSpan.FromMilliseconds(100);
             ErrorRetryInterval = TimeSpan.FromSeconds(60);            
 
             QueueFactory = new Func<IServiceProvider, IQueueProvider>(sp => new SingleNodeQueueProvider());

src/WorkflowCore/Models/WorkflowStep.cs

@@ -65,7 +65,6 @@ public virtual IStepBody ConstructBody(IServiceProvider serviceProvider)
             }
             return body;
         }
-
     }
 
     public enum ExecutionPipelineDirective { Next = 0, Defer = 1, EndWorkflow = 2 }

src/WorkflowCore/ServiceCollectionExtensions.cs

@@ -24,8 +24,8 @@ public static void AddWorkflow(this IServiceCollection services, Action<Workflow
             services.AddSingleton<IWorkflowRegistry, WorkflowRegistry>();
             services.AddSingleton<WorkflowOptions>(options);
 
-            services.AddTransient<IBackgroundTask, WorkflowTask>();
-            services.AddTransient<IBackgroundTask, EventTask>();
+            services.AddTransient<IBackgroundTask, WorkflowTaskDispatcher>();
+            services.AddTransient<IBackgroundTask, EventTaskDispatcher>();
             services.AddTransient<IBackgroundTask, RunnablePoller>();
 
             services.AddSingleton<IWorkflowHost, WorkflowHost>();

src/WorkflowCore/Services/EventTask.cs

@@ -0,0 +1,97 @@
+using Microsoft.Extensions.Logging;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Services
+{
+    class EventTaskDispatcher : QueueTaskDispatcher, IBackgroundTask
+    {
+        private readonly IPersistenceProvider _persistenceStore;
+        private readonly IDistributedLockProvider _lockProvider;
+        private readonly IDateTimeProvider _datetimeProvider;
+
+        protected override QueueType Queue => QueueType.Event;
+
+        public EventTaskDispatcher(IPersistenceProvider persistenceStore, IQueueProvider queueProvider, ILoggerFactory loggerFactory, IServiceProvider serviceProvider, IWorkflowRegistry registry, IDistributedLockProvider lockProvider, WorkflowOptions options, IDateTimeProvider datetimeProvider)
+            : base(queueProvider, loggerFactory, options)
+        {
+            _persistenceStore = persistenceStore;
+            _lockProvider = lockProvider;
+            _datetimeProvider = datetimeProvider;
+        }
+
+        protected override async Task ProcessItem(string itemId, CancellationToken cancellationToken)
+        {
+            if (await _lockProvider.AcquireLock($"evt:{itemId}", cancellationToken))
+            {
+                try
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+                    var evt = await _persistenceStore.GetEvent(itemId);
+                    if (evt.EventTime <= _datetimeProvider.Now.ToUniversalTime())
+                    {
+                        var subs = await _persistenceStore.GetSubcriptions(evt.EventName, evt.EventKey, evt.EventTime);
+                        var success = true;
+
+                        foreach (var sub in subs.ToList())
+                            success = success && await SeedSubscription(evt, sub, cancellationToken);
+
+                        if (success)
+                            await _persistenceStore.MarkEventProcessed(itemId);
+                    }
+                }
+                finally
+                {
+                    await _lockProvider.ReleaseLock($"evt:{itemId}");
+                }
+            }
+            else
+            {
+                Logger.LogInformation($"Event locked {itemId}");
+            }
+        }
+        
+        private async Task<bool> SeedSubscription(Event evt, EventSubscription sub, CancellationToken cancellationToken)
+        {
+            if (await _lockProvider.AcquireLock(sub.WorkflowId, cancellationToken))
+            {
+                try
+                {
+                    var workflow = await _persistenceStore.GetWorkflowInstance(sub.WorkflowId);
+                    var pointers = workflow.ExecutionPointers.Where(p => p.EventName == sub.EventName && p.EventKey == sub.EventKey && !p.EventPublished);
+                    foreach (var p in pointers)
+                    {
+                        p.EventData = evt.EventData;
+                        p.EventPublished = true;
+                        p.Active = true;
+                    }
+                    workflow.NextExecution = 0;
+                    await _persistenceStore.PersistWorkflow(workflow);
+                    await _persistenceStore.TerminateSubscription(sub.Id);
+                    return true;
+                }
+                catch (Exception ex)
+                {
+                    Logger.LogError(ex.Message);
+                    return false;
+                }
+                finally
+                {
+                    await _lockProvider.ReleaseLock(sub.WorkflowId);
+                    await QueueProvider.QueueWork(sub.WorkflowId, QueueType.Workflow);
+                }
+            }
+            else
+            {
+                Logger.LogInformation("Workflow locked {0}", sub.WorkflowId);
+                return false;
+            }
+        }
+    }
+}