Я пытаюсь использовать поток данных TPL для создания конвейера. Пока все работает нормально, а мой конвейер определяется следующим образом (хотя моя проблема связана только с вещателем, представлениеСуществует, отправлено):
// Define tasks
var productListingBatchBuffer = new BufferBlock<PostSubmissionState>();
var splitFile = new TransformBlock<PostSubmissionState, PostSubmissionState>(s => SplitFile(s));
var saveFile = new TransformBlock<PostSubmissionState, PostSubmissionState>(s => SaveFile(s));
var postSubmission = new TransformBlock<PostSubmissionState, PostSubmissionState>(s => PostSubmission(s));
var broadcaster = new BroadcastBlock<PostSubmissionState>(state => state);
var submissionSucceeded = new ActionBlock<PostSubmissionState>(s => SubmissionSucceeded(s));
var submissionFailed = new ActionBlock<PostSubmissionState>(s => SubmissionFailed(s));
// Link em up
productListingBatchBuffer.LinkTo(splitFile, new DataflowLinkOptions() { PropagateCompletion = true });
splitFile.LinkTo(saveFile, new DataflowLinkOptions() { PropagateCompletion = true });
saveFile.LinkTo(postSubmission, new DataflowLinkOptions() { PropagateCompletion = true });
postSubmission.LinkTo(broadcaster, new DataflowLinkOptions() { PropagateCompletion = true });
broadcaster.LinkTo(submissionSucceeded, new DataflowLinkOptions() { PropagateCompletion = true }, state => state.PostSucceeded);
broadcaster.LinkTo(submissionFailed, new DataflowLinkOptions() { PropagateCompletion = true }, state => !state.PostSucceeded);
Проблема, которую я имею, связана с распространением Исключений. Поскольку мой BroadcastBlock распространяет свое завершение (и, следовательно, любой Fault) на два блока, если исключение действительно происходит, оно распространяется на оба блока. Таким образом, когда я делаю
Task.WaitAll(submissionSucceeded.Completion, submissionFailed.Completion);
Я получаю сводное исключение, содержащее два исключения. Прямо сейчас самое лучшее, что я могу сделать, это отфильтровать их, т.е.:
try
{
Task.WaitAll(submissionSucceeded.Completion, submissionFailed.Completion);
}
catch (AggregateException ex)
{
var uniqueExceptions = new AggregateException(ex.Flatten().InnerExceptions.Distinct());
Console.WriteLine("An exception was thrown.\n{0}", uniqueExceptions.Flatten());
}
но мне интересно, есть ли лучший способ сделать это. то есть, если возникает только одно исключение, я хочу, чтобы одно исключение было поднято. Я новичок в Dataflow, поэтому просто обнаруживаю все соглашения.