将单个结果与多个其他结果连接起来的数据块

Question

在我的应用程序中，我想将多个字符串与替换值字典连接起来。

readTemplateBlock 获取 FileInfos 和 returns 它们的内容作为字符串。
getReplacersBlock 得到（一次）单个替换字典。
joinTemplateAndReplacersBlock 应该将 readTemplateBlock 的每一项与一个 getReplacersBlock 结果连接起来。

在我当前的设置中，它要求我 post 为我 post.

的每个文件再次使用相同的替换字典

// Build
var readTemplateBlock = new TransformBlock<FileInfo, string>(file => File.ReadAllText(file.FullName));
var getReplacersBlock = new WriteOnceBlock<IDictionary<string, string>>(null);
var joinTemplateAndReplacersBlock = new JoinBlock<string, IDictionary<string, string>>();

// Assemble
var propagateComplete = new DataflowLinkOptions {PropagateCompletion = true};

readTemplateBlock.LinkTo(joinTemplateAndReplacersBlock.Target1, propagateComplete);
getReplacersBlock.LinkTo(joinTemplateAndReplacersBlock.Target2, propagateComplete);
joinTemplateAndReplacersBlock.LinkTo(replaceTemplateBlock, propagateComplete);

// Post
foreach (var template in templateFilenames)
{
    getFileBlock.Post(template);
}
getFileBlock.Complete();

getReplacersBlock.Post(replacers);
getReplacersBlock.Complete();

我还缺少更好的方块吗？也许我忽略了一个配置选项？

Answer 1

我不知道如何使用内置的数据流块来做到这一点。我可以看到的替代方案：

使用带有小 BoundedCapacity 的 BufferBlock 以及不断向其发送值的 Task。 Task 究竟如何获取值可能会有所不同，但如果你喜欢 WriteOnceBlock，你可以重用并封装它：

static IPropagatorBlock<T, T> CreateWriteOnceRepeaterBlock<T>()
{
    var target = new WriteOnceBlock<T>(null);
    var source = new BufferBlock<T>(new DataflowBlockOptions { BoundedCapacity = 1 });

    Task.Run(
        async () =>
        {
            var value = await target.ReceiveAsync();

            while (true)
            {
                await source.SendAsync(value);
            }
        });

    return DataflowBlock.Encapsulate(target, source);
}

然后您将使用 CreateWriteOnceRepeaterBlock<IDictionary<string, string>>() 而不是 new WriteOnceBlock<IDictionary<string, string>>(null)。

编写一个类似于 WriteOnceBlock 的自定义块，其行为完全符合您的要求。看看the source of WriteOnceBlock有多大，这大概不太吸引人吧

为此使用 TaskCompletionSource 而不是数据流块。

假设您当前的代码如下所示（为简洁起见，使用 C# 7 和 System.ValueTuple 包）：

void ReplaceTemplateBlockAction(Tuple<string, IDictionary<string, string>> tuple)
{
    var (template, replacers) = tuple;
    …
}

…

var getReplacersBlock = new WriteOnceBlock<IDictionary<string, string>>(null);
var replaceTemplateBlock = new ActionBlock<Tuple<string, IDictionary<string, string>>>(
    ReplaceTemplateBlockAction);
…
getReplacersBlock.Post(replacers);

您可以改用：

void ReplaceTemplateBlockAction(string template, IDictionary<string, string>>> replacers)
{
    …
}

…

var getReplacersTcs = new TaskCompletionSource<IDictionary<string, string>>();
var replaceTemplateBlock = new ActionBlock<string>(
    async template => ReplaceTemplateBlockAction(template, await getReplacersTcs.Task));
…
getReplacersTcs.SetResult(replacers);

将单个结果与多个其他结果连接起来的数据块

A datablock to join a single result with multiple other results

.net

c#

tpl-dataflow