DataflowBlock ITargetSource.AsObservable () nie wyzwala OnNext ()

Question

Próbuję użyć bloku przepływu danych i muszę szpiegować elementy przechodzące do testów jednostkowych.

Aby to zrobić, używam AsObservable()metody on ISourceBlock<T>of my TransformBlock<Tinput, T>, więc po wykonaniu mogę sprawdzić, czy każdy blok mojego potoku wygenerował oczekiwane wartości.

Rurociąg

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

MyObserver

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);
    }
}

Zasadniczo więc subskrybuję mojego obserwatora w bloku transformacji i oczekuję, że każda przechodząca wartość zostanie zarejestrowana na mojej liście „wartości” obserwatorów.

Ale gdy IsCompleteustawiono wartość true i OnError()pomyślnie zarejestrowano wyjątek, OnNext()metoda nigdy nie zostanie wywołana, chyba że jest to ostatni blok potoku ... Nie mogę dowiedzieć się dlaczego, ponieważ „nextblock” połączony z tym sourceBlock pomyślnie odebrać dane, udowadniając, że niektóre dane opuszczają blok.

Z tego, co rozumiem, AsObservablema raportować wszystkie wartości opuszczające blok, a nie tylko wartości, które nie zostały zużyte przez inne połączone bloki ...

Co ja robię źle ?

Answer 1

Twoje wiadomości są pochłaniane , _nextBlockzanim masz szansę je przeczytać.

Jeśli zakomentujesz tę linię _block.LinkTo(_nextBlock);, prawdopodobnie zadziała.

AsObservablejedynym celem jest właśnie umożliwienie blok być spożywane z RX . Nie zmienia to wewnętrznego działania bloku w celu rozgłaszania wiadomości do wielu celów . Potrzebujesz do tego specjalnego blokuBroadcastBlock

Sugerowałbym nadawanie do innego bloku i używanie tego doSubscribe

Misją BroadcastBlock w życiu jest umożliwienie wszystkim docelowym połączonym z bloku uzyskania kopii każdego opublikowanego elementu

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;

Wynik

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