title: Using async/await and yield return with TPL Dataflow
body: <p>I am trying to implement a data processing pipeline using <code>TPL Dataflow</code>. However, I am relatively new to dataflow and not completely sure how to use it properly for the problem I am trying to solve.</p>
  
  <p><strong>Problem</strong>:</p>
  
  <p>I am trying to iterate through the list of files and process each file to read some data and then further process that data. Each file is roughly <code>700MB</code> to <code>1GB</code> in size. Each file contains <code>JSON</code> data. In order to process these files in parallel and not run of of memory, I am trying to use <code>IEnumerable&lt;&gt;</code> with <code>yield return</code> and then further process the data.</p>
  
  <p>Once I get list of files, I want to process maximum 4-5 files at a time in parallel. My confusion comes from:</p>
  
  <ul>
  <li>How to use <code>IEnumerable&lt;&gt;</code> and <code>yeild return</code> with <code>async/await</code> and dataflow. Came across <a href="https://stackoverflow.com/a/14809985">this answer</a> by <a href="https://stackoverflow.com/users/41071/svick">svick</a>, but still not sure how to convert <code>IEnumerable&lt;&gt;</code> to <code>ISourceBlock</code> and then link all blocks together and track completion.</li>
  <li>In my case, <code>producer</code> will be really fast (going through list of files), but <code>consumer</code> will be very slow (processing each file - read data, deserialize <code>JSON</code>). In this case, how to track completion.</li>
  <li>Should I use <code>LinkTo</code> feature of datablocks to connect various blocks? or use method such as <code>OutputAvailableAsync()</code> and <code>ReceiveAsync()</code> to propagate data from one block to another.</li>
  </ul>
  
  <p><strong>Code</strong>:</p>
  
  <pre><code>private const int ProcessingSize= 4;
  private BufferBlock&lt;string&gt; _fileBufferBlock;
  private ActionBlock&lt;string&gt; _processingBlock;
  private BufferBlock&lt;DataType&gt; _messageBufferBlock;
  
  public Task ProduceAsync()
  {
      PrepareDataflow(token);
      var bufferTask = ListFilesAsync(_fileBufferBlock, token);
  
      var tasks = new List&lt;Task&gt; { bufferTask, _processingBlock.Completion };
      return Task.WhenAll(tasks);
  }
  
  private async Task ListFilesAsync(ITargetBlock&lt;string&gt; targetBlock, CancellationToken token)
  {
      ...
      // Get list of file Uris
      ...
      foreach(var fileNameUri in fileNameUris)
          await targetBlock.SendAsync(fileNameUri, token);
  
      targetBlock.Complete();
  }
  
  private async Task ProcessFileAsync(string fileNameUri, CancellationToken token)
  {
      var httpClient = new HttpClient();
      try
      {
          using (var stream = await httpClient.GetStreamAsync(fileNameUri))
          using (var sr = new StreamReader(stream))
          using (var jsonTextReader = new JsonTextReader(sr))
          {
              while (jsonTextReader.Read())
              {
                  if (jsonTextReader.TokenType == JsonToken.StartObject)
                  {
                      try
                      {
                          var data = _jsonSerializer.Deserialize&lt;DataType&gt;(jsonTextReader)
                          await _messageBufferBlock.SendAsync(data, token);
                      }
                      catch (Exception ex)
                      {
                          _logger.Error(ex, $"JSON deserialization failed - {fileNameUri}");
                      }
                  }
              }
          }
      }
      catch(Exception ex)
      {
          // Should throw?
          // Or if converted to block then report using Fault() method?
      }
      finally
      {
          httpClient.Dispose();
          buffer.Complete();
      }
  }
  
  private void PrepareDataflow(CancellationToken token)
  {
      _fileBufferBlock = new BufferBlock&lt;string&gt;(new DataflowBlockOptions
      {
          CancellationToken = token
      });
  
      var actionExecuteOptions = new ExecutionDataflowBlockOptions
      {
          CancellationToken = token,
          BoundedCapacity = ProcessingSize,
          MaxMessagesPerTask = 1,
          MaxDegreeOfParallelism = ProcessingSize
      };
      _processingBlock = new ActionBlock&lt;string&gt;(async fileName =&gt;
      {
          try
          {
              await ProcessFileAsync(fileName, token);
          }
          catch (Exception ex)
          {
              _logger.Fatal(ex, $"Failed to process fiel: {fileName}, Error: {ex.Message}");
              // Should fault the block?
          }
      }, actionExecuteOptions);
  
      _fileBufferBlock.LinkTo(_processingBlock, new DataflowLinkOptions { PropagateCompletion = true });
  
      _messageBufferBlock = new BufferBlock&lt;DataType&gt;(new ExecutionDataflowBlockOptions
      {
          CancellationToken = token,
          BoundedCapacity = 50000
      });
      _messageBufferBlock.LinkTo(DataflowBlock.NullTarget&lt;DataType&gt;());
  }
  </code></pre>
  
  <p>In the above code, I am not using <code>IEnumerable&lt;DataType&gt;</code> and <code>yield return</code> as I cannot use it with <code>async/await</code>. So I am linking input buffer to <code>ActionBlock&lt;DataType&gt;</code> which in turn posts to another queue. However by using <code>ActionBlock&lt;&gt;</code>, I cannot link it to next block for processing and have to manually <code>Post/SendAsync</code> from <code>ActionBlock&lt;&gt;</code> to <code>BufferBlock&lt;&gt;</code>. Also, in this case, not sure, how to track completion.</p>
  
  <p>This code works, but, I am sure there could be better solution then this and I can just link all the block (instead of <code>ActionBlock&lt;DataType&gt;</code> and then sending messages from it to <code>BufferBlock&lt;DataType&gt;</code>)</p>
  
  <p>Another option could be to convert <code>IEnumerable&lt;&gt;</code> to <code>IObservable&lt;&gt;</code> using <code>Rx</code>, but again I am not much familiar with <code>Rx</code> and don't know exactly how to mix <code>TPL Dataflow</code> and <code>Rx</code></p>