(C#) TPL Dataflow 라이브러리

_stockInfoBroadCaster.Complete();
try
{
    _stockInfoBroadCaster.Completion.Wait();
}
catch (Exception ex)
{
    // 데이터흐름 완료 오류
}

private BroadcastBlock<List<StockInfoModel>> _broadcast;

public MainViewModel()
{
        // 브로드캐스트 생성
        _broadcast = new BroadcastBlock<List<StockInfoModel>>(null);
}

private async Task MonitoringExecute(object param)
{
        // 브로드캐스팅된 블록(주식정보)을 수신받아
        // 정보 표시 ViewModel에 데이터 전달 역할을 하는 블록 생성
        ActionBlock<List<StockInfoModel>> updateStockInfoStart = this.UpdateStockInfoStart();
        // 브로드캐스트에 ActionBlock 연결
        _broadcast.LinkTo(updateStockInfoStart);

        // 브로드캐스트에 추가 블록을 연결할 수 있다.
        ActionBlock<List<StockInfoModel>> foo = Foo();
        _broadcast.LinkTo(foo);

        // 브로드캐스팅으로 데이터를 수신받고
        // 가공된 데이터를 ViewModel에 전달 역할을 하는 블록 생성
        ActionBlock<Tuple<double, double>> updateChangeCalc = this.UpdateChangeCalc();
        var changeCalc = new TransformBlock<List<StockInfoModel>, Tuple<double, double>>(p =>
        {
            var stockInfo1_currPrice = p[0].CurrentPrice;
            var stockInfo2_currPrice = p[1].CurrentPrice;

            double percentChangeByItem1 = (stockInfo1_currPrice - StockInfo1VM.PreviousPrice) * 100 / StockInfo1VM.PreviousPrice;
            double percentChangeByItem2 = (stockInfo2_currPrice - StockInfo2VM.PreviousPrice) * 100 / StockInfo2VM.PreviousPrice;

            StockInfo1VM.PreviousPrice = stockInfo1_currPrice;
            StockInfo2VM.PreviousPrice = stockInfo2_currPrice;

            return new Tuple<double, double>(percentChangeByItem1, percentChangeByItem2);
        });

        // 브로드캐스트에 TransformBlock 연결
        _broadcast.LinkTo(changeCalc);
        // TransformBlock에 ActionBlock 연결
        changeCalc.LinkTo(updateChangeCalc);

        // 브로드캐스팅 시작
        StockInfoBroadCast stockInfoBroadCast = new(_broadcast);
        await stockInfoBroadCast.RealTimeData();
}

private ActionBlock<List<StockInfoModel>> UpdateStockInfoStart()
{
        return new ActionBlock<List<StockInfoModel>>(p => {
            StockInfo1VM.UpdateStockInfo(p[0]);
            StockInfo2VM.UpdateStockInfo(p[1]);
        },
        new ExecutionDataflowBlockOptions()
        {
            TaskScheduler =
            TaskScheduler.FromCurrentSynchronizationContext()
        });
}

private ActionBlock<List<StockInfoModel>> Foo()
{
        return new ActionBlock<List<StockInfoModel>>(p => { /* do something */ },
        new ExecutionDataflowBlockOptions()
            {
                TaskScheduler =
                TaskScheduler.FromCurrentSynchronizationContext()
            });
}

private ActionBlock<Tuple<double, double>> UpdateChangeCalc()
{
        return new ActionBlock<Tuple<double, double>>(p =>
        {
            StockInfo1VM.PercentChange = p.Item1;
            StockInfo2VM.PercentChange = p.Item2;
        },
        new ExecutionDataflowBlockOptions()
        {
            TaskScheduler =
            TaskScheduler.FromCurrentSynchronizationContext()
        });
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using TPL_DataFlowBroadCasting_Example.Models;

namespace TPL_DataFlowBroadCasting_Example;

public class StockInfoBroadCast
{
    private readonly BroadcastBlock<List<StockInfoModel>> _stockInfoBroadCaster;

    public StockInfoBroadCast(BroadcastBlock<List<StockInfoModel>> broadcaster)
    {
        _stockInfoBroadCaster = broadcaster;
    }

    public async Task RealTimeData()
    {
        var random1 = new Random();
        var random2 = new Random();

        StockInfoModel stockInfo1 = new()
        {
            StockName = "삼성전자",
            StartPrice = random1.Next(1000, 10000)
        };

        StockInfoModel stockInfo2 = new()
        {
            StockName = "삼성전자우",
            StartPrice = random2.Next(1000, 10000)
        };

        for (int i = 0; i < 100; i++)
        {
            await Task.Delay(1200);
            stockInfo1.CurrentPrice = random1.Next(1000, 10000);
            stockInfo2.CurrentPrice = random2.Next(1000, 10000);
            await _stockInfoBroadCaster.SendAsync(new() { stockInfo1, stockInfo2 });
        }
        _stockInfoBroadCaster.Complete();
        try
        {
            _stockInfoBroadCaster.Completion.Wait();
        }
        catch (Exception ex)
        {
            // 데이터흐름 완료 오류
        }
    }
}