Implement PipeBatchedAsync

[JKamsker]

Please tell me your opinion on this. Do you know any way i could test this?

[electricessence]

I think what you are trying to do is exactly what the current batching does. The BatchingChannelReader works because it creates another channel to use as a buffer: https://github.com/Open-NET-Libraries/Open.ChannelExtensions/blob/master/Open.ChannelExtensions/BufferingChannelReader.cs (base class for

You have a couple options from here. 1) You could write another BatchingChannelReader or modify the existing to allow for setting the bounds and possibly attempt to fill the channel before stopping. 2) Just use the .Pipe extensions to pull as many batches into another channel from that buffer.

But why? What benefit do you gain, when the BatchingChannelReader is as lazy as you can get with batching like this.

I think the easiest and most reliable solution is this:

var batchedReader = sourceChannelReader
    .Batch(100, singleReader: true, allowSynchronousContinuations: true)
    .Pipe(batch=>{ /* transform */ }, capacity: 10);

This will create batches of 100 entries and hold up to 10 batches at a time.

Is that what you were trying to do?

[electricessence]

Need more feedback to proceed.

[JKamsker]

My implementation allows:

• Creating a batch, giving it to the method body and "SelectMany" it to the next channel, it will be a Channel again after the call, not a Channel<IEnumerable>.
• Specify a minimal-size

None of your implementations currently support those usecases. Im not able to extend your code in such a way, maybe you can :)

My usecase:

• Take all files in a folder
• concurrently generate checksums
• Call an api which accept batches with the hashes (PipeBatched)
• Upload each required file simultaneously
• Call another api which also accepts items in bulk

[electricessence]

SelectMany is just a the opposite of a Batch which is a Join. You certainly can do everything your saying.

Here's a way you could do that for your scenario:

await Directory
    .EnumerateFiles(path)
    .ToChannel(1000)
    .Transform(fileName => new FileInfo(fileName))
    .PipeAsync(3, async fileInfo =>
    {
        var contents = await fileInfo.OpenText().ReadToEndAsync();
        var hash = HashFunction(contents);
        return (contents, hash);
    }, capacity: 100, singleReader: false)
    .Batch(20)
    .ReadAllConcurrentlyAsync(3, async batch =>
    {
        // From here it's all custom routing and distribution. **
    });

** Making more extensions to cover every use case doesn't make sense if you can get it done with the existing toolkit.

• You need to have separate channels that handle the final route and configure them uniquely.
• You also need to consider that one fault might cause the whole house cards coming down.
• Your scenario might be complex enough that DataFlow might be the better choice or even a combination of the two.
• You can also call .AsAsyncEnumerable() at any point and switch to using System.Linq.Async if you wanted to get a bit more fancy (since you mentioned SelectMany).

At my work, I recently developed an Async Messaging Toolkit that does have a .Group extension that will distribute to multiple targets, but in the end, you still need to create those targets, and have each target be smart enough to handle any faults.

Hope that helps, lemme know if I missed anything, or if there's something more I can do.