add foreach.RunParallel option to allow synchronous execution of child branches

Author: dviry

README.md

@@ -160,6 +160,8 @@ These are also available as separate Nuget packages.
 
 * [Parallel ForEach](src/samples/WorkflowCore.Sample09)
 
+* [Sync ForEach](src/samples/WorkflowCore.Sample09s)
+
 * [While Loop](src/samples/WorkflowCore.Sample10)
 
 * [If Statement](src/samples/WorkflowCore.Sample11)

WorkflowCore.sln

@@ -141,6 +141,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.DSL", "src\Wor
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Sample18", "src\samples\WorkflowCore.Sample18\WorkflowCore.Sample18.csproj", "{5BE6D628-B9DB-4C76-AAEB-8F3800509A84}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.Sample09s", "src\samples\WorkflowCore.Sample09s\WorkflowCore.Sample09s.csproj", "{E32CF21A-29CC-46D1-8044-FCC327F2B281}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -343,6 +345,10 @@ Global
 		{5BE6D628-B9DB-4C76-AAEB-8F3800509A84}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{5BE6D628-B9DB-4C76-AAEB-8F3800509A84}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{5BE6D628-B9DB-4C76-AAEB-8F3800509A84}.Release|Any CPU.Build.0 = Release|Any CPU
+		{E32CF21A-29CC-46D1-8044-FCC327F2B281}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{E32CF21A-29CC-46D1-8044-FCC327F2B281}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{E32CF21A-29CC-46D1-8044-FCC327F2B281}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{E32CF21A-29CC-46D1-8044-FCC327F2B281}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -400,6 +406,7 @@ Global
 		{78217204-B873-40B9-8875-E3925B2FBCEC} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
 		{20B98905-08CB-4854-8E2C-A31A078383E9} = {EF47161E-E399-451C-BDE8-E92AAD3BD761}
 		{5BE6D628-B9DB-4C76-AAEB-8F3800509A84} = {5080DB09-CBE8-4C45-9957-C3BB7651755E}
+		{E32CF21A-29CC-46D1-8044-FCC327F2B281} = {5080DB09-CBE8-4C45-9957-C3BB7651755E}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {DC0FA8D3-6449-4FDA-BB46-ECF58FAD23B4}

src/WorkflowCore/Interface/IStepBuilder.cs

@@ -186,6 +186,13 @@ public interface IStepBuilder<TData, TStepBody>
         /// <returns></returns>
         IContainerStepBuilder<TData, Foreach, Foreach> ForEach(Expression<Func<TData, IEnumerable>> collection);
 
+        /// <summary>
+        /// Execute a block of steps, once for each item in a collection in a RunParallel foreach
+        /// </summary>
+        /// <param name="collection">Resolves a collection for iterate over</param>
+        /// <returns></returns>
+        IContainerStepBuilder<TData, Foreach, Foreach> ForEach(Expression<Func<TData, IEnumerable>> collection, Expression<Func<TData, bool>> runParallel);
+
         /// <summary>
         /// Repeat a block of steps until a condition becomes true
         /// </summary>

src/WorkflowCore/Models/IteratorPersistenceData.cs

@@ -0,0 +1,7 @@
+namespace WorkflowCore.Models
+{
+    public class IteratorPersistenceData : ControlPersistenceData
+    {
+        public int Index { get; set; } = 0;
+    }
+}

src/WorkflowCore/Primitives/Foreach.cs

@@ -8,28 +8,45 @@ namespace WorkflowCore.Primitives
 {
     public class Foreach : ContainerStepBody
     {
-        public IEnumerable Collection { get; set; }                
+        public IEnumerable Collection { get; set; }
+        public bool RunParallel { get; set; } = true;
 
         public override ExecutionResult Run(IStepExecutionContext context)
         {
             if (context.PersistenceData == null)
             {
                 var values = Collection.Cast<object>();
-                return ExecutionResult.Branch(new List<object>(values), new ControlPersistenceData() { ChildrenActive = true });
+                if (RunParallel)
+                {
+                    return ExecutionResult.Branch(new List<object>(values), new IteratorPersistenceData() { ChildrenActive = true });
+                }
+                else
+                {
+                    return ExecutionResult.Branch(new List<object>(new object[] { values.ElementAt(0) }), new IteratorPersistenceData() { ChildrenActive = true });
+                }
             }
 
-            if (context.PersistenceData is ControlPersistenceData)
+            if (context.PersistenceData is IteratorPersistenceData persistanceData && persistanceData?.ChildrenActive == true)
             {
-                if ((context.PersistenceData as ControlPersistenceData).ChildrenActive)
+                if (context.Workflow.IsBranchComplete(context.ExecutionPointer.Id))
                 {
-                    if (context.Workflow.IsBranchComplete(context.ExecutionPointer.Id))
+                    if (!RunParallel)
                     {
-                        return ExecutionResult.Next();
+                        var values = Collection.Cast<object>();
+                        persistanceData.Index++;
+                        if (persistanceData.Index < values.Count())
+                        {
+                            return ExecutionResult.Branch(new List<object>(new object[] { values.ElementAt(persistanceData.Index) }), persistanceData);
+                        }
                     }
+
+                    return ExecutionResult.Next();
                 }
+
+                return ExecutionResult.Persist(persistanceData);
             }
 
             return ExecutionResult.Persist(context.PersistenceData);
-        }        
+        }
     }
 }

src/WorkflowCore/Services/FluentBuilders/StepBuilder.cs

@@ -306,6 +306,25 @@ public IContainerStepBuilder<TData, Foreach, Foreach> ForEach(Expression<Func<TD
             return stepBuilder;
         }
 
+        public IContainerStepBuilder<TData, Foreach, Foreach> ForEach(Expression<Func<TData, IEnumerable>> collection, Expression<Func<TData, bool>> runParallel)
+        {
+            var newStep = new WorkflowStep<Foreach>();
+
+            Expression<Func<Foreach, IEnumerable>> inputExpr = (x => x.Collection);
+            newStep.Inputs.Add(new MemberMapParameter(collection, inputExpr));
+
+            Expression<Func<Foreach, bool>> pExpr = (x => x.RunParallel);
+            newStep.Inputs.Add(new MemberMapParameter(runParallel, pExpr));
+
+            WorkflowBuilder.AddStep(newStep);
+            var stepBuilder = new StepBuilder<TData, Foreach>(WorkflowBuilder, newStep);
+
+            Step.Outcomes.Add(new ValueOutcome() { NextStep = newStep.Id });
+
+            return stepBuilder;
+        }
+
+
         public IContainerStepBuilder<TData, While, While> While(Expression<Func<TData, bool>> condition)
         {
             var newStep = new WorkflowStep<While>();

src/providers/WorkflowCore.Persistence.MongoDB/Services/MongoPersistenceProvider.cs

@@ -79,6 +79,7 @@ static MongoPersistenceProvider()
 
             BsonClassMap.RegisterClassMap<ControlPersistenceData>(x => x.AutoMap());
             BsonClassMap.RegisterClassMap<SchedulePersistenceData>(x => x.AutoMap());
+            BsonClassMap.RegisterClassMap<IteratorPersistenceData>(x => x.AutoMap());
         }
 
         static bool indexesCreated = false;

src/samples/WorkflowCore.Sample09/README.md

@@ -1,4 +1,4 @@
-# Foreach sample
+# Foreach Parallel sample
 
 Illustrates how to implement a parallel foreach within your workflow.
 
@@ -14,7 +14,7 @@ builder
 	.Then<SayGoodbye>();
 ```
 
-or get the collectioin from workflow data.
+or get the collection from workflow data.
 
 ```c#
 builder

src/samples/WorkflowCore.Sample09s/ForEachSyncWorkflow.cs

@@ -0,0 +1,26 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Sample09s
+{    
+    public class ForEachSyncWorkflow : IWorkflow
+    {
+        public string Id => "ForeachSync";
+        public int Version => 1;
+
+        public void Build(IWorkflowBuilder<object> builder)
+        {
+            builder
+                .StartWith<SayHello>()
+                .ForEach(data => new List<int>() { 1, 2, 3, 4 }, data => false)
+                    .Do(x => x
+                        .StartWith<DisplayContext>()
+                            .Input(step => step.Item, (data, context) => context.Item)
+                        .Then<DoSomething>())
+                .Then<SayGoodbye>();
+        }        
+    }    
+}

src/samples/WorkflowCore.Sample09s/Program.cs

@@ -0,0 +1,45 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using System;
+using WorkflowCore.Interface;
+
+namespace WorkflowCore.Sample09s
+{
+    class Program
+    {
+        public static void Main(string[] args)
+        {
+            IServiceProvider serviceProvider = ConfigureServices();
+
+            //start the workflow host
+            var host = serviceProvider.GetService<IWorkflowHost>();
+            host.RegisterWorkflow<ForEachSyncWorkflow>();
+            host.Start();
+            
+            Console.WriteLine("Starting workflow...");
+            string workflowId = host.StartWorkflow("Foreach").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=.;Database=WorkflowCore3;Trusted_Connection=True;", true, true));            
+            //services.AddWorkflow(x => x.UseSqlite(@"Data Source=database2.db;", true));            
+            //services.AddWorkflow(x => x.UsePostgreSQL(@"Server=127.0.0.1;Port=32768;Database=workflow_test002;User Id=postgres;", true, true));
+
+
+            var serviceProvider = services.BuildServiceProvider();
+
+            return serviceProvider;
+        }
+
+    }
+}

src/samples/WorkflowCore.Sample09s/README.md

@@ -0,0 +1,29 @@
+# Foreach Sync sample
+
+Illustrates how to implement a synchronous foreach within your workflow.
+
+
+```c#
+builder
+	.StartWith<SayHello>()
+	.ForEach(data => new List<int>() { 1, 2, 3, 4 }, data => false)
+		.Do(x => x
+			.StartWith<DisplayContext>()
+				.Input(step => step.Item, (data, context) => context.Item)
+			.Then<DoSomething>())
+	.Then<SayGoodbye>();
+```
+
+or get the collection from workflow data.
+
+```c#
+builder
+	.StartWith<SayHello>()
+	.ForEach(data => data.MyCollection, data => false)
+		.Do(x => x
+			.StartWith<DisplayContext>()
+				.Input(step => step.Item, (data, context) => context.Item)
+			.Then<DoSomething>())
+	.Then<SayGoodbye>();
+
+```

src/samples/WorkflowCore.Sample09s/Steps/DisplayContext.cs

@@ -0,0 +1,20 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Sample09s
+{
+    public class DisplayContext : StepBody
+    {        
+
+        public object Item { get; set; }
+
+        public override ExecutionResult Run(IStepExecutionContext context)
+        {
+            Console.WriteLine($"Working on item {Item}");
+            return ExecutionResult.Next();
+        }
+    }
+}

src/samples/WorkflowCore.Sample09s/Steps/DoSomething.cs

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

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

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

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

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

src/samples/WorkflowCore.Sample09s/WorkflowCore.Sample09s.csproj

@@ -0,0 +1,19 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>netcoreapp3.0</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Logging.Debug" Version="3.1.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\providers\WorkflowCore.Persistence.MongoDB\WorkflowCore.Persistence.MongoDB.csproj" />
+    <ProjectReference Include="..\..\providers\WorkflowCore.Persistence.PostgreSQL\WorkflowCore.Persistence.PostgreSQL.csproj" />
+    <ProjectReference Include="..\..\providers\WorkflowCore.Persistence.SqlServer\WorkflowCore.Persistence.SqlServer.csproj" />
+    <ProjectReference Include="..\..\WorkflowCore\WorkflowCore.csproj" />
+  </ItemGroup>
+
+</Project>