3.控制结构

控制结构

分支

可以在工作流中定义多个分支，更根据表达式的值选择执行哪个分支。

使用CreateBranch()方法定义分支

//创建两个分支
var branch1 = builder.CreateBranch()
    .StartWith<PrintMessage>()
        .Input(step => step.Message, data => "hi from 1")
    .Then<PrintMessage>()
        .Input(step => step.Message, data => "bye from 1");

var branch2 = builder.CreateBranch()
    .StartWith<PrintMessage>()
        .Input(step => step.Message, data => "hi from 2")
    .Then<PrintMessage>()
        .Input(step => step.Message, data => "bye from 2");

//如果data.value1的值为one，则执行branch1
//如果data.value1的值为two，则执行branch2
builder
    .StartWith<HelloWorld>()
    .Decide(data => data.Value1)
        .Branch((data, outcome) => data.Value1 == "one", branch1)
        .Branch((data, outcome) => data.Value1 == "two", branch2);

foreach

public class ForEachWorkflow : IWorkflow
{
    public string Id => "Foreach";
    public int Version => 1;

    public void Build(IWorkflowBuilder<object> builder)
    {
        builder
            .StartWith<SayHello>()
            .ForEach(data => new List<int>() { 1, 2, 3, 4 })
                .Do(x => x
                    .StartWith<DisplayContext>()
                        .Input(step => step.Message, (data, context) => context.Item)
                    .Then<DoSomething>())
            .Then<SayGoodbye>();
    }        
}

while

public class WhileWorkflow : IWorkflow<MyData>
{
    public string Id => "While";
    public int Version => 1;

    public void Build(IWorkflowBuilder<MyData> builder)
    {
        builder
            .StartWith<SayHello>()
            .While(data => data.Counter < 3)
                .Do(x => x
                    .StartWith<DoSomething>()
                    .Then<IncrementStep>()
                        .Input(step => step.Value1, data => data.Counter)
                        .Output(data => data.Counter, step => step.Value2))
            .Then<SayGoodbye>();
    }        
}

if

public class IfWorkflow : IWorkflow<MyData>
{ 
    public void Build(IWorkflowBuilder<MyData> builder)
    {
        builder
            .StartWith<SayHello>()
            .If(data => data.Counter < 3).Do(then => then
                .StartWith<PrintMessage>()
                    .Input(step => step.Message, data => "Value is less than 3")
            )
            .If(data => data.Counter < 5).Do(then => then
                .StartWith<PrintMessage>()
                    .Input(step => step.Message, data => "Value is less than 5")
            )
            .Then<SayGoodbye>();
    }        
}

并行

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<Task1dot1>()
                        .Then<Task1dot2>()
                .Do(then =>
                    then.StartWith<Task2dot1>()
                        .Then<Task2dot2>()
                .Do(then =>
                    then.StartWith<Task3dot1>()
                        .Then<Task3dot2>()
            .Join()
            .Then<SayGoodbye>();
    }        
}

按计划执行

在未来按照计划执行，使得工作流在后台异步执行

builder
    .StartWith(context => Console.WriteLine("Hello"))
    .Schedule(data => TimeSpan.FromSeconds(5)).Do(schedule => schedule
        .StartWith(context => Console.WriteLine("Doing scheduled tasks"))
    )
    .Then(context => Console.WriteLine("Doing normal tasks"));

延迟执行

延迟工作流的当前分支一段时间

builder
    .StartWith(context => Console.WriteLine("Hello"))
    .Delay(p => TimeSpan.FromSeconds(5) );

重复

设置一组重复的后台步骤，知道满足特定条件

builder
    .StartWith(context => Console.WriteLine("Hello"))
    .Recur(data => TimeSpan.FromSeconds(5), data => data.Counter > 5).Do(recur => recur
        .StartWith(context => Console.WriteLine("Doing recurring task"))
    )
    .Then(context => Console.WriteLine("Carry on"));

案例 Parallel ForEach

• steps

public class SayHello : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("Hello");
        return ExecutionResult.Next();
    }
}
public class DisplayContext : StepBody
{        

    public object Item { get; set; }

    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine($"Working on item {Item}");
        return ExecutionResult.Next();
    }
}
public class DoSomething : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("Doing something...");
        return ExecutionResult.Next();
    }
}
public class SayGoodbye : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("Goodbye");
        return ExecutionResult.Next();
    }
}

• workflow

public class ForEachWorkflow : IWorkflow
{
    public string Id => "Foreach";
    public int Version => 1;

    public void Build(IWorkflowBuilder<object> builder)
    {
        builder
            .StartWith<SayHello>()
            .ForEach(data => new List<int> { 1, 2, 3, 4 })
                .Do(x => x
                    .StartWith<DisplayContext>()
                        .Input(step => step.Item, (data, context) => context.Item)
                    .Then<DoSomething>())
            .Then<SayGoodbye>();
    }        
}    

• program

IServiceCollection services = new ServiceCollection();
services.AddLogging();
services.AddWorkflow();
var serviceProvider = services.BuildServiceProvider();
var host = serviceProvider.GetService<IWorkflowHost>();
host.RegisterWorkflow<ForEachWorkflow>();
host.Start();
Console.WriteLine("Starting workflow...");
string workflowId = host.StartWorkflow("Foreach").Result;
Console.ReadLine();
host.Stop();

案例If

• steps

public class PrintMessage : StepBody
{
    public string Message { get; set; }

    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine(Message);
        return ExecutionResult.Next();
    }
}
public class SayGoodbye : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("Goodbye");
        return ExecutionResult.Next();
    }
}
public class SayHello : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("Hello");
        return ExecutionResult.Next();
    }
}

• workflow

public class MyData
{
    public int Counter { get; set; }
}
public class IfWorkflow : IWorkflow<MyData>
{
    public string Id => "if-sample";
    public int Version => 1;

    public void Build(IWorkflowBuilder<MyData> builder)
    {
        builder
            .StartWith<SayHello>()
            .If(data => data.Counter < 3).Do(then => then
                .StartWith<PrintMessage>()
                    .Input(step => step.Message, data => "Value is less than 3")
            )
            .If(data => data.Counter < 5).Do(then => then
                .StartWith<PrintMessage>()
                    .Input(step => step.Message, data => "Value is less than 5")
            )
            .Then<SayGoodbye>();
    }        
}

• program

IServiceCollection services = new ServiceCollection();
services.AddLogging();
services.AddWorkflow();
var serviceProvider = services.BuildServiceProvider();
var host = serviceProvider.GetService<IWorkflowHost>();
host.RegisterWorkflow<IfWorkflow, MyData>();
host.Start();

Console.WriteLine("Starting workflow...");
string workflowId = host.StartWorkflow("if-sample", new MyData { Counter = 4 }).Result;

Console.ReadLine();
host.Stop();

案例 Parallel Tasks

• steps

public class PrintMessage : StepBody
{
    public string Message { get; set; }

    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("当前线程："+ Thread.GetCurrentProcessorId().ToString()+ "message:"+Message);
        return ExecutionResult.Next();
    }
}
public class SayHello : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("Hello");
        return ExecutionResult.Next();
    }
}
public class SayGoodbye : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("Goodbye");
        return ExecutionResult.Next();
    }
}

• workflow

public class MyData
{
    public int Counter { get; set; }
}
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")
                        .Then<PrintMessage>()
                            .Input(step => step.Message, data => "Item 2.3"))
                .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>();
    }        
}

• program

IServiceCollection services = new ServiceCollection();
services.AddLogging();
services.AddWorkflow();
var serviceProvider = services.BuildServiceProvider();
var host = serviceProvider.GetService<IWorkflowHost>();
var controller = serviceProvider.GetService<IWorkflowController>();
controller.RegisterWorkflow<ParallelWorkflow, MyData>();//使用容器进行注册

Console.WriteLine("Starting workflow...");
controller.StartWorkflow<MyData>("parallel-sample");//使用容器开始

Console.ReadLine();
host.Stop();

案例Schedule

• workflow

class ScheduleWorkflow : IWorkflow
{
    public string Id => "schedule-sample";
    public int Version => 1;

    public void Build(IWorkflowBuilder<object> builder)
    {
        builder
            .StartWith(context => Console.WriteLine("Hello"))
            .Schedule(data => TimeSpan.FromSeconds(5)).Do(schedule => schedule
                .StartWith(context => Console.WriteLine("Doing scheduled tasks"))
            )
            .Then(context => { Console.WriteLine("Doing normal tasks"); Console.WriteLine("等待5s"); });
    }
}

案例Recur

public class MyData
{
    public int Counter { get; set; }
}
class RecurSampleWorkflow : IWorkflow<MyData>
{
    public string Id => "recur-sample";
    public int Version => 1;

    public void Build(IWorkflowBuilder<MyData> builder)
    {
        builder
            .StartWith(context => Console.WriteLine("Hello"))
            //5s一次，直到data.Counter > 5
            .Recur(data => TimeSpan.FromSeconds(5), data => data.Counter > 5).Do(recur => recur
                .StartWith(context => Console.WriteLine("Doing recurring task"))
            )
            .Then(context => Console.WriteLine("Carry on"));
    }
}

案例Branch

• steps

public class PrintMessage : StepBody
{
    public string Message { get; set; }

    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine(Message);
        return ExecutionResult.Next();
    }
}
public class SayHello : StepBody
{
    public override ExecutionResult Run(IStepExecutionContext context)
    {
        Console.WriteLine("Hello");
        return ExecutionResult.Next();
    }
}

• workflow

public class MyData
{
    public int Value { get; set; }
}

public class OutcomeWorkflow : IWorkflow<MyData>
{
    public string Id => "outcome-sample";
    public int Version => 1;

    public void Build(IWorkflowBuilder<MyData> builder)
    {
        var branch1 = builder.CreateBranch()
            .StartWith<PrintMessage>()
                .Input(step => step.Message, data => "hi from 1")
            .Then<PrintMessage>()
                .Input(step => step.Message, data => "bye from 1");

        var branch2 = builder.CreateBranch()
            .StartWith<PrintMessage>()
                .Input(step => step.Message, data => "hi from 2")
            .Then<PrintMessage>()
                .Input(step => step.Message, data => "bye from 2");
//如果data.value1的值为one，则执行branch1
//如果data.value1的值为two，则执行branch2
        builder
            .StartWith<SayHello>()
            .Decide(data => data.Value)
                .Branch(1, branch1)
                .Branch(2, branch2);
    }        
}

• program

IServiceCollection services = new ServiceCollection();
services.AddLogging();
services.AddWorkflow();
var serviceProvider = services.BuildServiceProvider();
var host = serviceProvider.GetService<IWorkflowHost>();
host.RegisterWorkflow<OutcomeWorkflow, MyData>();
host.Start();

Console.WriteLine("Starting workflow...");
host.StartWorkflow("outcome-sample", new MyData { Value = 2 });


Console.ReadLine();
host.Stop();