TransformBlock output for output

Question

My scenario is that I have a BufferBlock<Stream> receiving a Stream from an external source, say a file system or some kind of FTP server. This Stream file will go to another block and pass processing.

The only catch is that some of these files are archived, and I would like to add a Block in the middle that would unzip the files if necessary and create some Stream output for each of its records.

However, I do not want to use TransformBlockMany , because this means that I have to completely get the ZIP Stream and immediately create the Stream array.

I would like this Block receive a ZIP Stream , start unpacking and Push to the next stream whenever Entry is ready, so the Process Block can start processing immediately after unpacking the first file, and not wait until everything is unpacked.

How do I get around this?

Answer 1

I realized that my problem is not to use yield / async together. But after refactoring, I got rid of this need and came up with the following (simplified) version:

 var block = new TransformManyBlock<Stream, Stream>((input) => { var archive = new System.IO.Compression.ZipArchive(input, System.IO.Compression.ZipArchiveMode.Read, true); foreach (ZipArchiveEntry entry in archive.Entries) { if (string.IsNullOrWhiteSpace(entry.Name)) //is a folder continue; yield return entry.Open(); } }); 

Answer 2

You can configure an intermediate block for your fragmented logic with a block predicate so that you can check if the stream is an archive or not, for example:

 var buffer = new BufferBlock<Stream>(); var unzipper = new TransformManyBlock<Stream, Stream>(input => { /* unzip here */ }); var processBlock = new ActionBlock<Stream>(input => { /* process streams here */ }); buffer.LinkTo(unzipper, input => /* check is stream a zip archive */); unzipper.LinkTo(processBlock); buffer.LinkTo(processBlock); 

As for using async with yield , you can try the AsyncEnumerable package available on GitHub and NuGet .