如何根据条件阻止 TransformBlock 处理剩余的排队消息？

Question

以下是在控制台项目中使用 TPL DataFlow 的简单工作流代码。

三个Test对象TestA、TestB和TestC被投递到起始bufferBlock。这链接到一个 TransformBlock，它评估每个测试的 PerformTestAsync() 方法，该方法返回一个 Task<TestResult>。 TransformBlock 与将测试结果写入控制台的 ActionBlock 链接。

所有这些似乎都可以正常工作。但是，我正在努力做的是更改代码，以便第一次await t.PerformTestAsync() 返回TestResult.Failed 我希望 TransformBlock 不再处理消息，当然也不再传递给 ActionBlock，除了失败的结果。因此，对于我的示例代码，我只想在控制台窗口中看到“OK”和“Failed”，而 testC.PerformTestAsync() 则根本不会被调用。

我怎样才能做到这一点？

代码：

class Program
{
    static void Main(string[] args)
    {

        // Create workflow blocks
        var bufferBlock = new BufferBlock<TestBase>();
        var transformBlock = new TransformBlock<TestBase, TestResult>(async t => await t.PerformTestAsync());
        var actionBlock = new ActionBlock<TestResult>(i => Console.WriteLine(i));
        // Link Blocks
        bufferBlock.LinkTo(transformBlock, new DataflowLinkOptions() { PropagateCompletion = true });
        transformBlock.LinkTo(actionBlock, new DataflowLinkOptions() { PropagateCompletion = true });

        // Create Tests
        var tests = new List<TestBase>() { new TestA(), new TestB(), new TestC() };

        // Post them into start of workflow
        foreach (var test in tests)
        {
            bufferBlock.Post<TestBase>(test);
        }
        bufferBlock.Complete();

        actionBlock.Completion.Wait();
        Console.ReadLine();
    }

}


public enum TestResult
{
    OK,
    Error,
    Failed
}

public abstract class TestBase
{
    private readonly string _name;
    public TestBase(string name)
    {
        _name = name;
    }
    public abstract Task<TestResult> PerformTestAsync();

}


public class TestA : TestBase
{
    public TestA() : base("Test A")
    {
    }

    public override Task<TestResult> PerformTestAsync()
    {
        // Do some processing for this test...
        return Task.FromResult(TestResult.OK);
    }
}

public class TestB : TestBase
{
    public TestB() : base("Test B")
    {
    }

    public override Task<TestResult> PerformTestAsync()
    {
        // Do some processing for this test...
        return Task.FromResult(TestResult.Failed);
    }
}

public class TestC : TestBase
{
    public TestC() : base("Test C")
    {
    }

    public override Task<TestResult> PerformTestAsync()
    {
        // Do some processing for this test...
        return Task.FromResult(TestResult.OK);
    }
}

Answer 1

当await t.PerformTestAsync() 返回TestReuslt.Failed 时，抛出异常。这将使流程出错并阻止任何进一步的处理。然后流程将在故障状态下完成。不会处理其他项目。

var transformBlock = new TransformBlock<TestBase, TestResult>(async t =>
{
    var result = await t.PerformTestAsync();
    if (result == TestResult.Failed)
        throw new InvalidOperationException();
    return result;
});

请注意，您抛出的异常将传播到最后一个块的Completion 任务，即您的ActionBlock。当您等待该任务时，您将能够处理错误的流程或根据您的选择忽略。