Data block for combining one result with several other results

Question

In my application, I want to combine several lines with a dictionary of replacement values.

readTemplateBlockreceives a FileInfos file and returns their contents as a string.
getReplacersBlockreceives (once) one substitute dictionary.
joinTemplateAndReplacersBlockmust be attached to each element readTemplateBlockwith one result getReplacersBlock.

In my current setup, I need to send the same substitute dictionary again for every file that I publish.

// Build
var readTemplateBlock = new TransformBlock<FileInfo, string>(file => File.ReadAllText(file.FullName));
var getReplacersBlock = new WriteOnceBlock<IDictionary<string, string>>(null);
var joinTemplateAndReplacersBlock = new JoinBlock<string, IDictionary<string, string>>();

// Assemble
var propagateComplete = new DataflowLinkOptions {PropagateCompletion = true};

readTemplateBlock.LinkTo(joinTemplateAndReplacersBlock.Target1, propagateComplete);
getReplacersBlock.LinkTo(joinTemplateAndReplacersBlock.Target2, propagateComplete);
joinTemplateAndReplacersBlock.LinkTo(replaceTemplateBlock, propagateComplete);

// Post
foreach (var template in templateFilenames)
{
    getFileBlock.Post(template);
}
getFileBlock.Complete();

getReplacersBlock.Post(replacers);
getReplacersBlock.Complete();

Is there a better block that I am missing? Maybe a configuration option that I missed?

Answer 1

, , . , :

• BufferBlock BoundedCapacity Task, . Task , WriteOnceBlock, :

  static IPropagatorBlock<T, T> CreateWriteOnceRepeaterBlock<T>()
  {
      var target = new WriteOnceBlock<T>(null);
      var source = new BufferBlock<T>(new DataflowBlockOptions { BoundedCapacity = 1 });
  
      Task.Run(
          async () =>
          {
              var value = await target.ReceiveAsync();
  
              while (true)
              {
                  await source.SendAsync(value);
              }
          });
  
      return DataflowBlock.Encapsulate(target, source);
  }
  

  CreateWriteOnceRepeaterBlock<IDictionary<string, string>>() new WriteOnceBlock<IDictionary<string, string>>(null).

• , WriteOnceBlock, , . , WriteOnceBlock, , , .

• TaskCompletionSource .

  , ( # 7 System.ValueTuple ):

  void ReplaceTemplateBlockAction(Tuple<string, IDictionary<string, string>> tuple)
  {
      var (template, replacers) = tuple;
      …
  }
  
  …
  
  var getReplacersBlock = new WriteOnceBlock<IDictionary<string, string>>(null);
  var replaceTemplateBlock = new ActionBlock<Tuple<string, IDictionary<string, string>>>(
      ReplaceTemplateBlockAction);
  …
  getReplacersBlock.Post(replacers);
  

  :

  void ReplaceTemplateBlockAction(string template, IDictionary<string, string>>> replacers)
  {
      …
  }
  
  …
  
  var getReplacersTcs = new TaskCompletionSource<IDictionary<string, string>>();
  var replaceTemplateBlock = new ActionBlock<string>(
      async template => ReplaceTemplateBlockAction(template, await getReplacersTcs.Task));
  …
  getReplacersTcs.SetResult(replacers);