DataflowEx 永远不会完成

Question

我正在尝试将开源库 DataflowEx 与下一个 Dataflow 声明一起使用。

class RequestClientFlow :Dataflow<string>{
    private readonly ILogger _logger;
    private readonly Dataflow<string, WebProxy> _webproxyDataflow;
    private readonly Dataflow<WebProxy, HttpClient> _httpClientDataflow;

    public RequestClientFlow(ILogger logger) : this(DataflowOptions.Default){
        _logger = logger;
    }

    public Dataflow<WebProxy, HttpClient> HttpClientDataflow => _httpClientDataflow;

    public RequestClientFlow(DataflowOptions dataflowOptions) : base(dataflowOptions){
        _webproxyDataflow = new TransformBlock<string,WebProxy>(s => {
            _logger.WriteLine("aaaa");
            return new WebProxy();
        }).ToDataflow();
        _httpClientDataflow = new TransformBlock<WebProxy,HttpClient>(proxy => {
            _logger.WriteLine("bbbb");
            return new HttpClient();
        }).ToDataflow();
        _webproxyDataflow.LinkTo(_httpClientDataflow);
        RegisterChild(_webproxyDataflow);
        RegisterChild(_httpClientDataflow);
    }

    public override ITargetBlock<string> InputBlock => _webproxyDataflow.InputBlock;
}

当我像消费它时

var requestClientFlow = new RequestClientFlow(this);
requestClientFlow.Post("");
requestClientFlow.Complete();
await requestClientFlow.InputBlock.Completion;

完成并显示我的输出

18:32:54.3773|aaaa 18:32:54.3773|bbbb

1 次通过，0 次失败，0 次跳过，耗时 1.45 秒（xUnit.net 2.3.1 构建 3858)。

但是我的理解来自我也应该能够使用的框架文档

    requestClientFlow.Complete();
    await requestClientFlow.CompletionTask;

甚至

await requestClientFlow.SignalAndWaitForCompletionAsync();

它没有完成。有人可以帮我理解我做错了什么吗？

Answer 1

您的流程无法完成，因为最后一个块是 TransformBlock。在您的第一个示例中，您 await 完成了实际上完成的 Input 块。 Output 块无法完成，因为其输出缓冲区中的项目无处可去。 DataflowEx 库在流的最后一个块上是正确的 awaiting。您可以在末尾添加ActionBlock 或NullTarget 以实现补全。

就DataflowEx而言，最终流程应该正在实现

public interface IDataflow<in TIn> : IDataflow
{
    ITargetBlock<TIn> InputBlock { get; }
}

正如图书馆 github 页面上的示例所示：

public class AggregatorFlow : Dataflow<string>
{
    //...//

    public AggregatorFlow() : base(DataflowOptions.Default)
    {
        _splitter = new TransformBlock<string, KeyValuePair<string, int>>(s => this.Split(s));
        _dict = new Dictionary<string, int>();

        //***Note The ActionBlock here***
        _aggregater = new ActionBlock<KeyValuePair<string, int>>(p => this.Aggregate(p));

        //Block linking
        _splitter.LinkTo(_aggregater, new DataflowLinkOptions() { PropagateCompletion = true });

        /* IMPORTANT */
        RegisterChild(_splitter);
        RegisterChild(_aggregater);
    }

    //...//
}