Использование AsObservable для наблюдения за блоками потока данных TPL без использования сообщений

Проблема с вашим кодом заключается в том, что вы подключаете двух потребителей block1. Затем поток данных просто дает значение тому, какой потребитель появляется первым.

Поэтому вам нужно транслировать значения из block1 в два других блока, чтобы затем иметь возможность потреблять их независимо.

Просто примечание: не делайте .Publish().RefCount(), так как это не делает то, что вы думаете. Это эффективно сделает один запуск доступным только для наблюдения, что во время этого одного запуска позволит нескольким наблюдателям подключаться и видеть одни и те же значения. Это не имеет никакого отношения ни к источнику данных, ни к тому, как взаимодействуют блоки потока данных.

Попробуйте этот код:

// Set up mesh
var block1 = new TransformBlock<int, int>(i => i + 20);
var block_boadcast = new BroadcastBlock<int>(i => i, new DataflowBlockOptions());
var block_buffer = new System.Threading.Tasks.Dataflow.BufferBlock<int>();
var block2 = new ActionBlock<int>(i => Debug.Print("block2:" + i.ToString()));
var obs = block_buffer.AsObservable();
var l1 = block1.LinkTo(block_boadcast);
var l2 = block_boadcast.LinkTo(block2);
var l3 = block_boadcast.LinkTo(block_buffer);

// Progress
obs.Subscribe(i => Debug.Print("progress:" + i.ToString()));

// Start
var vals = Enumerable.Range(1, 5);
foreach (var v in vals)
{
    block1.Post(v);
}
block1.Complete();

Это дает мне:

block2:21
block2:22
block2:23
block2:24
block2:25
progress:21
progress:22
progress:23
progress:24
progress:25

Что, я думаю, ты и хотел.

Кроме того, использование Rx для этого может быть лучшим вариантом во всех отношениях. Это намного мощнее и декларативнее, чем любой вариант TPL или Dataflow.

Ваш код сводится к следующему:

Observable
    .Range(1, 5)
    .Select(i => i + 20)
    .Do(i => Debug.Print("progress:" + i.ToString()));
    .Subscribe(i => Debug.Print("block2:" + i.ToString()));

Это в значительной степени дает вам тот же результат.

Есть два варианта, которые следует учитывать при создании наблюдаемого блока потока данных. Вы также можете:

1. выдавать уведомление каждый раз, когда сообщение обрабатывается, или
2. выдавать уведомление, когда ранее обработанное сообщение, хранящееся в выходном буфере блока, принимается связанным блоком.

Оба варианта имеют плюсы и минусы. Первый вариант обеспечивает своевременные, но неупорядоченные уведомления. Второй вариант обеспечивает упорядоченные, но отложенные уведомления, а также должен иметь дело с одноразовой связью между блоками. Что должно произойти с наблюдаемым, когда связь между двумя блоками удаляется вручную до того, как блоки будут завершены?

Ниже приведена реализация первого варианта, который создает TransformBlock вместе с непотребляющим IObservable этого блока. Также существует реализация для ActionBlock эквивалента, основанная на первой реализации (хотя ее можно было бы реализовать и самостоятельно путем копирования-вставки и адаптации реализации TransformBlock, так как кода не так много).

public static TransformBlock<TInput, TOutput>
    CreateObservableTransformBlock<TInput, TOutput>(
    Func<TInput, Task<TOutput>> transform,
    out IObservable<(TInput Input, TOutput Output,
        int StartedIndex, int CompletedIndex)> observable,
    ExecutionDataflowBlockOptions dataflowBlockOptions = null)
{
    if (transform == null) throw new ArgumentNullException(nameof(transform));
    dataflowBlockOptions = dataflowBlockOptions ?? new ExecutionDataflowBlockOptions();

    var semaphore = new SemaphoreSlim(1);
    int startedIndexSeed = 0;
    int completedIndexSeed = 0;

    var notificationsBlock = new BufferBlock<(TInput, TOutput, int, int)>(
        new DataflowBlockOptions() { BoundedCapacity = 100 });

    var transformBlock = new TransformBlock<TInput, TOutput>(async item =>
    {
        var startedIndex = Interlocked.Increment(ref startedIndexSeed);
        var result = await transform(item).ConfigureAwait(false);
        await semaphore.WaitAsync().ConfigureAwait(false);
        try
        {
            // Send the notifications in synchronized fashion
            var completedIndex = Interlocked.Increment(ref completedIndexSeed);
            await notificationsBlock.SendAsync(
                (item, result, startedIndex, completedIndex)).ConfigureAwait(false);
        }
        finally
        {
            semaphore.Release();
        }
        return result;
    }, dataflowBlockOptions);

    _ = transformBlock.Completion.ContinueWith(t =>
    {
        if (t.IsFaulted) ((IDataflowBlock)notificationsBlock).Fault(t.Exception);
        else notificationsBlock.Complete();
    }, TaskScheduler.Default);

    observable = notificationsBlock.AsObservable();
    // A dummy subscription to prevent buffering in case of no external subscription.
    observable.Subscribe(
        DataflowBlock.NullTarget<(TInput, TOutput, int, int)>().AsObserver());
    return transformBlock;
}

// Overload with synchronous lambda
public static TransformBlock<TInput, TOutput>
    CreateObservableTransformBlock<TInput, TOutput>(
    Func<TInput, TOutput> transform,
    out IObservable<(TInput Input, TOutput Output,
        int StartedIndex, int CompletedIndex)> observable,
    ExecutionDataflowBlockOptions dataflowBlockOptions = null)
{
    return CreateObservableTransformBlock(item => Task.FromResult(transform(item)),
        out observable, dataflowBlockOptions);
}

// ActionBlock equivalent (requires the System.Reactive package)
public static ITargetBlock<TInput>
    CreateObservableActionBlock<TInput>(
    Func<TInput, Task> action,
    out IObservable<(TInput Input, int StartedIndex, int CompletedIndex)> observable,
    ExecutionDataflowBlockOptions dataflowBlockOptions = null)
{
    if (action == null) throw new ArgumentNullException(nameof(action));
    var block = CreateObservableTransformBlock<TInput, object>(
        async item => { await action(item).ConfigureAwait(false); return null; },
        out var sourceObservable, dataflowBlockOptions);
    block.LinkTo(DataflowBlock.NullTarget<object>());
    observable = sourceObservable
        .Select(entry => (entry.Input, entry.StartedIndex, entry.CompletedIndex));
    return block;
}

// ActionBlock equivalent with synchronous lambda
public static ITargetBlock<TInput>
    CreateObservableActionBlock<TInput>(
    Action<TInput> action,
    out IObservable<(TInput Input, int StartedIndex, int CompletedIndex)> observable,
    ExecutionDataflowBlockOptions dataflowBlockOptions = null)
{
    return CreateObservableActionBlock(
        item => { action(item); return Task.CompletedTask; },
        out observable, dataflowBlockOptions);
}

Пример использования в Windows Forms:

private async void Button1_Click(object sender, EventArgs e)
{
    var block = CreateObservableTransformBlock((int i) => i + 20,
        out var observable,
        new ExecutionDataflowBlockOptions() { BoundedCapacity = 1 });

    var vals = Enumerable.Range(1, 20).ToList();
    TextBox1.Clear();
    ProgressBar1.Value = 0;

    observable.ObserveOn(SynchronizationContext.Current).Subscribe(onNext: x =>
    {
        TextBox1.AppendText($"Value {x.Input} transformed to {x.Output}\r\n");
        ProgressBar1.Value = (x.CompletedIndex * 100) / vals.Count;
    }, onError: ex =>
    {
        TextBox1.AppendText($"An exception occured: {ex.Message}\r\n");
    },
    onCompleted: () =>
    {
        TextBox1.AppendText("The job completed successfully\r\n");
    });

    block.LinkTo(DataflowBlock.NullTarget<int>());

    foreach (var i in vals) await block.SendAsync(i);
    block.Complete();
}

В приведенном выше примере тип переменной observable:

IObservable<(int Input, int Output, int StartedIndex, int CompletedIndex)>

Два индекса отсчитываются от 1.

Попробуйте заменить:

obs.ForEachAsync(i => Debug.Print("progressBlock:" + i.ToString()));

с участием:

obs.Subscribe(i => Debug.Print("progressBlock:" + i.ToString()));

Я полагаю, что метод ForEachAsync не подключается должным образом/он срабатывает, но что-то странное происходит с асинхронной частью.

Указав BoundedCapacity для блока внутри цепочки, вы создаете ситуацию, когда некоторые из ваших сообщений отклоняются целевыми блоками, поскольку буфер для ActionBlock заполнен, и сообщение отклоняется.

Создавая наблюдаемое из вашего буферного блока, вы создаете условие гонки: два потребителя ваших данных получают сообщения одновременно. Блоки в TPL Dataflow передают данные первому доступному потребителю, что приводит к недетерминированному состоянию приложения.

Теперь вернемся к вашей проблеме. Вы можете ввести BroadcastBlock, так как он предоставляет копию данных всем потребителям, а не только первому, но в этом случае вам нужно снять ограничение на размер буфера, блок вещания похож на телевизионный канал. , вы не можете получить предыдущее шоу, у вас есть только текущее.

Дополнительные примечания: вы не проверяете возвращаемое значение метода Post, вы можете рассмотреть использование await SendAsync, и для лучшего эффекта регулирования установите BoundedCapacity для блока начальной точки, а не для среднего.