TPL DataFlow: создание пользовательского splitblock

Question

просто нужна помощь в создании настраиваемого splitblock с помощью библиотеки потока данных, которая является частью TPL в .Net.

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

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

public abstract class SplitBlock<TInput, TOutputLeft, TOutputRight> 

Моя проблема заключается в том, что я не действительно знаю, как идти дальше. Все, что я знаю, что мне нужно два TransformBlocks:

var leftBlock = new TransformBlock<TInput, TOutputLeft>(...) 
var rightblock = new TransformBlock<TInput, TOutputRight>(...) 

Во всех моих попытках я закончил тем, что несколько ITargetBlocks для хранения на входе левого и правого блок, но это не может быть права, не так ли?

Я ценю каждый намек, который вы можете дать.

Answer 1

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

public class SplitBlock<TInput, TOutputLeft, TOutputRight> 
{ 
    public ITargetBlock<TInput> Input { get; } 
    public ISourceBlock<TOutputLeft> LeftOutput { get; } 
    public ISourceBlock<TOutputRight> RightOutput { get; } 
} 

С, что реализация следует естественным образом: один входной блок, соединенный с двумя выходными блоками. Единственный вопрос заключается в том, следует ли делать фактическую обработку в выходных блоках (например, вы предложили свои два TransformBlock s) или во входном блоке.

Если вы хотите иметь обработку в выходных блоках, входной блок может быть ActionBlock, который отправляет вход на оба выхода, а выходы будут TransformBlock s, как вы предложили.

public class SplitBlock<TInput, TOutputLeft, TOutputRight> 
{ 
    private ActionBlock<TInput> input; 
    private TransformBlock<TInput, TOutputLeft> leftOutput; 
    private TransformBlock<TInput, TOutputRight> rightOutput; 

    public ITargetBlock<TInput> Input { get { return input; } } 
    public ISourceBlock<TOutputLeft> LeftOutput { get { return leftOutput; } } 
    public ISourceBlock<TOutputRight> RightOutput { get { return rightOutput; } } 

    public SplitBlock(
     Func<TInput, TOutputLeft> leftTransform, 
     Func<TInput, TOutputRight> rightTransform) 
    { 
     input = new ActionBlock<TInput>(
      x => 
      { 
       leftOutput.Post(x); 
       rightOutput.Post(x); 
      }); 
     leftOutput = new TransformBlock<TInput, TOutputLeft>(leftTransform); 
     rightOutput = new TransformBlock<TInput, TOutputRight>(rightTransform); 

     // TODO handle fault in input correctly 
     input.Completion.ContinueWith(
      _ => 
      { 
       leftOutput.Complete(); 
       rightOutput.Complete(); 
      }); 
    } 
} 

(Это предполагает, что левые и правые преобразования может обрабатывать один и тот же вход в то же время.)

С другой стороны, если вы хотите, чтобы выполнить обработку входного блока (что делает более смысл для меня), то вы могли бы ActionBlock в качестве входных данных и BufferBlock с, как выходы, с входным блоком обработки ввода и затем отправить результаты на выходных блоков:

public class SplitBlock<TInput, TOutputLeft, TOutputRight> 
{ 
    private ActionBlock<TInput> input; 
    private BufferBlock<TOutputLeft> leftOutput; 
    private BufferBlock<TOutputRight> rightOutput; 

    public ITargetBlock<TInput> Input { get { return input; } } 
    public ISourceBlock<TOutputLeft> LeftOutput { get { return leftOutput; } } 
    public ISourceBlock<TOutputRight> RightOutput { get { return rightOutput; } } 

    public SplitBlock(
     Func<TInput, Tuple<TOutputLeft, TOutputRight>> combinedTransform) 
    { 
     input = new ActionBlock<TInput>(
      value => 
      { 
       var result = combinedTransform(value); 
       leftOutput.Post(result.Item1); 
       rightOutput.Post(result.Item2); 
      }); 
     leftOutput = new BufferBlock<TOutputLeft>(); 
     rightOutput = new BufferBlock<TOutputRight>(); 

     // TODO handle fault in input correctly 
     input.Completion.ContinueWith(
      _ => 
      { 
       leftOutput.Complete(); 
       rightOutput.Complete(); 
      }); 
    } 

    public SplitBlock(
     Func<TInput, TOutputLeft> leftTransform, 
     Func<TInput, TOutputRight> rightTransform) 
     : this(x => Tuple.Create(leftTransform(x), rightTransform(x))) 
    {} 
}