DataflowBlock ITargetSource.AsObservable() 未触发 OnNext()

Question

我正在尝试使用数据流块，我需要监视通过的项目以进行单元测试。

为了做到这一点，我在我的TransformBlock<Tinput, T> 的ISourceBlock<T> 上使用AsObservable() 方法， 所以我可以在执行后检查我的管道的每个块都生成了预期值。

管道

{
   ...
   var observer = new MyObserver<string>();
   _block  = new TransformManyBlock<string, string>(MyHandler, options);
   _block.LinkTo(_nextBlock);
   _block.AsObservable().Subscribe(observer);
   _block.Post("Test");
   ...
}

我的观察者

public class MyObserver<T> : IObserver<T>
{
    public List<Exception> Errors = new List<Exception>();
    public bool IsComplete = false;
    public List<T> Values = new List<T>();

    public void OnCompleted()
    {
        IsComplete = true;
    }

    public void OnNext(T value)
    {
        Values.Add(value);
    }

    public void OnError(Exception e)
    {
        Errors.Add(e);
    }
}

所以基本上我为我的观察者订阅了转换块，并且我希望通过的每个值都注册到我的观察者“值”列表中。

但是，虽然IsComplete 设置为true，并且OnError() 成功注册异常， OnNext() 方法永远不会被调用，除非它是管道的最后一个块...... 我不知道为什么，因为链接到这个sourceBlock的“nextblock”成功接收到数据，证明有一些数据正在退出该块。

据我了解，AsObservable 应该报告退出块的每个值，而不仅仅是其他链接块未使用的值...

我做错了什么？

Answer 1

_nextBlock 正在使用您的消息，然后您才有机会阅读它们。

如果您注释掉这一行 _block.LinkTo(_nextBlock); 它可能会起作用。

AsObservable 唯一的目的就是让 block 从 RX 被消费。它不会将块的内部工作更改为将消息广播到多个目标。你需要一个特殊的块来处理BroadcastBlock

我建议 广播 到另一个 block 并将其用于Subscribe

BroadcastBlock 的使命是让所有目标链接到 获取每个已发布元素的副本的块

var options = new DataflowLinkOptions {PropagateCompletion = true};


var broadcastBlock = new BroadcastBlock<string>(x => x);
var bufferBlock = new BufferBlock<string>();
var actionBlock = new ActionBlock<string>(s => Console.WriteLine("Action " + s));

broadcastBlock.LinkTo(bufferBlock, options);
broadcastBlock.LinkTo(actionBlock, options);

bufferBlock.AsObservable().Subscribe(s => Console.WriteLine("peek " + s));

for (var i = 0; i < 5; i++)
   await broadcastBlock.SendAsync(i.ToString());

broadcastBlock.Complete();
await actionBlock.Completion;

输出

peek 0
Action 0
Action 1
Action 2
Action 3
Action 4
peek 1
peek 2
peek 3
peek 4