Merge branch 'master' into issue-1072

# Conflicts: # src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs
danielgerlag · Nov 11, 2022 · 0ef8be8 · 0ef8be8
2 parents 60cfe3d + 97974b2
commit 0ef8be8
Show file tree

Hide file tree

Showing 32 changed files with 606 additions and 123 deletions.
diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml
@@ -0,0 +1,3 @@
+# These are supported funding model platforms
+
+github: [danielgerlag]
diff --git a/src/Directory.Build.props b/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>
diff --git a/src/WorkflowCore.DSL/Models/v1/StepSourceV1.cs b/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;
     }
 }
diff --git a/src/WorkflowCore.DSL/Services/DefinitionLoader.cs b/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);

diff --git a/src/WorkflowCore/Interface/IWorkflowModifier.cs b/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);
     }
 }
diff --git a/src/WorkflowCore/Interface/Persistence/IWorkflowRepository.cs b/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]

diff --git a/src/WorkflowCore/Services/BackgroundTasks/QueueConsumer.cs b/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()

diff --git a/src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs b/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);

diff --git a/src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs b/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)

diff --git a/src/WorkflowCore/Services/DefaultProviders/TransientMemoryPersistenceProvider.cs b/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);
 

diff --git a/src/WorkflowCore/Services/FluentBuilders/StepBuilder.cs b/src/WorkflowCore/Services/FluentBuilders/StepBuilder.cs
@@ -522,5 +522,24 @@ public IStepBuilder<TData, Activity> Activity(string activityName, Expression<Fu
             Step.Outcomes.Add(new ValueOutcome { NextStep = newStep.Id });
             return stepBuilder;
         }
+
+        public 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)
+        {
+            var newStep = new WorkflowStep<Activity>();
+            newStep.CancelCondition = cancelCondition;
+
+            WorkflowBuilder.AddStep(newStep);
+            var stepBuilder = new StepBuilder<TData, Activity>(WorkflowBuilder, newStep);
+            stepBuilder.Input((step) => step.ActivityName, activityName);
+
+            if (parameters != null)
+                stepBuilder.Input((step) => step.Parameters, parameters);
+
+            if (effectiveDate != null)
+                stepBuilder.Input((step) => step.EffectiveDate, effectiveDate);
+
+            Step.Outcomes.Add(new ValueOutcome { NextStep = newStep.Id });
+            return stepBuilder;
+        }
     }
 }
diff --git a/src/WorkflowCore/Services/FluentBuilders/WorkflowBuilder.cs b/src/WorkflowCore/Services/FluentBuilders/WorkflowBuilder.cs
@@ -84,6 +84,11 @@ public void AttachBranch(IWorkflowBuilder branch)
                         if (step2.Children[i] == oldId)
                             step2.Children[i] = step.Id;
                     }
+
+                    if (step2.CompensationStepId == oldId)
+                    {
+                        step2.CompensationStepId = step.Id;
+                    }
                 }
             }
 
@@ -104,6 +109,11 @@ public void AttachBranch(IWorkflowBuilder branch)
                         if (step2.Children[i] == oldId)
                             step2.Children[i] = step.Id;
                     }
+
+                    if (step2.CompensationStepId == oldId)
+                    {
+                        step2.CompensationStepId = step.Id;
+                    }
                 }
             }
 
@@ -271,6 +281,10 @@ public IStepBuilder<TData, Activity> Activity(string activityName, Expression<Fu
         {
             return Start().Activity(activityName, parameters, effectiveDate, cancelCondition);
         }
+        public 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)
+        {
+            return Start().Activity(activityName, parameters, effectiveDate, cancelCondition);
+        }
 
         private IStepBuilder<TData, InlineStepBody> Start()
         {

diff --git a/...s/WorkflowCore.Persistence.EntityFramework/Services/EntityFrameworkPersistenceProvider.cs b/...s/WorkflowCore.Persistence.EntityFramework/Services/EntityFrameworkPersistenceProvider.cs
@@ -151,6 +151,32 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
                 await db.SaveChangesAsync(cancellationToken);
             }
         }
+
+		public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            using (var db = ConstructDbContext())
+            {
+                var uid = new Guid(workflow.Id);
+                var existingEntity = await db.Set<PersistedWorkflow>()
+                    .Where(x => x.InstanceId == uid)
+                    .Include(wf => wf.ExecutionPointers)
+                    .ThenInclude(ep => ep.ExtensionAttributes)
+                    .Include(wf => wf.ExecutionPointers)
+                    .AsTracking()
+                    .FirstAsync(cancellationToken);
+
+                var workflowPersistable = workflow.ToPersistable(existingEntity);
+
+                foreach (var subscription in subscriptions)
+                {
+                    subscription.Id = Guid.NewGuid().ToString();
+                    var subscriptionPersistable = subscription.ToPersistable();
+                    db.Set<PersistedSubscription>().Add(subscriptionPersistable);
+                }
+
+                await db.SaveChangesAsync(cancellationToken);
+            }
+        }
 
         public async Task TerminateSubscription(string eventSubscriptionId, CancellationToken cancellationToken = default)
         {

diff --git a/src/providers/WorkflowCore.Persistence.MongoDB/Services/MongoPersistenceProvider.cs b/src/providers/WorkflowCore.Persistence.MongoDB/Services/MongoPersistenceProvider.cs
@@ -149,6 +149,17 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
             await WorkflowInstances.ReplaceOneAsync(x => x.Id == workflow.Id, workflow, cancellationToken: cancellationToken);
         }
 
+        public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+        {
+            using (var session = await _database.Client.StartSessionAsync())
+            {
+                session.StartTransaction();
+                await PersistWorkflow(workflow, cancellationToken);
+                await EventSubscriptions.InsertManyAsync(subscriptions, cancellationToken: cancellationToken);
+                await session.CommitTransactionAsync();
+            }
+        }
+
         public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken cancellationToken = default)
         {
             var now = asAt.ToUniversalTime().Ticks;

diff --git a/src/providers/WorkflowCore.Persistence.MySQL/MysqlContext.cs b/src/providers/WorkflowCore.Persistence.MySQL/MysqlContext.cs
@@ -23,7 +23,7 @@ protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
             base.OnConfiguring(optionsBuilder);
 #if NETSTANDARD2_0
             optionsBuilder.UseMySql(_connectionString, _mysqlOptionsAction);
-#elif NETSTANDARD2_1_OR_GREATER
+#elif NETSTANDARD2_1_OR_GREATER || NET6_0_OR_GREATER
             optionsBuilder.UseMySql(_connectionString, ServerVersion.AutoDetect(_connectionString), _mysqlOptionsAction);
 #endif
         }

diff --git a/src/providers/WorkflowCore.Persistence.PostgreSQL/WorkflowCore.Persistence.PostgreSQL.csproj b/src/providers/WorkflowCore.Persistence.PostgreSQL/WorkflowCore.Persistence.PostgreSQL.csproj
@@ -23,7 +23,8 @@
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net6.0'">
-    <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="6.0.3" />
+	<PackageReference Include="Npgsql" Version="6.0.6" />
+    <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="6.0.6" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="6.0.3">
       <PrivateAssets>All</PrivateAssets>
     </PackageReference>
@@ -34,7 +35,7 @@
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.1'">
-    <PackageReference Include="Npgsql" Version="5.0.1.1" />
+    <PackageReference Include="Npgsql" Version="5.0.14" />
     <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="5.0.1" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="5.0.1">
       <PrivateAssets>All</PrivateAssets>

diff --git a/src/providers/WorkflowCore.Persistence.RavenDB/Services/RavendbPersistenceProvider.cs b/src/providers/WorkflowCore.Persistence.RavenDB/Services/RavendbPersistenceProvider.cs
@@ -1,6 +1,7 @@
 using Raven.Client.Documents;
 using Raven.Client.Documents.Linq;
 using Raven.Client.Documents.Operations;
+using Raven.Client.Documents.Session;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -56,21 +57,41 @@ public async Task PersistWorkflow(WorkflowInstance workflow, CancellationToken c
 		{
 			using (var session = _database.OpenAsyncSession())
 			{
-				session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.WorkflowDefinitionId, workflow.WorkflowDefinitionId);
-				session.Advanced.Patch<WorkflowInstance, int>(workflow.Id, x => x.Version, workflow.Version);
-				session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.Description, workflow.Description);
-				session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.Reference, workflow.Reference);
-				session.Advanced.Patch<WorkflowInstance, ExecutionPointerCollection>(workflow.Id, x => x.ExecutionPointers, workflow.ExecutionPointers);
-				session.Advanced.Patch<WorkflowInstance, long?>(workflow.Id, x => x.NextExecution, workflow.NextExecution);
-				session.Advanced.Patch<WorkflowInstance, WorkflowStatus>(workflow.Id, x => x.Status, workflow.Status);
-				session.Advanced.Patch<WorkflowInstance, object>(workflow.Id, x => x.Data, workflow.Data);
-				session.Advanced.Patch<WorkflowInstance, DateTime>(workflow.Id, x => x.CreateTime, workflow.CreateTime);
-				session.Advanced.Patch<WorkflowInstance, DateTime?>(workflow.Id, x => x.CompleteTime, workflow.CompleteTime);
+				PatchSession(session, workflow);
+				await session.SaveChangesAsync(cancellationToken);
+			}
+		}
+
+		public async Task PersistWorkflow(WorkflowInstance workflow, List<EventSubscription> subscriptions, CancellationToken cancellationToken = default)
+		{
+			using (var session = _database.OpenAsyncSession())
+			{
+				PatchSession(session, workflow);
+
+                foreach (var subscription in subscriptions)
+				{
+					await session.StoreAsync(subscription, cancellationToken);
+				}
 
 				await session.SaveChangesAsync(cancellationToken);
 			}
 		}
 
+		private void PatchSession(IAsyncDocumentSession session, WorkflowInstance workflow)
+        {
+			session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.WorkflowDefinitionId, workflow.WorkflowDefinitionId);
+			session.Advanced.Patch<WorkflowInstance, int>(workflow.Id, x => x.Version, workflow.Version);
+			session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.Description, workflow.Description);
+			session.Advanced.Patch<WorkflowInstance, string>(workflow.Id, x => x.Reference, workflow.Reference);
+			session.Advanced.Patch<WorkflowInstance, ExecutionPointerCollection>(workflow.Id, x => x.ExecutionPointers, workflow.ExecutionPointers);
+			session.Advanced.Patch<WorkflowInstance, long?>(workflow.Id, x => x.NextExecution, workflow.NextExecution);
+			session.Advanced.Patch<WorkflowInstance, WorkflowStatus>(workflow.Id, x => x.Status, workflow.Status);
+			session.Advanced.Patch<WorkflowInstance, object>(workflow.Id, x => x.Data, workflow.Data);
+			session.Advanced.Patch<WorkflowInstance, DateTime>(workflow.Id, x => x.CreateTime, workflow.CreateTime);
+			session.Advanced.Patch<WorkflowInstance, DateTime?>(workflow.Id, x => x.CompleteTime, workflow.CompleteTime);
+
+		}
+
 		public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt, CancellationToken cancellationToken = default)
 		{
 			var now = asAt.ToUniversalTime().Ticks;
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,3 @@
		# These are supported funding model platforms

		github: [danielgerlag]