在 JoinBlock 中，当另一个 Target 被填充时接收一个 Target

Question

我正在将 JoinBlock 连接到 WriteOnceBlock 和 BufferBlock 以填充目标 1 和 2。我的目标是每次 JoinBlock 从 BufferBlock 收到消息时，它也会请求 WriteOnceBlock 持有的值。

我的第一个猜测是我可以向 Target2 Completion 事件添加一个 ContinueWith 委托，但这不太正确 - 我需要附加到似乎不存在的 Filled 事件之类的东西。

我也尝试在非贪婪模式下使用连接块作为最后的努力，但这并没有改变输出。

我是不是漏掉了什么明显的东西？

示例：

static void Main(string[] args)
    {
        var writeOnceBlockTest = new WriteOnceBlock<int>(i => i);
        var queueBlockTest = new BufferBlock<string>();
        var joinBlockTest = new JoinBlock<int, string>();
        var actionBlockTest = new ActionBlock<System.Tuple<int, string>>(tuple => Console.WriteLine($"I received int {tuple.Item1} and string {tuple.Item2}."));

        writeOnceBlockTest.LinkTo(joinBlockTest.Target1);
        queueBlockTest.LinkTo(joinBlockTest.Target2, new DataflowLinkOptions{PropagateCompletion = true});
        joinBlockTest.LinkTo(actionBlockTest, new DataflowLinkOptions { PropagateCompletion = true });

        writeOnceBlockTest.Post(3);
        queueBlockTest.Post("String1");
        queueBlockTest.Post("String2");
        writeOnceBlockTest.Post(4);
        writeOnceBlockTest.Post(5);
        queueBlockTest.Post("String3");
        queueBlockTest.Post("String4");
        queueBlockTest.Complete();
        Console.ReadLine();
    }

期望输出：

I received int 3 and string String1.

I received int 3 and string String2.

I received int 3 and string String3.

I received int 3 and string String4.

实际输出：

I received int 3 and string String1.

Answer 1

JoinBlock 在这里不是正确的选择，尽管它看起来确实很合适。正如您所发现的，WriteOnceBlock 仅提供一次它的价值。但是，您可以多次读取该值。有了这个，您可以使用 TransformBlock 来获得您想要的行为。

public class JoinFlow
{
    [Test]
    public async Task TestWriteOnceBlock()
    {
        var writeOnceBlockTest = new WriteOnceBlock<int>(i => i);
        var queueBlockTest = new BufferBlock<string>();
        var transformBlockTest = new TransformBlock<string, Tuple<int, string>>(async str => Tuple.Create(await writeOnceBlockTest.ReceiveAsync(), str));
        var actionBlockTest = new ActionBlock<Tuple<int, string>>(tuple => Console.WriteLine($"I received int {tuple.Item1} and string {tuple.Item2}."));

        queueBlockTest.LinkTo(transformBlockTest, new DataflowLinkOptions { PropagateCompletion = true });
        transformBlockTest.LinkTo(actionBlockTest, new DataflowLinkOptions { PropagateCompletion = true });

        writeOnceBlockTest.Post(3);
        queueBlockTest.Post("String1");
        queueBlockTest.Post("String2");
        writeOnceBlockTest.Post(4);
        writeOnceBlockTest.Post(5);
        queueBlockTest.Post("String3");
        queueBlockTest.Post("String4");
        queueBlockTest.Complete();
        await actionBlockTest.Completion;
    }
}

输出：

I received int 3 and string String1.
I received int 3 and string String2.
I received int 3 and string String3.
I received int 3 and string String4.

在 JoinBlock 中，当另一个 Target 被填充时接收一个 Target

In a JoinBlock, receive a Target when the other Target is filled

.net

c#

tpl-dataflow