Workflow Core v1.7 (#237)

Author: danielgerlag

ReleaseNotes/1.7.0.md

@@ -0,0 +1,80 @@
+# Workflow Core 1.7.0
+
+* Various performance optimizations, any users of the EntityFramework persistence providers will have to update their persistence libraries to the latest version as well.
+* Added `CancelCondition` to fluent builder API.
+
+    ```
+    .CancelCondition(data => <<expression>>, <<Continue after cancellation>>)
+
+    ```
+
+    This allows you to specify a condition under which any active step can be prematurely cancelled.
+    For example, suppose you create a future scheduled task, but you want to cancel the future execution of this task if some condition becomes true.
+
+
+    ```c#
+    builder
+        .StartWith(context => Console.WriteLine("Hello"))
+        .Schedule(data => TimeSpan.FromSeconds(5)).Do(schedule => schedule
+            .StartWith<DoSomething>()
+            .Then<DoSomethingFurther>()
+        )
+        .CancelCondition(data => !data.SheduledTaskRequired)
+        .Then(context => Console.WriteLine("Doing normal tasks"));
+    ```
+
+    You could also use this implement a parallel flow where once a single path completes, all the other paths are cancelled.
+
+    ```c#
+    .Parallel()
+        .Do(then => then
+            .StartWith<DoSomething>()
+            .WaitFor("Approval", (data, context) => context.Workflow.IdNow)
+        )
+        .Do(then => then
+            .StartWith<DoSomething>()
+            .Delay(data => TimeSpan.FromDays(3))
+            .Then<EscalateIssue>()
+        )
+    .Join()
+        .CancelCondition(data => data.IsApproved, true)
+    .Then<MoveAlong>();
+    ```
+
+* Deprecated `WorkflowCore.LockProviders.RedLock` in favour of `WorkflowCore.Providers.Redis`
+* Create a new `WorkflowCore.Providers.Redis` library that includes providers for distributed locking, queues and event hubs.
+    * Provides Queueing support backed by Redis.
+    * Provides Distributed locking support backed by Redis.
+    * Provides event hub support backed by Redis.
+
+    This makes it possible to have a cluster of nodes processing your workflows.
+
+    ## Installing
+
+    Install the NuGet package "WorkflowCore.Providers.Redis"
+
+    Using Nuget package console
+        ```
+        PM> Install-Package WorkflowCore.Providers.Redis
+        ```
+    Using .NET CLI
+        ```
+        dotnet add package WorkflowCore.Providers.Redis
+        ```
+
+
+    ## Usage
+
+    Use the `IServiceCollection` extension methods when building your service provider
+    * .UseRedisQueues
+    * .UseRedisLocking
+    * .UseRedisEventHub
+
+    ```C#
+    services.AddWorkflow(cfg =>
+    {
+        cfg.UseRedisLocking("localhost:6379");
+        cfg.UseRedisQueues("localhost:6379", "my-app");
+        cfg.UseRedisEventHub("localhost:6379", "my-channel")
+    });
+    ```

WorkflowCore.sln

@@ -94,6 +94,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "ReleaseNotes", "ReleaseNote
 		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
+		ReleaseNotes\1.7.0.md = ReleaseNotes\1.7.0.md
 	EndProjectSection
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Sample14", "src\samples\WorkflowCore.Sample14\WorkflowCore.Sample14.csproj", "{6BC66637-B42A-4334-ADFB-DBEC9F29D293}"
@@ -120,7 +121,9 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Persistence.My
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Tests.MySQL", "test\WorkflowCore.Tests.MySQL\WorkflowCore.Tests.MySQL.csproj", "{DF7F7ECA-1771-40C9-9CD0-AFEFC44E60DE}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.Tests.DynamoDB", "WorkflowCore.Tests.DynamoDB\WorkflowCore.Tests.DynamoDB.csproj", "{0FF3F27E-E909-4ABA-BF82-39D1BA133EA7}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Tests.DynamoDB", "test\WorkflowCore.Tests.DynamoDB\WorkflowCore.Tests.DynamoDB.csproj", "{3ECEC028-7E2C-4983-B928-26C073B51BB7}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.Providers.Redis", "src\providers\WorkflowCore.Providers.Redis\WorkflowCore.Providers.Redis.csproj", "{435C6263-C6F8-4E93-B417-D861E9C22E18}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -304,10 +307,14 @@ Global
 		{DF7F7ECA-1771-40C9-9CD0-AFEFC44E60DE}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{DF7F7ECA-1771-40C9-9CD0-AFEFC44E60DE}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{DF7F7ECA-1771-40C9-9CD0-AFEFC44E60DE}.Release|Any CPU.Build.0 = Release|Any CPU
-		{0FF3F27E-E909-4ABA-BF82-39D1BA133EA7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{0FF3F27E-E909-4ABA-BF82-39D1BA133EA7}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{0FF3F27E-E909-4ABA-BF82-39D1BA133EA7}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{0FF3F27E-E909-4ABA-BF82-39D1BA133EA7}.Release|Any CPU.Build.0 = Release|Any CPU
+		{3ECEC028-7E2C-4983-B928-26C073B51BB7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{3ECEC028-7E2C-4983-B928-26C073B51BB7}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{3ECEC028-7E2C-4983-B928-26C073B51BB7}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{3ECEC028-7E2C-4983-B928-26C073B51BB7}.Release|Any CPU.Build.0 = Release|Any CPU
+		{435C6263-C6F8-4E93-B417-D861E9C22E18}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{435C6263-C6F8-4E93-B417-D861E9C22E18}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{435C6263-C6F8-4E93-B417-D861E9C22E18}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{435C6263-C6F8-4E93-B417-D861E9C22E18}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -360,7 +367,8 @@ Global
 		{5E82A137-0954-46A1-8C46-13C00F0E4842} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2}
 		{453E260D-DBDC-4DDC-BC9C-CA500CED7897} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2}
 		{DF7F7ECA-1771-40C9-9CD0-AFEFC44E60DE} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
-		{0FF3F27E-E909-4ABA-BF82-39D1BA133EA7} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
+		{3ECEC028-7E2C-4983-B928-26C073B51BB7} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
+		{435C6263-C6F8-4E93-B417-D861E9C22E18} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {DC0FA8D3-6449-4FDA-BB46-ECF58FAD23B4}

src/WorkflowCore/Interface/ICancellationProcessor.cs

@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Interface
+{
+    public interface ICancellationProcessor
+    {
+        void ProcessCancellations(WorkflowInstance workflow, WorkflowDefinition workflowDef, WorkflowExecutorResult executionResult);
+    }
+}

src/WorkflowCore/Interface/ILifeCycleEventHub.cs

@@ -10,5 +10,7 @@ public interface ILifeCycleEventHub
     {
         Task PublishNotification(LifeCycleEvent evt);
         void Subscribe(Action<LifeCycleEvent> action);
+        Task Start();
+        Task Stop();
     }
 }

src/WorkflowCore/Interface/IStepBuilder.cs

@@ -213,6 +213,13 @@ public interface IStepBuilder<TData, TStepBody>
         /// <param name="builder"></param>
         /// <returns></returns>
         IStepBuilder<TData, TStepBody> CompensateWithSequence(Action<IWorkflowBuilder<TData>> builder);
+
+        /// <summary>
+        /// Prematurely cancel the execution of this step on a condition
+        /// </summary>
+        /// <param name="cancelCondition"></param>
+        /// <returns></returns>
+        IStepBuilder<TData, TStepBody> CancelCondition(Expression<Func<TData, bool>> cancelCondition, bool proceedAfterCancel = false);
 
     }
 }

src/WorkflowCore/Models/ExecutionPointer.cs

@@ -6,6 +6,8 @@ namespace WorkflowCore.Models
 {
     public class ExecutionPointer
     {
+        private IReadOnlyCollection<string> _scope = new List<string>();
+
         public string Id { get; set; }
 
         public int StepId { get; set; }
@@ -26,8 +28,8 @@ public class ExecutionPointer
 
         public bool EventPublished { get; set; }
 
-        public object EventData { get; set; }                
-        
+        public object EventData { get; set; }
+
         public Dictionary<string, object> ExtensionAttributes { get; set; } = new Dictionary<string, object>();
 
         public string StepName { get; set; }
@@ -43,9 +45,12 @@ public class ExecutionPointer
         public object Outcome { get; set; }
 
         public PointerStatus Status { get; set; } = PointerStatus.Legacy;
-        
-        public Stack<string> Scope { get; set; } = new Stack<string>();
-                        
+
+        public IReadOnlyCollection<string> Scope
+        {
+            get => _scope;
+            set => _scope = new List<string>(value);
+        }
     }
 
     public enum PointerStatus
@@ -57,6 +62,7 @@ public enum PointerStatus
         Sleeping = 4,
         WaitingForEvent = 5,
         Failed = 6,
-        Compensated = 7
+        Compensated = 7,
+        Cancelled = 8
     }
 }

src/WorkflowCore/Models/ExecutionPointerCollection.cs

@@ -0,0 +1,103 @@
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+
+namespace WorkflowCore.Models
+{
+    public class ExecutionPointerCollection : ICollection<ExecutionPointer>
+    {
+        private readonly Dictionary<string, ExecutionPointer> _dictionary = new Dictionary<string, ExecutionPointer>();
+        private readonly Dictionary<string, ICollection<ExecutionPointer>> _scopeMap = new Dictionary<string, ICollection<ExecutionPointer>>();
+
+        public ExecutionPointerCollection()
+        {
+        }
+
+        public ExecutionPointerCollection(int capacity)
+        {
+            _dictionary = new Dictionary<string, ExecutionPointer>(capacity);
+        }
+
+        public ExecutionPointerCollection(ICollection<ExecutionPointer> pointers)
+        {
+            foreach (var ptr in pointers)
+            {
+                Add(ptr);
+            }
+        }
+
+        public IEnumerator<ExecutionPointer> GetEnumerator()
+        {
+            return _dictionary.Values.GetEnumerator();
+        }
+
+        IEnumerator IEnumerable.GetEnumerator()
+        {
+            return GetEnumerator();
+        }
+
+        public ExecutionPointer FindById(string id)
+        {
+            if (!_dictionary.ContainsKey(id))
+                return null;
+
+            return _dictionary[id];
+        }
+
+        public ICollection<ExecutionPointer> FindByScope(string stackFrame)
+        {
+            if (!_scopeMap.ContainsKey(stackFrame))
+                return new List<ExecutionPointer>();
+
+            return _scopeMap[stackFrame];
+        }
+
+        public void Add(ExecutionPointer item)
+        {
+            _dictionary.Add(item.Id, item);
+
+            foreach (var stackFrame in item.Scope)
+            {
+                if (!_scopeMap.ContainsKey(stackFrame))
+                    _scopeMap.Add(stackFrame, new List<ExecutionPointer>());
+                _scopeMap[stackFrame].Add(item);
+            }
+        }
+
+        public void Clear()
+        {
+            _dictionary.Clear();
+            _scopeMap.Clear();
+        }
+
+        public bool Contains(ExecutionPointer item)
+        {
+            return _dictionary.ContainsValue(item);
+        }
+
+        public void CopyTo(ExecutionPointer[] array, int arrayIndex)
+        {
+            _dictionary.Values.CopyTo(array, arrayIndex);
+        }
+
+        public bool Remove(ExecutionPointer item)
+        {
+            foreach (var stackFrame in item.Scope)
+            {
+                _scopeMap[stackFrame].Remove(item);
+            }
+
+            return _dictionary.Remove(item.Id);
+        }
+
+        public ExecutionPointer Find(Predicate<ExecutionPointer> match)
+        {
+            return _dictionary.Values.FirstOrDefault(x => match(x));
+        }
+
+        public int Count => _dictionary.Count;
+        public bool IsReadOnly => false;
+    }
+}

src/WorkflowCore/Models/WorkflowInstance.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Linq;
 
 namespace WorkflowCore.Models
 {
@@ -15,7 +16,7 @@ public class WorkflowInstance
 
         public string Reference { get; set; }
 
-        public List<ExecutionPointer> ExecutionPointers { get; set; } = new List<ExecutionPointer>();
+        public ExecutionPointerCollection ExecutionPointers { get; set; } = new ExecutionPointerCollection();
 
         public long? NextExecution { get; set; }
 
@@ -25,8 +26,14 @@ public class WorkflowInstance
 
         public DateTime CreateTime { get; set; }
 
-        public DateTime? CompleteTime { get; set; }        
+        public DateTime? CompleteTime { get; set; }
 
+        public bool IsBranchComplete(string parentId)
+        {
+            return ExecutionPointers
+                .FindByScope(parentId)
+                .All(x => x.EndTime != null);
+        }
     }
 
     public enum WorkflowStatus 

src/WorkflowCore/Models/WorkflowStep.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Linq.Expressions;
 using System.Reflection;
 using WorkflowCore.Interface;
 
@@ -33,6 +34,10 @@ public abstract class WorkflowStep
 
         public virtual bool RevertChildrenAfterCompensation => false;
 
+        public virtual LambdaExpression CancelCondition { get; set; }
+
+        public bool ProceedOnCancel { get; set; } = false;
+
         public virtual ExecutionPipelineDirective InitForExecution(WorkflowExecutorResult executorResult, WorkflowDefinition defintion, WorkflowInstance workflow, ExecutionPointer executionPointer)
         {
             return ExecutionPipelineDirective.Next;
@@ -61,6 +66,7 @@ public virtual void PrimeForRetry(ExecutionPointer pointer)
         /// <param name="executionPointer"></param>
         public virtual void AfterWorkflowIteration(WorkflowExecutorResult executorResult, WorkflowDefinition defintion, WorkflowInstance workflow, ExecutionPointer executionPointer)
         {
+            
         }
 
         public virtual IStepBody ConstructBody(IServiceProvider serviceProvider)