Workflow life cycle events (danielgerlag#215)

ReleaseNotes/1.6.9.md

@@ -0,0 +1,5 @@
+# Workflow Core 1.6.9
+
+This release adds functionality to subscribe to workflow life cycle events (WorkflowStarted, WorkflowComplete, WorkflowError, WorkflowSuspended, WorkflowResumed, StepStarted, StepCompleted, etc...)
+This can be achieved by either grabbing the `ILifeCycleEventHub` implementation from the IoC container and subscribing to events there, or attach an event on the workflow host class `IWorkflowHost.OnLifeCycleEvent`.
+This implementation only publishes events to the local node... we will still need to implement a distributed version of the EventHub to solve the problem for multi-node clusters.

WorkflowCore.sln

@@ -93,6 +93,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "ReleaseNotes", "ReleaseNote
 		ReleaseNotes\1.6.0.md = ReleaseNotes\1.6.0.md
 		ReleaseNotes\1.6.6.md = ReleaseNotes\1.6.6.md
 		ReleaseNotes\1.6.8.md = ReleaseNotes\1.6.8.md
+		ReleaseNotes\1.6.9.md = ReleaseNotes\1.6.9.md
 	EndProjectSection
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Sample14", "src\samples\WorkflowCore.Sample14\WorkflowCore.Sample14.csproj", "{6BC66637-B42A-4334-ADFB-DBEC9F29D293}"

src/WorkflowCore/Interface/IExecutionResultProcessor.cs

@@ -5,7 +5,7 @@ namespace WorkflowCore.Interface
 {
     public interface IExecutionResultProcessor
     {
-        void HandleStepException(WorkflowInstance workflow, WorkflowDefinition def, ExecutionPointer pointer, WorkflowStep step);
+        void HandleStepException(WorkflowInstance workflow, WorkflowDefinition def, ExecutionPointer pointer, WorkflowStep step, Exception exception);
         void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition def, ExecutionPointer pointer, WorkflowStep step, ExecutionResult result, WorkflowExecutorResult workflowResult);
     }
 }

src/WorkflowCore/Interface/ILifeCycleEventHub.cs

@@ -0,0 +1,14 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+using WorkflowCore.Models.LifeCycleEvents;
+
+namespace WorkflowCore.Interface
+{
+    public interface ILifeCycleEventHub
+    {
+        Task PublishNotification(LifeCycleEvent evt);
+        void Subscribe(Action<LifeCycleEvent> action);
+    }
+}

src/WorkflowCore/Interface/ILifeCycleEventPublisher.cs

@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Models.LifeCycleEvents;
+
+namespace WorkflowCore.Interface
+{
+    public interface ILifeCycleEventPublisher : IBackgroundTask
+    {
+        void PublishNotification(LifeCycleEvent evt);
+    }
+}

src/WorkflowCore/Interface/IWorkflowErrorHandler.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Interface
+{
+    public interface IWorkflowErrorHandler
+    {
+        WorkflowErrorHandling Type { get; }
+        void Handle(WorkflowInstance workflow, WorkflowDefinition def, ExecutionPointer pointer, WorkflowStep step, Exception exception, Queue<ExecutionPointer> bubbleUpQueue);
+    }
+}

src/WorkflowCore/Interface/IWorkflowHost.cs

@@ -2,6 +2,7 @@
 using System;
 using System.Threading.Tasks;
 using WorkflowCore.Models;
+using WorkflowCore.Models.LifeCycleEvents;
 
 namespace WorkflowCore.Interface
 {
@@ -19,6 +20,7 @@ public interface IWorkflowHost : IWorkflowController
 
 
         event StepErrorEventHandler OnStepError;
+        event LifeCycleEventHandler OnLifeCycleEvent;
         void ReportStepError(WorkflowInstance workflow, WorkflowStep step, Exception exception);
 
         //public dependencies to allow for extension method access
@@ -32,4 +34,5 @@ public interface IWorkflowHost : IWorkflowController
     }
 
     public delegate void StepErrorEventHandler(WorkflowInstance workflow, WorkflowStep step, Exception exception);
-}
+    public delegate void LifeCycleEventHandler(LifeCycleEvent evt);
+}

src/WorkflowCore/Models/LifeCycleEvents/LifeCycleEvent.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public abstract class LifeCycleEvent
+    {
+        public DateTime EventTimeUtc { get; set; }
+
+        public string WorkflowInsanceId { get; set; }
+
+        public string WorkflowDefinitionId { get; set; }
+
+        public int Version { get; set; }
+
+        public string Reference { get; set; }
+    }
+}

src/WorkflowCore/Models/LifeCycleEvents/StepCompleted.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class StepCompleted : LifeCycleEvent
+    {
+        public string ExecutionPointerId { get; set; }
+
+        public int StepId { get; set; }
+    }
+}

src/WorkflowCore/Models/LifeCycleEvents/StepStarted.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class StepStarted : LifeCycleEvent
+    {
+        public string ExecutionPointerId { get; set; }
+
+        public int StepId { get; set; }
+    }
+}

src/WorkflowCore/Models/LifeCycleEvents/WorkflowCompleted.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class WorkflowCompleted : LifeCycleEvent
+    {
+    }
+}

src/WorkflowCore/Models/LifeCycleEvents/WorkflowError.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class WorkflowError : LifeCycleEvent
+    {
+        public string Message { get; set; }
+
+        public string ExecutionPointerId { get; set; }
+
+        public int StepId { get; set; }
+    }
+}

src/WorkflowCore/Models/LifeCycleEvents/WorkflowResumed.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class WorkflowResumed : LifeCycleEvent
+    {
+    }
+}

src/WorkflowCore/Models/LifeCycleEvents/WorkflowStarted.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class WorkflowStarted : LifeCycleEvent
+    {
+    }
+}

src/WorkflowCore/Models/LifeCycleEvents/WorkflowSuspended.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class WorkflowSuspended : LifeCycleEvent
+    {
+    }
+}

src/WorkflowCore/Models/LifeCycleEvents/WorkflowTerminated.cs

@@ -0,0 +1,10 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class WorkflowTerminated : LifeCycleEvent
+    {
+    }
+}

src/WorkflowCore/Models/WorkflowOptions.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.DependencyInjection;
 using System;
+using Microsoft.Extensions.Logging;
 using WorkflowCore.Interface;
 using WorkflowCore.Services;
 
@@ -10,6 +11,7 @@ public class WorkflowOptions
         internal Func<IServiceProvider, IPersistenceProvider> PersistanceFactory;
         internal Func<IServiceProvider, IQueueProvider> QueueFactory;
         internal Func<IServiceProvider, IDistributedLockProvider> LockFactory;
+        internal Func<IServiceProvider, ILifeCycleEventHub> EventHubFactory;
         internal TimeSpan PollInterval;
         internal TimeSpan IdleTime;
         internal TimeSpan ErrorRetryInterval;
@@ -26,6 +28,7 @@ public WorkflowOptions(IServiceCollection services)
             QueueFactory = new Func<IServiceProvider, IQueueProvider>(sp => new SingleNodeQueueProvider());
             LockFactory = new Func<IServiceProvider, IDistributedLockProvider>(sp => new SingleNodeLockProvider());
             PersistanceFactory = new Func<IServiceProvider, IPersistenceProvider>(sp => new MemoryPersistenceProvider());
+            EventHubFactory = new Func<IServiceProvider, ILifeCycleEventHub>(sp => new SingleNodeEventHub(sp.GetService<ILoggerFactory>()));
         }
 
         public void UsePersistence(Func<IServiceProvider, IPersistenceProvider> factory)
@@ -43,6 +46,11 @@ public void UseQueueProvider(Func<IServiceProvider, IQueueProvider> factory)
             QueueFactory = factory;
         }
 
+        public void UseEventHub(Func<IServiceProvider, ILifeCycleEventHub> factory)
+        {
+            EventHubFactory = factory;
+        }
+
         public void UsePollInterval(TimeSpan interval)
         {
             PollInterval = interval;

src/WorkflowCore/ServiceCollectionExtensions.cs

@@ -11,6 +11,7 @@
 using WorkflowCore.Primitives;
 using WorkflowCore.Services.BackgroundTasks;
 using WorkflowCore.Services.DefinitionStorage;
+using WorkflowCore.Services.ErrorHandlers;
 
 namespace Microsoft.Extensions.DependencyInjection
 {
@@ -26,12 +27,20 @@ public static IServiceCollection AddWorkflow(this IServiceCollection services, A
             services.AddTransient<IPersistenceProvider>(options.PersistanceFactory);
             services.AddSingleton<IQueueProvider>(options.QueueFactory);
             services.AddSingleton<IDistributedLockProvider>(options.LockFactory);
+            services.AddSingleton<ILifeCycleEventHub>(options.EventHubFactory);
             services.AddSingleton<IWorkflowRegistry, WorkflowRegistry>();
             services.AddSingleton<WorkflowOptions>(options);
+            services.AddSingleton<ILifeCycleEventPublisher, LifeCycleEventPublisher>();
 
             services.AddTransient<IBackgroundTask, WorkflowConsumer>();
             services.AddTransient<IBackgroundTask, EventConsumer>();
             services.AddTransient<IBackgroundTask, RunnablePoller>();
+            services.AddTransient<IBackgroundTask>(sp => sp.GetService<ILifeCycleEventPublisher>());
+
+            services.AddTransient<IWorkflowErrorHandler, CompensateHandler>();
+            services.AddTransient<IWorkflowErrorHandler, RetryHandler>();
+            services.AddTransient<IWorkflowErrorHandler, TerminateHandler>();
+            services.AddTransient<IWorkflowErrorHandler, SuspendHandler>();
 
             services.AddSingleton<IWorkflowController, WorkflowController>();
             services.AddSingleton<IWorkflowHost, WorkflowHost>();

src/WorkflowCore/Services/DefaultProviders/SingleNodeEventHub.cs

@@ -0,0 +1,45 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using WorkflowCore.Interface;
+using WorkflowCore.Models.LifeCycleEvents;
+
+namespace WorkflowCore.Services
+{
+    public class SingleNodeEventHub : ILifeCycleEventHub
+    {
+        private ICollection<Action<LifeCycleEvent>> _subscribers = new HashSet<Action<LifeCycleEvent>>();
+        private readonly ILogger _logger;
+
+        public SingleNodeEventHub(ILoggerFactory loggerFactory)
+        {
+            _logger = loggerFactory.CreateLogger<SingleNodeEventHub>();
+        }
+
+        public Task PublishNotification(LifeCycleEvent evt)
+        {
+            Task.Run(() =>
+            {
+                foreach (var subscriber in _subscribers)
+                {
+                    try
+                    {
+                        subscriber(evt);
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogWarning(default(EventId), ex, $"Error on event subscriber: {ex.Message}");
+                    }
+                }
+            });
+            return Task.CompletedTask;
+        }
+
+        public void Subscribe(Action<LifeCycleEvent> action)
+        {
+            _subscribers.Add(action);
+        }
+    }
+}