5.Workflow Core之中间件

工作流可以在工作流开始或者完成之前或者之后使用中间件进行扩展。

步骤中间件

步骤中间件可以为给定的步骤执行其他代码。

使用

创建实现了IWorkflowStepMiddleware的中间件类。

//该中间件将工作流ID和步骤ID添加log上下文中
public class LogCorrelationStepMiddleware : IWorkflowStepMiddleware
{
    private readonly ILogger<LogCorrelationStepMiddleware> _log;

    public LogCorrelationStepMiddleware(
        ILogger<LogCorrelationStepMiddleware> log)
    {
        _log = log;
    }

    public async Task<ExecutionResult> HandleAsync(
        IStepExecutionContext context,
        IStepBody body,
        WorkflowStepDelegate next)
    {
        var workflowId = context.Workflow.Id;
        var stepId = context.Step.Id;

        // Uses log scope to add a few attributes to the scope
        using (_log.BeginScope("{@WorkflowId}", workflowId))
        using (_log.BeginScope("{@StepId}", stepId))
        {
            // Calling next ensures step gets executed
            //必须确保使用next()作为中间件的一部分，如果不使用则步骤不会运行
            return await next();
        }
    }
}

前流程中间件

在工作流启动之前运行的中间件，可以更改工作流上的属性。

// 使用中间件在工作流实例上设置“说明”属性
public class AddDescriptionWorkflowMiddleware : IWorkflowMiddleware
{
    //前中间件声明
    public WorkflowMiddlewarePhase Phase =>
        WorkflowMiddlewarePhase.PreWorkflow;

    public Task HandleAsync(
        WorkflowInstance workflow,
        WorkflowDelegate next
    )
    {
        if (workflow.Data is IDescriptiveWorkflowParams descriptiveParams)
        {
            workflow.Description = descriptiveParams.Description;
        }
        return next();
    }
}

// 解释参数接口
public interface IDescriptiveWorkflowParams
{
    string Description { get; }
}

//解释参数
public MyWorkflowParams : IDescriptiveWorkflowParams
{
    public string Description => $"Run task '{TaskName}'";

    public string TaskName { get; set; }
}

前中间件的异常处理

前中间件和后中间件的异常处理方式不同，前中间件在工作流之前运行，所以前工作流中间件引发的异常会冒泡到StartWorkflow方法，并且由调用StartWorkflow的调用方法来捕获异常。

public async Task MyMethodThatStartsAWorkflow()
{
    try
    {
        await host.StartWorkflow("HelloWorld", 1, null);
    }
    catch(Exception ex)
    {
        // Handle the exception appropriately
    }
}

后流程中间件

在工作流完成之后运行。

//将工作流摘要打印到工作台
public class PrintWorkflowSummaryMiddleware : IWorkflowMiddleware
{
    private readonly ILogger<PrintWorkflowSummaryMiddleware> _log;

    public PrintWorkflowSummaryMiddleware(
        ILogger<PrintWorkflowSummaryMiddleware> log
    )
    {
        _log = log;
    }
	//后中间件声明
    public WorkflowMiddlewarePhase Phase =>
        WorkflowMiddlewarePhase.PostWorkflow;

    public Task HandleAsync(
        WorkflowInstance workflow,
        WorkflowDelegate next
    )
    {
        if (!workflow.CompleteTime.HasValue)
        {
            return next();
        }
		//持续时间
        var duration = workflow.CompleteTime.Value - workflow.CreateTime;
        _log.LogInformation($@"Workflow {workflow.Description} completed in {duration:g}");
		//每个步骤
        foreach (var step in workflow.ExecutionPointers)
        {
            var stepName = step.StepName;
            var stepDuration = (step.EndTime - step.StartTime) ?? TimeSpan.Zero;
            _log.LogInformation($"  - Step {stepName} completed in {stepDuration:g}");
        }

        return next();
    }
}

后中间件的异常处理

在后中间件执行时，工作流已经完成。默认情况下，工作流中间件引发异常，会记录该异常，并且工作流会照常执行完成，但是可以设置更改。要更改工作流默认的后中间件错误处理，要在依赖关系注入框架中注册IWorkflowMiddlewareErrorHandler。

//中间件异常处理
public class CustomHandler : IWorkflowMiddlewareErrorHandler
{
    public Task HandleAsync(Exception ex)
    {
        // Handle your error asynchronously
    }
}
services.AddWorkflow();
//注册
services.AddTransient<IWorkflowMiddlewareErrorHandler, CustomHandler>();

注册中间件

要想使用中间件，需将中间件注册到容器中。

public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        ...
        // Add workflow middleware
        services.AddWorkflowMiddleware<AddDescriptionWorkflowMiddleware>();
        services.AddWorkflowMiddleware<PrintWorkflowSummaryMiddleware>();
        // Add step middleware
        services.AddWorkflowStepMiddleware<LogCorrelationStepMiddleware>();
        services.AddWorkflowStepMiddleware<PollyRetryMiddleware>();
        ...
    }
}

案例

• steps

public class Log1 : StepBodyAsync
{
    public override Task<ExecutionResult> RunAsync(IStepExecutionContext context)
    {
        Console.WriteLine("Log1:步骤执行");
        return Task.FromResult(ExecutionResult.Next());
    }
}
public class Log2 : StepBodyAsync
{
    public override Task<ExecutionResult> RunAsync(IStepExecutionContext context)
    {
        Console.WriteLine("Log2:步骤执行");
        return Task.FromResult(ExecutionResult.Next());
    }
}

• workflow

public class FlakyConnectionWorkflow : IWorkflow
{
    public string Id => "flaky-sample";

    public int Version => 1;

    public void Build(IWorkflowBuilder<object> builder)
    {
        builder
           .StartWith<Log1>()
           .Then<Log2>();
    }
}

• middleware

  • workflowMiddleware

  public class pre : IWorkflowMiddleware
  {
      public WorkflowMiddlewarePhase Phase => WorkflowMiddlewarePhase.PreWorkflow;
      public Task HandleAsync(WorkflowInstance workflow, WorkflowDelegate next)
      {
          System.Console.WriteLine("前流程中间件1");
          return next();
      }
  }
  
  public class post : IWorkflowMiddleware
  {
      public WorkflowMiddlewarePhase Phase => WorkflowMiddlewarePhase.PostWorkflow;
  
      public Task HandleAsync(WorkflowInstance workflow, WorkflowDelegate next)
      {
          Console.WriteLine("后中间件");
          return next();
      }
  }

  • WorkflowStepMiddleware

  internal class stepM1 : IWorkflowStepMiddleware
  {
      public async Task<ExecutionResult> HandleAsync(IStepExecutionContext context, IStepBody body, WorkflowStepDelegate next)
      {
          Console.WriteLine("步骤中间件1");
          return await next();
      }
  }
  internal class stepM2 : IWorkflowStepMiddleware
  {
      public async Task<ExecutionResult> HandleAsync(IStepExecutionContext context, IStepBody body, WorkflowStepDelegate next)
      {
          Console.WriteLine("步骤中间件2");
          return await next();
      }
  }

• program

IServiceCollection services = new ServiceCollection();
services.AddWorkflow();
services.AddWorkflowStepMiddleware<stepM1>();
services.AddWorkflowStepMiddleware<stepM2>();

services.AddWorkflowMiddleware<pre>();
services.AddWorkflowMiddleware<pre2>();
services.AddWorkflowMiddleware<post>();
services.AddLogging();
var serviceProvider = services.BuildServiceProvider();
var host = serviceProvider.GetService<IWorkflowHost>();
host.RegisterWorkflow<FlakyConnectionWorkflow>();
host.Start();

host.StartWorkflow("flaky-sample");
Console.ReadLine();