Added support for executing a sub-workflow

src/WorkflowCore.Testing/WorkflowCore.Testing.csproj

@@ -12,6 +12,7 @@
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.0" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="3.1.0" />
     <PackageReference Include="Microsoft.Extensions.Logging" Version="3.1.0" />
+    <PackageReference Include="xunit.abstractions" Version="2.0.3" />
   </ItemGroup>
 
   <ItemGroup>

src/WorkflowCore/Interface/IWorkflowModifier.cs

@@ -183,5 +183,15 @@ IStepBuilder<TData, Activity> Activity(string activityName, Expression<Func<TDat
     /// <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);
+
+    /// <summary>
+    /// Execute a sub-workflow
+    /// </summary>
+    /// <param name="subWorkflowId">Id of the sub-workflow to start</param>
+    /// <param name="parameters">The data to pass to the sub-workflow</param>
+    /// <param name="cancelCondition">A condition that when true will cancel this sub-workflow</param>
+    /// <returns></returns>
+    IStepBuilder<TData, SubWorkflowStepBody> SubWorkflow(string subWorkflowId, Expression<Func<TData, object>> parameters = null,
+        Expression<Func<TData, bool>> cancelCondition = null);
     }
 }

src/WorkflowCore/Models/ExecutionPointer.cs

@@ -51,6 +51,10 @@ public IReadOnlyCollection<string> Scope
             get => _scope;
             set => _scope = new List<string>(value);
         }
+
+        public bool IsComplete => Status == PointerStatus.Complete;
+
+        public bool HasChildren => Children?.Count > 0;
     }
 
     public enum PointerStatus

src/WorkflowCore/Models/LifeCycleEvents/SubWorkflowLifeCycleEvent.cs

@@ -0,0 +1,7 @@
+namespace WorkflowCore.Models.LifeCycleEvents
+{
+    public class SubWorkflowLifeCycleEvent : LifeCycleEvent
+    {
+
+    }
+}

src/WorkflowCore/Primitives/SubWorkflowStepBody.cs

@@ -1,15 +1,60 @@
 using System;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using WorkflowCore.Interface;
 using WorkflowCore.Models;
+using WorkflowCore.Models.LifeCycleEvents;
 
 namespace WorkflowCore.Primitives
 {
     public class SubWorkflowStepBody : StepBody
     {
+        private readonly IScopeProvider _scopeProvider;
+
+        public SubWorkflowStepBody(IScopeProvider scopeProvider)
+        {
+            _scopeProvider = scopeProvider;
+        }
+
         public override ExecutionResult Run(IStepExecutionContext context)
         {
-            // TODO: What is this supposed to do?
-            throw new NotImplementedException();
+            var scope = _scopeProvider.CreateScope(context);
+            var workflowController = scope.ServiceProvider.GetRequiredService<IWorkflowController>();
+            var logger = scope.ServiceProvider.GetRequiredService<ILoggerFactory>().CreateLogger(
+                typeof(SubWorkflowStepBody).Namespace + "." + nameof(SubWorkflowStepBody));
+
+            if (!context.ExecutionPointer.EventPublished)
+            {
+                var result = workflowController.StartWorkflow(SubWorkflowId, context.Workflow.Data, context.Workflow.Id).Result;
+
+                logger.LogDebug("Started sub workflow {Name} with id='{SubId}' from workflow {WorkflowDefinitionId} ({Id})", 
+                    SubWorkflowId, result, context.Workflow.WorkflowDefinitionId, context.Workflow.Id);
+
+                logger.LogDebug("Workflow {Name} ({SubId}) is waiting for event SubWorkflowLifeCycleEvent with key='{EventKey}'", 
+                    SubWorkflowId, result, result);
+
+                var effectiveDate = DateTime.MinValue;
+                return ExecutionResult.WaitForEvent(nameof(SubWorkflowLifeCycleEvent), result, effectiveDate);
+            }
+
+            logger.LogDebug("Sub workflow {Name} ({SubId}) completed", SubWorkflowId, 
+                context.ExecutionPointer.EventKey);
+
+            var persistenceProvider = scope.ServiceProvider.GetRequiredService<IPersistenceProvider>();
+            var workflowInstance = persistenceProvider.GetWorkflowInstance(context.ExecutionPointer.EventKey).Result;
+            if (workflowInstance.Status == WorkflowStatus.Terminated)
+            {
+                throw new NotImplementedException(workflowInstance.Status.ToString());    
+            }
+
+            Result = workflowInstance.Data;
+            return ExecutionResult.Next();
         }
+
+        public string SubWorkflowId { get; set; }
+
+        public object Parameters { get; set; }
+
+        public object Result { get; set; }
     }
 }

src/WorkflowCore/Primitives/WaitFor.cs

@@ -25,7 +25,8 @@ public override ExecutionResult Run(IStepExecutionContext context)
                     effectiveDate = EffectiveDate;
                 }
 
-                return ExecutionResult.WaitForEvent(EventName, EventKey, effectiveDate);
+                var eventKey = context.Workflow.Reference ?? EventKey;
+                return ExecutionResult.WaitForEvent(EventName, eventKey, effectiveDate);
             }
 
             EventData = context.ExecutionPointer.EventData;

src/WorkflowCore/ServiceCollectionExtensions.cs

@@ -81,6 +81,7 @@ public static IServiceCollection AddWorkflow(this IServiceCollection services, A
             services.AddTransient<ISyncWorkflowRunner, SyncWorkflowRunner>();
 
             services.AddTransient<Foreach>();
+            services.AddTransient<SubWorkflowStepBody>();
 
             return services;
         }

src/WorkflowCore/Services/ExecutionResultProcessor.cs

@@ -29,12 +29,16 @@ public ExecutionResultProcessor(IExecutionPointerFactory pointerFactory, IDateTi
 
         public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition def, ExecutionPointer pointer, WorkflowStep step, ExecutionResult result, WorkflowExecutorResult workflowResult)
         {
+            var stepInfo = $"{step.Name ?? step.BodyType.Name} ({step.Id})";
+
             pointer.PersistenceData = result.PersistenceData;
             pointer.Outcome = result.OutcomeValue;
             if (result.SleepFor.HasValue)
             {
                 pointer.SleepUntil = _datetimeProvider.UtcNow.Add(result.SleepFor.Value);
                 pointer.Status = PointerStatus.Sleeping;
+                _logger.LogDebug("Step {StepName} on workflow {WorkflowDefinitionId} ({WorkflowId}) will sleep for {SleepUntil}",
+                    stepInfo, workflow.WorkflowDefinitionId, workflow.Id, result.SleepFor.Value);
             }
 
             if (!string.IsNullOrEmpty(result.EventName))
@@ -54,6 +58,9 @@ public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition
                     SubscribeAsOf = result.EventAsOf,
                     SubscriptionData = result.SubscriptionData
                 });
+
+                _logger.LogDebug("Step {StepName} on workflow {WorkflowDefinitionId} ({WorkflowId}) waiting for event {EventName}",
+                    stepInfo, workflow.WorkflowDefinitionId, workflow.Id, pointer.EventName);
             }
 
             if (result.Proceed)
@@ -87,6 +94,9 @@ public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition
                     WorkflowDefinitionId = workflow.WorkflowDefinitionId,
                     Version = workflow.Version
                 });
+
+                _logger.LogDebug("Step {StepName} on workflow {WorkflowDefinitionId} ({WorkflowId}) completed",
+                    stepInfo, workflow.WorkflowDefinitionId, workflow.Id);
             }
             else
             {

src/WorkflowCore/Services/FluentBuilders/StepBuilder.cs

@@ -588,5 +588,34 @@ public IStepBuilder<TData, Activity> Activity(Expression<Func<TData, IStepExecut
             Step.Outcomes.Add(new ValueOutcome { NextStep = newStep.Id });
             return stepBuilder;
         }
+
+        public IStepBuilder<TData, SubWorkflowStepBody> SubWorkflow(
+            string subWorkflowId, 
+            Expression<Func<TData, object>> parameters = null,
+            Expression<Func<TData, bool>> cancelCondition = null)
+        {
+            var newStep = new WorkflowStep<SubWorkflowStepBody>();
+            newStep.CancelCondition = cancelCondition;
+
+            WorkflowBuilder.AddStep(newStep);
+            var stepBuilder = new StepBuilder<TData, SubWorkflowStepBody>(WorkflowBuilder, newStep);
+            stepBuilder.Input((step) => step.SubWorkflowId, (data) => subWorkflowId);
+
+            if (parameters != null)
+                stepBuilder.Input((step) => step.Parameters, parameters);
+
+            // use the result of the sub workflow as an output
+            // merge it with parent workflow data
+            stepBuilder.Output((body, data) =>
+            {
+                foreach (var prop in typeof(TData).GetProperties())
+                {
+                    prop.SetValue(data, prop.GetValue(body.Result));
+                }
+            });
+
+            Step.Outcomes.Add(new ValueOutcome { NextStep = newStep.Id });
+            return stepBuilder;
+        }
     }
 }

src/WorkflowCore/Services/FluentBuilders/WorkflowBuilder.cs

@@ -296,10 +296,16 @@ 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);
         }
+
+        public IStepBuilder<TData, SubWorkflowStepBody> SubWorkflow(string subWorkflowId, Expression<Func<TData, object>> parameters = null, Expression<Func<TData, bool>> cancelCondition = null)
+        {
+            return Start().SubWorkflow(subWorkflowId, parameters, cancelCondition);
+        }
 
         private IStepBuilder<TData, InlineStepBody> Start()
         {

src/WorkflowCore/Services/WorkflowExecutor.cs

@@ -89,7 +89,7 @@ public async Task<WorkflowExecutorResult> Execute(WorkflowInstance workflow, Can
                         WorkflowId = workflow.Id,
                         ExecutionPointerId = pointer.Id,
                         ErrorTime = _datetimeProvider.UtcNow,
-                        Message = ex.Message
+                        Message = ex.ToString()
                     });
 
                     _executionResultProcessor.HandleStepException(workflow, def, pointer, step, ex);
@@ -156,9 +156,11 @@ private async Task ExecuteStep(WorkflowInstance workflow, WorkflowStep step, Exe
                 CancellationToken = cancellationToken
             };
 
+            var stepInfo = $"{step.Name ?? step.BodyType.Name} ({step.Id})";
+
             using (var scope = _scopeProvider.CreateScope(context))
             {
-                _logger.LogDebug("Starting step {StepName} on workflow {WorkflowId}", step.Name, workflow.Id);
+                _logger.LogDebug("Starting step {StepName} on workflow {WorkflowDefinitionId} ({WorkflowId})", stepInfo, workflow.WorkflowDefinitionId, workflow.Id);
 
                 IStepBody body = step.ConstructBody(scope.ServiceProvider);
                 var stepExecutor = scope.ServiceProvider.GetRequiredService<IStepExecutor>();
@@ -221,6 +223,13 @@ private async Task DetermineNextExecutionTime(WorkflowInstance workflow, Workflo
 
             if (workflow.Status == WorkflowStatus.Complete)
             {
+                await OnComplete(workflow, def);
+                return;
+            }
+
+            if (workflow.Status == WorkflowStatus.Terminated)
+            {
+                await OnTerminated(workflow, def);
                 return;
             }
 
@@ -236,7 +245,7 @@ private async Task DetermineNextExecutionTime(WorkflowInstance workflow, Workflo
                 workflow.NextExecution = Math.Min(pointerSleep, workflow.NextExecution ?? pointerSleep);
             }
 
-            foreach (var pointer in workflow.ExecutionPointers.Where(x => x.Active && (x.Children ?? new List<string>()).Count > 0))
+            foreach (var pointer in workflow.ExecutionPointers.Where(x => x.Active && x.HasChildren))
             {
                 if (!workflow.ExecutionPointers.FindByScope(pointer.Id).All(x => x.EndTime.HasValue))
                     continue;
@@ -256,6 +265,11 @@ private async Task DetermineNextExecutionTime(WorkflowInstance workflow, Workflo
                 return;
             }
 
+            await OnComplete(workflow, def);
+        }
+
+        private async Task OnComplete(WorkflowInstance workflow, WorkflowDefinition def)
+        {
             workflow.Status = WorkflowStatus.Complete;
             workflow.CompleteTime = _datetimeProvider.UtcNow;
 
@@ -264,6 +278,8 @@ private async Task DetermineNextExecutionTime(WorkflowInstance workflow, Workflo
                 var middlewareRunner = scope.ServiceProvider.GetRequiredService<IWorkflowMiddlewareRunner>();
                 await middlewareRunner.RunPostMiddleware(workflow, def);
             }
+
+            _logger.LogDebug("Workflow {WorkflowDefinitionId} ({Id}) completed", workflow.WorkflowDefinitionId, workflow.Id);
 
             _publisher.PublishNotification(new WorkflowCompleted
             {
@@ -274,5 +290,28 @@ private async Task DetermineNextExecutionTime(WorkflowInstance workflow, Workflo
                 Version = workflow.Version
             });
         }
+
+        private async Task OnTerminated(WorkflowInstance workflow, WorkflowDefinition def)
+        {
+            workflow.Status = WorkflowStatus.Terminated;
+            workflow.CompleteTime = _datetimeProvider.UtcNow;
+
+            using (var scope = _serviceProvider.CreateScope())
+            {
+                var middlewareRunner = scope.ServiceProvider.GetRequiredService<IWorkflowMiddlewareRunner>();
+                await middlewareRunner.RunPostMiddleware(workflow, def);
+            }
+
+            _logger.LogDebug("Workflow {WorkflowDefinitionId} ({Id}) terminated", workflow.WorkflowDefinitionId, workflow.Id);
+
+            _publisher.PublishNotification(new WorkflowTerminated
+            {
+                EventTimeUtc = _datetimeProvider.UtcNow,
+                Reference = workflow.Reference,
+                WorkflowInstanceId = workflow.Id,
+                WorkflowDefinitionId = workflow.WorkflowDefinitionId,
+                Version = workflow.Version
+            });
+        }
     }
 }

src/WorkflowCore/Services/WorkflowHost.cs

@@ -165,6 +165,15 @@ public Task<bool> TerminateWorkflow(string workflowId)
 
         public void HandleLifeCycleEvent(LifeCycleEvent evt)
         {
+            switch (evt)
+            {
+                // publish the event as sub workflow lifecycle event
+                case WorkflowCompleted _:
+                case WorkflowTerminated _:
+                    _workflowController.PublishEvent(nameof(SubWorkflowLifeCycleEvent), evt.WorkflowInstanceId, evt.Reference);
+                    break;
+            }
+
             OnLifeCycleEvent?.Invoke(evt);
         }
 

src/providers/WorkflowCore.Persistence.EntityFramework/Services/EntityFrameworkPersistenceProvider.cs

@@ -100,6 +100,10 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id, CancellationT
         {
             using (var db = ConstructDbContext())
             {
+                if (!Guid.TryParse(Id, out _))
+                {
+
+                }
                 var uid = new Guid(Id);
                 var raw = await db.Set<PersistedWorkflow>()
                     .Include(wf => wf.ExecutionPointers)