Decision Branches (#14)

docs/primitives.md

@@ -76,6 +76,62 @@ Outputs:
   ResponseBody: step.ResponseBody
 ```
 
+## Branching
+
+You can define multiple independent branches within your workflow and select one based on an expression value.
+Hook up your branches via the `SelectNextStep` property, instead of a `NextStepId`.  The expressions will be matched to the step Ids listed in `SelectNextStep`, and the matching next step(s) will be scheduled to execute next.  If more then one step is matched, then the workflow will have multiple parallel paths.
+
+```json
+{
+  "Id": "decide-workflow",
+  "Version": 1,
+  "Steps": [
+    {
+      "Id": "Start",
+      "StepType": "Decide",
+      "SelectNextStep": {
+      	"A": "data.Value1 == 2",
+      	"B": "data.Value1 == 3"
+      }
+    },    
+    {
+      "Id": "A",
+      "StepType": "EmitLog",
+      "Inputs": {
+        "Message": "\"Hi from A!\""
+      }
+    },
+    {
+      "Id": "B",
+      "StepType": "EmitLog",
+      "Inputs": {
+        "Message": "\"Hi from B!\""
+      }
+    }
+  ]
+}
+
+```
+
+```yaml
+Id: decide-workflow
+Version: 1
+Steps:
+- Id: Start
+  StepType: Decide
+  SelectNextStep:
+    A: data.Value1 == 2
+    B: data.Value1 == 3
+- Id: A
+  StepType: EmitLog
+  Inputs:
+    Message: '"Hi from A!"'
+- Id: B
+  StepType: EmitLog
+  Inputs:
+    Message: '"Hi from B!"'
+```
+
 # Error handling
 
 TODO

src/Conductor.Domain/Conductor.Domain.csproj

@@ -11,7 +11,7 @@
     <PackageReference Include="Newtonsoft.Json" Version="12.0.3" />
     <PackageReference Include="SharpYaml" Version="1.6.5" />
     <PackageReference Include="StackExchange.Redis" Version="2.0.601" />
-    <PackageReference Include="WorkflowCore" Version="3.0.2" />
+    <PackageReference Include="WorkflowCore" Version="3.1.1" />
   </ItemGroup>
 
 </Project>

src/Conductor.Domain/Interfaces/IExpressionEvaluator.cs

@@ -1,9 +1,12 @@
+using System.Collections.Generic;
 using WorkflowCore.Interface;
 
 namespace Conductor.Domain.Interfaces
 {
     public interface IExpressionEvaluator
     {
         object EvaluateExpression(string sourceExpr, object pData, IStepExecutionContext pContext);
+        object EvaluateExpression(string sourceExpr, IDictionary<string, object> parameteters);
+        bool EvaluateOutcomeExpression(string sourceExpr, object data, object outcome);
     }
 }

src/Conductor.Domain/Models/Step.cs

@@ -33,6 +33,7 @@ public class Step
 
         public Dictionary<string, string> Outputs { get; set; } = new Dictionary<string, string>();
 
-
+        public Dictionary<string, string> SelectNextStep { get; set; } = new Dictionary<string, string>();
+
     }
 }

src/Conductor.Domain/Services/ExpressionEvaluator.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Linq;
 using System.Text;
 using Conductor.Domain.Interfaces;
 using WorkflowCore.Interface;
@@ -29,5 +30,33 @@ public object EvaluateExpression(string sourceExpr, object pData, IStepExecution
             return resolvedValue;
         }
 
+        public object EvaluateExpression(string sourceExpr, IDictionary<string, object> parameteters)
+        {
+            var exprParams = new Dictionary<string, object>()
+            {                
+                ["environment"] = Environment.GetEnvironmentVariables(),
+                ["readFile"] = new Func<string, byte[]>(File.ReadAllBytes),
+                ["readText"] = new Func<string, Encoding, string>(File.ReadAllText)
+            };
+
+            parameteters.ToList().ForEach(x => exprParams.Add(x.Key, x.Value));
+
+            object resolvedValue = _scriptHost.EvaluateExpression(sourceExpr, exprParams);
+            return resolvedValue;
+        }
+
+        public bool EvaluateOutcomeExpression(string sourceExpr, object data, object outcome)
+        {
+            object resolvedValue = _scriptHost.EvaluateExpression(sourceExpr, new Dictionary<string, object>()
+            {
+                ["data"] = data,
+                ["outcome"] = outcome,
+                ["environment"] = Environment.GetEnvironmentVariables(),
+                ["readFile"] = new Func<string, byte[]>(File.ReadAllBytes),
+                ["readText"] = new Func<string, Encoding, string>(File.ReadAllText)
+            });
+            return Convert.ToBoolean(resolvedValue);
+        }
+
     }
 }

src/Conductor.Domain/Services/WorkflowLoader.cs

@@ -123,8 +123,7 @@ private WorkflowStepCollection ConvertSteps(ICollection<Step> source, Type dataT
                         compensatables.Add(nextStep);
                 }
 
-                if (!string.IsNullOrEmpty(nextStep.NextStepId))
-                    targetStep.Outcomes.Add(new StepOutcome() { ExternalNextStepId = $"{nextStep.NextStepId}" });
+                AttachOutcomes(nextStep, dataType, targetStep);
 
                 result.Add(targetStep);
 
@@ -215,7 +214,22 @@ private void AttachOutputs(Step source, Type dataType, Type stepType, WorkflowSt
                 step.Outputs.Add(new ActionParameter<IStepBody, object>(acn));
             }
         }
-
+
+        private void AttachOutcomes(Step source, Type dataType, WorkflowStep step)
+        {            
+            if (!string.IsNullOrEmpty(source.NextStepId))
+                step.Outcomes.Add(new ValueOutcome() { ExternalNextStepId = $"{source.NextStepId}" });
+
+            foreach (var nextStep in source.SelectNextStep)
+            {
+                Expression<Func<ExpandoObject, object, bool>> sourceExpr = (data, outcome) => _expressionEvaluator.EvaluateOutcomeExpression(nextStep.Value, data, outcome);
+                step.Outcomes.Add(new ExpressionOutcome<ExpandoObject>(sourceExpr)
+                {
+                    ExternalNextStepId = $"{nextStep.Key}"
+                });
+            }
+        }
+
         private Type FindType(string name)
         {
             name = name.Trim();

src/Conductor.Steps/Conductor.Steps.csproj

@@ -6,7 +6,7 @@
 
   <ItemGroup>
     <PackageReference Include="RestSharp" Version="106.6.10" />
-    <PackageReference Include="WorkflowCore" Version="3.0.2" />
+    <PackageReference Include="WorkflowCore" Version="3.1.1" />
   </ItemGroup>
 
   <ItemGroup>

src/Conductor/Conductor.csproj

@@ -5,7 +5,7 @@
     <AspNetCoreHostingModel>InProcess</AspNetCoreHostingModel>
     <DockerDefaultTargetOS>Linux</DockerDefaultTargetOS>
     <UserSecretsId>0b178d89-9937-49c8-b1f1-efb5f96e516d</UserSecretsId>
-    <Version>0.1.0-alpha</Version>
+    <Version>1.0.0</Version>
     <StartupObject>Conductor.Program</StartupObject>
   </PropertyGroup>
 
@@ -27,7 +27,7 @@
     <PackageReference Include="Microsoft.Extensions.Logging" Version="3.1.0" />
     <PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.4.10" />
     <PackageReference Include="Microsoft.VisualStudio.Web.CodeGeneration.Design" Version="2.2.3" />
-    <PackageReference Include="WorkflowCore" Version="3.0.2" />
+    <PackageReference Include="WorkflowCore" Version="3.1.1" />
     <PackageReference Include="WorkflowCore.Providers.Redis" Version="3.0.0" />
   </ItemGroup>
 

tests/Conductor.IntegrationTests/Scenarios/DecisionScenario.cs

@@ -0,0 +1,101 @@
+using System;
+using System.Collections.Generic;
+using System.Dynamic;
+using System.Net;
+using FluentAssertions;
+using System.Threading;
+using System.Threading.Tasks;
+using Conductor.Domain.Models;
+using Conductor.Models;
+using Newtonsoft.Json.Linq;
+using RestSharp;
+using Xunit;
+
+namespace Conductor.IntegrationTests.Scenarios
+{
+    [Collection("Conductor")]
+    public class DecisionScenario : Scenario
+    {
+
+        public DecisionScenario(Setup setup) : base(setup)
+        {
+        }
+
+        [Fact]
+        public async void Scenario()
+        {
+            dynamic add1inputs = new ExpandoObject();
+            add1inputs.Value1 = "data.Value1";
+            add1inputs.Value2 = "data.Value2";
+
+            dynamic add2inputs = new ExpandoObject();
+            add2inputs.Value1 = "data.Value1";
+            add2inputs.Value2 = "data.Value3";
+
+            var definition = new Definition()
+            {
+                Id = Guid.NewGuid().ToString(),
+                Steps = new List<Step>()
+                {
+                    new Step()
+                    {
+                        Id = "Decide",
+                        StepType = "Decide",
+                        SelectNextStep = new Dictionary<string, string>()
+                        {
+                            ["A"] = "data.Flag == 1",
+                            ["B"] = "data.Flag == 0"
+                        }
+                    },
+                    new Step()
+                    {
+                        Id = "A",
+                        StepType = "AddTest",
+                        Inputs = add1inputs,
+                        Outputs = new Dictionary<string, string>()
+                        {
+                            ["Result"] = "step.Result"
+                        }
+                    },
+                    new Step()
+                    {
+                        Id = "B",
+                        StepType = "AddTest",
+                        Inputs = add2inputs,
+                        Outputs = new Dictionary<string, string>()
+                        {
+                            ["Result"] = "step.Result"
+                        }
+                    }
+                }
+            };
+
+            var registerRequest = new RestRequest(@"/definition", Method.POST);
+            registerRequest.AddJsonBody(definition);
+            var registerResponse = _client.Execute(registerRequest);
+            registerResponse.StatusCode.Should().Be(HttpStatusCode.NoContent);
+            Thread.Sleep(1000);
+
+            var startRequest1 = new RestRequest($"/workflow/{definition.Id}", Method.POST);
+            startRequest1.AddJsonBody(new { Value1 = 2, Value2 = 3, Value3 = 4, Flag = 1 });
+            var startResponse1 = _client.Execute<WorkflowInstance>(startRequest1);
+            startResponse1.StatusCode.Should().Be(HttpStatusCode.Created);
+
+            var startRequest2 = new RestRequest($"/workflow/{definition.Id}", Method.POST);
+            startRequest2.AddJsonBody(new { Value1 = 2, Value2 = 3, Value3 = 4, Flag = 0 });
+            var startResponse2 = _client.Execute<WorkflowInstance>(startRequest2);
+            startResponse2.StatusCode.Should().Be(HttpStatusCode.Created);
+
+            var instance1 = await WaitForComplete(startResponse1.Data.WorkflowId);
+            instance1.Status.Should().Be("Complete");
+            var data1 = JObject.FromObject(instance1.Data);
+            data1["Result"].Value<int>().Should().Be(5);
+
+            var instance2 = await WaitForComplete(startResponse2.Data.WorkflowId);
+            instance2.Status.Should().Be("Complete");
+            var data2 = JObject.FromObject(instance2.Data);
+            data2["Result"].Value<int>().Should().Be(6);
+        }
+
+    }
+}

tests/Conductor.IntegrationTests/Setup.cs

@@ -19,6 +19,7 @@ public Setup()
                 .UseCompose()
                 .FromFile(@"docker-compose.yml")
                 .RemoveOrphans()
+                //.ForceBuild()
                 .WaitForHttp("conductor1", @"http://localhost:5101/api/info")
                 .WaitForHttp("conductor2", @"http://localhost:5102/api/info")
                 .Build().Start();