Merge branch 'master' into issue-1072

# Conflicts:
#	src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs

Author: bobo10001

.github/FUNDING.yml

@@ -0,0 +1,3 @@
+# These are supported funding model platforms
+
+github: [danielgerlag]

src/Directory.Build.props

@@ -4,10 +4,10 @@
     <PackageLicenseUrl>https://github.com/danielgerlag/workflow-core/blob/master/LICENSE.md</PackageLicenseUrl>
     <RepositoryType>git</RepositoryType>
     <RepositoryUrl>https://github.com/danielgerlag/workflow-core.git</RepositoryUrl>
-    <Version>3.6.4</Version>
-    <AssemblyVersion>3.6.4.0</AssemblyVersion>
-    <FileVersion>3.6.4.0</FileVersion>
+    <Version>3.7.0</Version>
+    <AssemblyVersion>3.7.0.0</AssemblyVersion>
+    <FileVersion>3.7.0.0</FileVersion>
     <PackageIconUrl>https://github.com/danielgerlag/workflow-core/raw/master/src/logo.png</PackageIconUrl>
-    <PackageVersion>3.6.4</PackageVersion>
+    <PackageVersion>3.7.0</PackageVersion>
   </PropertyGroup>
 </Project>

src/WorkflowCore.DSL/Models/v1/StepSourceV1.cs

@@ -7,7 +7,7 @@ namespace WorkflowCore.Models.DefinitionStorage.v1
     public class StepSourceV1
     {
         public string StepType { get; set; }
-        
+
         public string Id { get; set; }
 
         public string Name { get; set; }
@@ -29,8 +29,9 @@ public class StepSourceV1
         public ExpandoObject Inputs { get; set; } = new ExpandoObject();
 
         public Dictionary<string, string> Outputs { get; set; } = new Dictionary<string, string>();
-        
+
         public Dictionary<string, string> SelectNextStep { get; set; } = new Dictionary<string, string>();
 
+        public bool ProceedOnCancel { get; set; } = false;
     }
 }

src/WorkflowCore.DSL/Services/DefinitionLoader.cs

@@ -101,6 +101,7 @@ private WorkflowStepCollection ConvertSteps(ICollection<StepSourceV1> source, Ty
                 targetStep.ErrorBehavior = nextStep.ErrorBehavior;
                 targetStep.RetryInterval = nextStep.RetryInterval;
                 targetStep.ExternalId = $"{nextStep.Id}";
+                targetStep.ProceedOnCancel = nextStep.ProceedOnCancel;
 
                 AttachInputs(nextStep, dataType, stepType, targetStep);
                 AttachOutputs(nextStep, dataType, stepType, targetStep);

src/WorkflowCore/Interface/IWorkflowModifier.cs

@@ -152,5 +152,15 @@ IContainerStepBuilder<TData, Recur, TStepBody> Recur(Expression<Func<TData, Time
     IStepBuilder<TData, Activity> Activity(string activityName, Expression<Func<TData, object>> parameters = null,
         Expression<Func<TData, DateTime>> effectiveDate = null, Expression<Func<TData, bool>> cancelCondition = null);
 
+    /// <summary>
+    /// Wait here until an external activity is complete
+    /// </summary>
+    /// <param name="activityName">The name used to identify the activity to wait for</param>
+    /// <param name="parameters">The data to pass the external activity worker</param>
+    /// <param name="effectiveDate">Listen for events as of this effective date</param>
+    /// <param name="cancelCondition">A conditon that when true will cancel this WaitFor</param>
+    /// <returns></returns>
+    IStepBuilder<TData, Activity> Activity(Expression<Func<TData, IStepExecutionContext, string>> activityName, Expression<Func<TData, object>> parameters = null,
+        Expression<Func<TData, DateTime>> effectiveDate = null, Expression<Func<TData, bool>> cancelCondition = null);
     }
 }

src/WorkflowCore/Interface/Persistence/IWorkflowRepository.cs

@@ -12,6 +12,8 @@ public interface IWorkflowRepository
 
         Task PersistWorkflow(WorkflowInstance workflow, CancellationToken cancellationToken = default);
 
+        Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default);
+
         Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken cancellationToken = default);
 
         [Obsolete]

src/WorkflowCore/Services/BackgroundTasks/QueueConsumer.cs

@@ -53,8 +53,11 @@ public virtual void Start()
         public virtual void Stop()
         {
             _cancellationTokenSource.Cancel();
-            DispatchTask.Wait();
-            DispatchTask = null;
+            if (DispatchTask != null)
+            {
+                DispatchTask.Wait();
+                DispatchTask = null;
+            }
         }
 
         private async Task Execute()

src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs

@@ -55,7 +55,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
                     finally
                     {
                         WorkflowActivity.Enrich(result);
-                        await _persistenceStore.PersistWorkflow(workflow, cancellationToken);
+                        await _persistenceStore.PersistWorkflow(workflow, result.Subscriptions, cancellationToken);
                         await QueueProvider.QueueWork(itemId, QueueType.Index);
                         _greylist.Remove($"wf:{itemId}");
                     }
@@ -68,7 +68,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
                 {
                     foreach (var sub in result.Subscriptions)
                     {
-                        await SubscribeEvent(sub, _persistenceStore, cancellationToken);
+                        await TryProcessSubscription(sub, _persistenceStore, cancellationToken);
                     }
 
                     await _persistenceStore.PersistErrors(result.Errors, cancellationToken);
@@ -98,12 +98,8 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()
 
         }
 
-        private async Task SubscribeEvent(EventSubscription subscription, IPersistenceProvider persistenceStore, CancellationToken cancellationToken)
+        private async Task TryProcessSubscription(EventSubscription subscription, IPersistenceProvider persistenceStore, CancellationToken cancellationToken)
         {
-            //TODO: move to own class
-            Logger.LogDebug("Subscribing to event {EventName} {EventKey} for workflow {WorkflowId} step {StepId}", subscription.EventName, subscription.EventKey, subscription.WorkflowId, subscription.StepId);
-
-            await persistenceStore.CreateEventSubscription(subscription, cancellationToken);
             if (subscription.EventName != Event.EventTypeActivity)
             {
                 var events = await persistenceStore.GetEvents(subscription.EventName, subscription.EventKey, subscription.SubscribeAsOf, cancellationToken);

src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs

@@ -46,6 +46,25 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken _
             }
         }
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            lock (_instances)
+            {
+                var existing = _instances.First(x => x.Id == workflow.Id);
+                _instances.Remove(existing);
+                _instances.Add(workflow);
+
+                lock (_subscriptions)
+                {
+                    foreach (var subscription in subscriptions)
+                    {
+                        subscription.Id = Guid.NewGuid().ToString();
+                        _subscriptions.Add(subscription);
+                    }
+                }
+            }
+        }
+
         public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken _ = default)
         {
             lock (_instances)

src/WorkflowCore/Services/DefaultProviders/TransientMemoryPersistenceProvider.cs

@@ -50,6 +50,16 @@ public TransientMemoryPersistenceProvider(ISingletonMemoryProvider innerService)
 
         public Task PersistWorkflow(WorkflowInstance workflow, CancellationToken _ = default) => _innerService.PersistWorkflow(workflow);
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            await PersistWorkflow(workflow, cancellationToken);
+
+            foreach(var subscription in subscriptions)
+            {
+                await CreateEventSubscription(subscription, cancellationToken);
+            }
+        }
+
         public Task TerminateSubscription(string eventSubscriptionId, CancellationToken _ = default) => _innerService.TerminateSubscription(eventSubscriptionId);
         public Task<EventSubscription> GetSubscription(string eventSubscriptionId, CancellationToken _ = default) => _innerService.GetSubscription(eventSubscriptionId);