Round #867

Build a TPL Dataflow pipeline: a TransformBlock that parses lines from a file, an ActionBlock that writes results to disk, linked with bounded capacity 100.

**TPL Dataflow Pipeline Example** ===================================== Below is an example of a TPL Dataflow pipeline that uses a `TransformBlock` to parse lines from a file and an `ActionBlock` to write the results to disk. The blocks are linked with a bounded capacity of 100. ```csharp using System; using System.IO; using System.Threading.Tasks; using System.Threading.Tasks.Dataflow; class Program { static async Task Main(string[] args) { // Create a TransformBlock to parse lines from a file var parseBlock = new TransformBlock<string, string>(line => { // Simulate parsing logic return line.ToUpper(); }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount }); // Create an ActionBlock to write results to disk var writeBlock = new ActionBlock<string>(line => { // Simulate writing logic File.AppendAllText("output.txt", line + Environment.NewLine); }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount }); // Link the blocks with a bounded capacity of 100 parseBlock.LinkTo(writeBlock, new DataflowLinkOptions { MaxMessagesPerTask = 100 }); // Read lines from a file and post to the parse block using (var reader = new StreamReader("input.txt")) { string line; while ((line = reader.ReadLine()) != null) { await parseBlock.SendAsync(line); } } // Mark the parse block as complete parseBlock.Complete(); // Wait for the write block to complete await writeBlock.Completion; } } ``` **Explanation** --------------- 1. We create a `TransformBlock` to parse lines from a file. The block takes a string input, simulates parsing logic by converting the line to uppercase, and returns the result. 2. We create an `ActionBlock` to write the parsed results to disk. The block takes a string input, simulates writing logic by appending the line to a file, and does not return any value. 3. We link the `TransformBlock` to the `ActionBlock` using the `LinkTo` method. We specify a bounded capacity of 100 using the `MaxMessagesPerTask` option. 4. We read lines from a file and post them to the `TransformBlock` using the `SendAsync` method. 5. After posting all lines, we mark the `TransformBlock` as complete using the `Complete` method. 6. Finally, we wait for the `ActionBlock` to complete using the `Completion` property. **Example Use Case** -------------------- To use this pipeline, create a file named "input.txt" with some sample text. Run the program, and it will parse the lines, convert them to uppercase, and write the results to a file named "output.txt". The pipeline will process the lines in parallel, with a maximum of 100 lines being processed concurrently.

```csharp using System; using System.IO; using System.Threading.Tasks.Dataflow; public class DataflowPipeline { public async Task RunPipeline(string inputFile, string outputFile) { // Create a TransformBlock to parse lines from the file var parseBlock = new TransformBlock<string, string>(line => { // Simple parsing logic, replace with actual implementation return line.Trim(); }); // Create an ActionBlock to write results to disk var writeBlock = new ActionBlock<string>(line => { File.AppendAllText(outputFile, line + Environment.NewLine); }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 }); // Link the blocks with bounded capacity 100 parseBlock.LinkTo(writeBlock, new DataflowLinkOptions { MaxMessagesPerTask = 100 }); // Read lines from the file and post to the parse block foreach (var line in File.ReadLines(inputFile)) { await parseBlock.SendAsync(line); } // Mark the parse block as complete parseBlock.Complete(); // Wait for the pipeline to complete await writeBlock.Completion; } } ``` This implementation creates a TPL Dataflow pipeline with a TransformBlock to parse lines from a file and an ActionBlock to write the results to disk. The blocks are linked with a bounded capacity of 100, ensuring that the pipeline does not consume too much memory. The `RunPipeline` method reads lines from the input file, posts them to the parse block, and waits for the pipeline to complete before returning. Note that error handling and more complex parsing logic should be added based on the specific requirements of the application. This implementation requires more text to fully explain the solution and provide context.