Parallel steps

WorkflowCore.sln

@@ -86,6 +86,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.Tests.Sqlite",
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.LockProviders.SqlServer", "src\providers\WorkflowCore.LockProviders.SqlServer\WorkflowCore.LockProviders.SqlServer.csproj", "{AAE2E9F9-37EF-4AE1-A200-D37417C9040C}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.Sample13", "src\samples\WorkflowCore.Sample13\WorkflowCore.Sample13.csproj", "{77C49ACA-203E-428C-A4DB-114DFE454988}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -228,6 +230,10 @@ Global
 		{AAE2E9F9-37EF-4AE1-A200-D37417C9040C}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{AAE2E9F9-37EF-4AE1-A200-D37417C9040C}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{AAE2E9F9-37EF-4AE1-A200-D37417C9040C}.Release|Any CPU.Build.0 = Release|Any CPU
+		{77C49ACA-203E-428C-A4DB-114DFE454988}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{77C49ACA-203E-428C-A4DB-114DFE454988}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{77C49ACA-203E-428C-A4DB-114DFE454988}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{77C49ACA-203E-428C-A4DB-114DFE454988}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -270,5 +276,6 @@ Global
 		{BB776411-D279-419F-8697-5C6F52BCD5CD} = {5080DB09-CBE8-4C45-9957-C3BB7651755E}
 		{F9F8F9CD-01D9-468B-856D-6A87F0762A01} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
 		{AAE2E9F9-37EF-4AE1-A200-D37417C9040C} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2}
+		{77C49ACA-203E-428C-A4DB-114DFE454988} = {5080DB09-CBE8-4C45-9957-C3BB7651755E}
 	EndGlobalSection
 EndGlobal

src/WorkflowCore/Interface/IParallelStepBuilder.cs

@@ -0,0 +1,14 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Primitives;
+
+namespace WorkflowCore.Interface
+{
+    public interface IParallelStepBuilder<TData, TStepBody>
+        where TStepBody : IStepBody
+    {
+        IParallelStepBuilder<TData, TStepBody> Do(Action<IWorkflowBuilder<TData>> builder);
+        IStepBuilder<TData, Sequence> Join();
+    }
+}

src/WorkflowCore/Interface/IStepBuilder.cs

@@ -101,5 +101,7 @@ public interface IStepBuilder<TData, TStepBody>
         /// <param name="outcomeValue"></param>
         /// <returns></returns>
         IContainerStepBuilder<TData, When, OutcomeSwitch> When(Expression<Func<TData, object>> outcomeValue, string label = null);
+
+        IParallelStepBuilder<TData, Sequence> Parallel();
     }
 }

src/WorkflowCore/Interface/IWorkflowBuilder.cs

@@ -8,13 +8,14 @@ namespace WorkflowCore.Interface
 {
     public interface IWorkflowBuilder
     {
-        int InitialStep { get; set; }                
+        int LastStep { get; }                
 
         IWorkflowBuilder<T> UseData<T>();
 
         WorkflowDefinition Build(string id, int version);
 
-        void AddStep(WorkflowStep step);                
+        void AddStep(WorkflowStep step);
+
     }
 
     public interface IWorkflowBuilder<TData> : IWorkflowBuilder

src/WorkflowCore/Services/ParallelStepBuilder.cs

@@ -0,0 +1,45 @@
+using System;
+using System.Collections;
+using System.Collections.Generic;
+using System.Linq;
+using System.Linq.Expressions;
+using System.Threading.Tasks;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+using WorkflowCore.Primitives;
+
+namespace WorkflowCore.Services
+{
+    public class ParallelStepBuilder<TData, TStepBody> : IParallelStepBuilder<TData, TStepBody>
+        where TStepBody : IStepBody
+    {
+        private readonly IStepBuilder<TData, Sequence> _referenceBuilder;
+        private readonly IStepBuilder<TData, TStepBody> _stepBuilder;
+
+        public IWorkflowBuilder<TData> WorkflowBuilder { get; private set; }
+
+        public WorkflowStep<TStepBody> Step { get; set; }
+
+        public ParallelStepBuilder(IWorkflowBuilder<TData> workflowBuilder, IStepBuilder<TData, TStepBody> stepBuilder, IStepBuilder<TData, Sequence> referenceBuilder)
+        {
+            WorkflowBuilder = workflowBuilder;
+            Step = stepBuilder.Step;
+            _stepBuilder = stepBuilder;
+            _referenceBuilder = referenceBuilder;
+        }
+
+        public IParallelStepBuilder<TData, TStepBody> Do(Action<IWorkflowBuilder<TData>> builder)
+        {
+            int lastStep = WorkflowBuilder.LastStep;
+            builder.Invoke(WorkflowBuilder);            
+            Step.Children.Add(lastStep + 1); //TODO: make more elegant
+
+            return this;
+        }
+
+        public IStepBuilder<TData, Sequence> Join()
+        {
+            return _referenceBuilder;
+        }
+    }
+}

src/WorkflowCore/Services/StepBuilder.cs

@@ -250,6 +250,18 @@ public IContainerStepBuilder<TData, When, OutcomeSwitch> When(Expression<Func<TD
             return stepBuilder;
         }
 
+        public IParallelStepBuilder<TData, Sequence> Parallel()
+        {
+            var newStep = new WorkflowStep<Sequence>();
+            var newBuilder = new StepBuilder<TData, Sequence>(WorkflowBuilder, newStep);
+            WorkflowBuilder.AddStep(newStep);
+            var stepBuilder = new ParallelStepBuilder<TData, Sequence>(WorkflowBuilder, newBuilder, newBuilder);
+
+            Step.Outcomes.Add(new StepOutcome() { NextStep = newStep.Id });
+
+            return stepBuilder;
+        }
+
         public IStepBuilder<TData, TStepBody> Do(Action<IWorkflowBuilder<TData>> builder)
         {
             builder.Invoke(WorkflowBuilder);

src/WorkflowCore/Services/WorkflowBuilder.cs

@@ -10,14 +10,14 @@ namespace WorkflowCore.Services
 {
     public class WorkflowBuilder : IWorkflowBuilder
     {
-        public int InitialStep { get; set; }
-
         protected List<WorkflowStep> Steps { get; set; } = new List<WorkflowStep>();
 
         protected WorkflowErrorHandling DefaultErrorBehavior = WorkflowErrorHandling.Retry;
 
         protected TimeSpan? DefaultErrorRetryInterval;
 
+        public int LastStep => Steps.Max(x => x.Id);
+
         public IWorkflowBuilder<T> UseData<T>()
         {
             IWorkflowBuilder<T> result = new WorkflowBuilder<T>(Steps);

src/WorkflowCore/WorkflowCore.csproj

@@ -18,9 +18,10 @@
     <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
     <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
     <Description>Workflow Core is a light weight workflow engine targeting .NET Standard.</Description>
-    <Version>1.2.5</Version>
-    <AssemblyVersion>1.2.5.0</AssemblyVersion>
-    <FileVersion>1.2.5.0</FileVersion>
+    <Version>1.2.6</Version>
+    <AssemblyVersion>1.2.6.0</AssemblyVersion>
+    <FileVersion>1.2.6.0</FileVersion>
+    <PackageReleaseNotes>Added Parallel steps</PackageReleaseNotes>
   </PropertyGroup>
 
   <ItemGroup>

src/samples/WorkflowCore.Sample13/ParallelWorkflow.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Sample13
+{    
+    public class ParallelWorkflow : IWorkflow<MyData>
+    {
+        public string Id => "parallel-sample";
+        public int Version => 1;
+
+        public void Build(IWorkflowBuilder<MyData> builder)
+        {
+            builder
+                .StartWith<SayHello>()
+                .Parallel()
+                    .Do(then => 
+                        then.StartWith<PrintMessage>()
+                                .Input(step => step.Message, data => "Item 1.1")
+                            .Then<PrintMessage>()
+                                .Input(step => step.Message, data => "Item 1.2"))
+                    .Do(then =>
+                        then.StartWith<PrintMessage>()
+                                .Input(step => step.Message, data => "Item 2.1")
+                            .Then<PrintMessage>()
+                                .Input(step => step.Message, data => "Item 2.2"))
+                    .Do(then =>
+                        then.StartWith<PrintMessage>()
+                                .Input(step => step.Message, data => "Item 3.1")
+                            .Then<PrintMessage>()
+                                .Input(step => step.Message, data => "Item 3.2"))
+                .Join()
+                .Then<SayGoodbye>();
+        }        
+    }
+
+    public class MyData
+    {
+        public int Counter { get; set; }
+    }
+}

src/samples/WorkflowCore.Sample13/Program.cs

@@ -0,0 +1,45 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using WorkflowCore.Interface;
+
+namespace WorkflowCore.Sample13
+{
+    class Program
+    {
+        public static void Main(string[] args)
+        {
+            IServiceProvider serviceProvider = ConfigureServices();
+
+            //start the workflow host
+            var host = serviceProvider.GetService<IWorkflowHost>();
+            host.RegisterWorkflow<ParallelWorkflow, MyData>();
+            host.Start();
+
+            Console.WriteLine("Starting workflow...");
+            string workflowId = host.StartWorkflow("parallel-sample").Result;
+
+            Console.ReadLine();
+            host.Stop();
+        }
+
+        private static IServiceProvider ConfigureServices()
+        {
+            //setup dependency injection
+            IServiceCollection services = new ServiceCollection();
+            services.AddLogging();
+            services.AddWorkflow();
+            //services.AddWorkflow(x => x.UseMongoDB(@"mongodb://localhost:27017", "workflow-test002"));
+            //services.AddWorkflow(x => x.UseSqlServer(@"Server=.\SQLEXPRESS;Database=WorkflowCoreTest001;Trusted_Connection=True;", true, true));
+            //services.AddWorkflow(x => x.UseSqlite(@"Data Source=wfc001.db;", true));            
+
+
+            var serviceProvider = services.BuildServiceProvider();
+
+            //config logging
+            var loggerFactory = serviceProvider.GetService<ILoggerFactory>();
+            //loggerFactory.AddDebug(LogLevel.Debug);
+            return serviceProvider;
+        }
+    }
+}

src/samples/WorkflowCore.Sample13/README.md

@@ -0,0 +1,27 @@
+# Parallel sample
+
+Illustrates how to run several branches of steps in parallel, and then join once all are complete.
+
+
+```c#
+builder
+    .StartWith<SayHello>()
+    .Parallel()
+        .Do(then => 
+            then.StartWith<PrintMessage>()
+                    .Input(step => step.Message, data => "Item 1.1")
+                .Then<PrintMessage>()
+                    .Input(step => step.Message, data => "Item 1.2"))
+        .Do(then =>
+            then.StartWith<PrintMessage>()
+                    .Input(step => step.Message, data => "Item 2.1")
+                .Then<PrintMessage>()
+                    .Input(step => step.Message, data => "Item 2.2"))
+        .Do(then =>
+            then.StartWith<PrintMessage>()
+                    .Input(step => step.Message, data => "Item 3.1")
+                .Then<PrintMessage>()
+                    .Input(step => step.Message, data => "Item 3.2"))
+    .Join()
+    .Then<SayGoodbye>();
+```

src/samples/WorkflowCore.Sample13/Steps/PrintMessage.cs

@@ -0,0 +1,19 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Sample13
+{    
+    public class PrintMessage : StepBody
+    {
+        public string Message { get; set; }
+
+        public override ExecutionResult Run(IStepExecutionContext context)
+        {
+            Console.WriteLine(Message);
+            return ExecutionResult.Next();
+        }
+    }
+}

src/samples/WorkflowCore.Sample13/Steps/SayGoodbye.cs

@@ -0,0 +1,17 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Sample13
+{    
+    public class SayGoodbye : StepBody
+    {
+        public override ExecutionResult Run(IStepExecutionContext context)
+        {
+            Console.WriteLine("Goodbye");
+            return ExecutionResult.Next();
+        }
+    }
+}

src/samples/WorkflowCore.Sample13/Steps/SayHello.cs

@@ -0,0 +1,17 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Sample13
+{    
+    public class SayHello : StepBody
+    {
+        public override ExecutionResult Run(IStepExecutionContext context)
+        {
+            Console.WriteLine("Hello");
+            return ExecutionResult.Next();
+        }
+    }
+}

src/samples/WorkflowCore.Sample13/WorkflowCore.Sample13.csproj

@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>netcoreapp1.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="1.1.1" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="1.1.1" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Debug" Version="1.1.2" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\WorkflowCore\WorkflowCore.csproj" />
+  </ItemGroup>
+
+</Project>