MockFileSystem.File.Move raises FileSystemWatcher event with delay #2

[teneko]

Continuation of #930 (I am not the author)

I do not get a reliable way to get the hand on the aggregated change events similiar to author of the above issue.

I also think to block the thread as the method doc indicates is a bit harsh.

I use

using var watcher = fileSystem.FileSystemWatcher.New("/");
watcher.IncludeSubdirectories = true;
watcher.EnableRaisingEvents = true;

const string fileName = "test.txt"

List<WatcherChangeTypes?> changes = [];

using var changeWatcher = fileSystem.Watcher.OnTriggered(x => {
    if (x.Path.EndsWith(fileName)) {
        changes.Add(x.ChangeType);
    }
});

..operate on file system..

const int timeoutInSeconds = 5 * 1000;
changeWatcher.Wait(timeout: timeoutInSeconds);

// Assumptions
changes.FirstOrDefault().ShouldBe(WatcherChangeTypes.Created);
changes.LastOrDefault().ShouldBe(WatcherChangeTypes.Deleted);

Whatever timeout I choose, the wait timeout is exhausted no matter what leading to an timeout exception. Only if I set a breakpoint between "..operate on file system.." and "changeWatcher.Wait" it works as inteded. Pretty sure the notifications are not buffered / not replayed!?

Can we please enable the following workflow:

using var watcher = fileSystem.FileSystemWatcher.New("/");
watcher.IncludeSubdirectories = true;
watcher.EnableRaisingEvents = true;

const string fileName = "test.txt"

List<WatcherChangeTypes?> changes = [];

using var changeWatcher = fileSystem.Watcher.OnTriggered(x => {
    if (x.Path.EndsWith(fileName)) {
        changes.Add(x.ChangeType);
    }
});

const int timeoutInSeconds = 5 * 1000;
// count = 3 (Created -> Changed -> Deleted)
var changeWatchedTask = changeWatcher.WaitAsync(timeout: timeoutInSeconds, count: 3);

..operate on file system..

// Assumptions
await changeWatchedTask; // if timeout exhausts throw as usual
changes.FirstOrDefault().ShouldBe(WatcherChangeTypes.Created);
changes.LastOrDefault().ShouldBe(WatcherChangeTypes.Deleted);

Here the corresponding extension method making above drawn workflow work:

file static class AwaitableChangeDescriptionCallbackExtensions
{
    extension<TValue>(IAwaitableCallback<TValue> awaitableCallback)
    {
        public Task WaitAsync(
            Func<TValue, bool>? filter = null,
            int timeout = 30000,
            int count = 1,
            Action? executeWhenWaiting = null,
            CancellationToken cancellationToken = default)
        {
            return Task.Factory.StartNew(
                static state => {
                    var (awaitableCallback, filter, timeout, count, executeWhenWaiting, cancellationToken) =
                        (ValueTuple<IAwaitableCallback<TValue>, Func<TValue, bool>?, int, int, Action?, CancellationToken>)state!;
                    
                    cancellationToken.ThrowIfCancellationRequested();

                    awaitableCallback.Wait(
                        filter,
                        timeout: timeout,
                        count: count,
                        executeWhenWaiting is not null ? ExecuteWhenWaiting : null);

                    return;

                    void ExecuteWhenWaiting()
                    {
                        cancellationToken.ThrowIfCancellationRequested();
                        executeWhenWaiting();
                    }
                },
                (awaitableCallback, filter, timeout, count, executeWhenWaiting, cancellationToken),
                cancellationToken,
                TaskCreationOptions.DenyChildAttach,
                TaskScheduler.Default);
        }
    }
}

[vbreuss]

With the current API you have to execute your operations on the file system during the callback:

using var changeWatcher = fileSystem.Watcher.OnTriggered(x => {
    if (x.Path.EndsWith(fileName)) {
        changes.Add(x.ChangeType);
    },
    executeWhenWaiting: () => 
    {
        // ..operate on file system..
    }
});

I agree that this is probably not the most intuitive syntax and will have to think about possible alternatives...

[teneko]

It would make much more sense to define assumptions like in the workaround I proposed. I mean building promises that fulfill once file system conditions have been met. I mean you currently can do that now, it is currently just glue code/boiletplate code you need to write manually.

I think there is not much you can do. You have a file system that fires events.. that gives us the way to observe the file system changes. that's great. before we change the file system, we make assumptions/define expectations. I also could use TaskCompletionSource in triggerCallback of ChangeHandler.OnTriggered and SetResult(..) once the expectation has met, or SetCanceled after x seconds (e.g. via cancellationToken.CancelAfter) or use ((TaskCompletionSource)tcs).Task.WithCancellation. So many ways.

[vbreuss]

My current thought is, that if we change this, I would like to add some more functionality (like getting the actual values of the callbacks). Something along the line to add the following method to the IAwaitableCallback signature:

/// <summary>
///     Waits asynchronously until the callback is executed.
/// </summary>
/// <param name="filter">
///     (optional) A filter for the callback to count.<br />
///     Defaults to <see langword="null" /> which will consider all callbacks.
/// </param>
/// <param name="count">
///     (optional) The number of callbacks to wait.<br />
///     Defaults to <c>1</c>
/// </param>
/// <param name="cancellationToken">
///     (optional) A <see cref="CancellationToken" /> to cancel waiting.<br />
///     Throws a <see cref="OperationCanceledException" /> if the token was canceled before the callback was executed.
/// </param>
/// <returns></returns>
Task<TValue[]> WaitAsync(Func<TValue, bool>? filter = null, int count = 1, CancellationToken? cancellationToken = null);

I would then start collecting the callbacks directly when IAwaitableCallback is created and this would allow the following workflow:

using var watcher = fileSystem.FileSystemWatcher.New("/");
watcher.IncludeSubdirectories = true;
watcher.EnableRaisingEvents = true;

const string fileName = "test.txt"

List<WatcherChangeTypes?> changes = [];

using var changeWatcher = fileSystem.Watcher.OnTriggered(x => {
    if (x.Path.EndsWith(fileName)) {
        changes.Add(x.ChangeType);
    }
});

..operate on file system..

const int timeoutInSeconds = 5 * 1000;
// count = 3 (Created -> Changed -> Deleted)
// Assumptions
await changeWatcher.WaitAsync(timeout: timeoutInSeconds, count: 3); // if timeout exhausts throw as usual
changes.FirstOrDefault().ShouldBe(WatcherChangeTypes.Created);
changes.LastOrDefault().ShouldBe(WatcherChangeTypes.Deleted);

In a synchronous scenario you could use the same workflow but call .Wait and it would block (we would no longer need the ExecuteWhileWaiting callbacks and could remove them in a future update).

What are your thoughts, @teneko?

[teneko]

I am no sure I did understand you correctly.

1. You want to filter (using filter argument /callback) on type of TValue (e.g. ChangeDescription), and those that match, i.e. returning true, to collect and return?

Then I ask you, where does count contribute to? To the next x events (regardless of filter), or the next x filter match?

2. You must put var changeDescriptions = await changeWatcher.WaitAsync(timeout: timeoutInSeconds, count: 3); before "..operate on file system..", otherwise you miss events as they may already have happened.

3. What about aggregating? In my case I have a super isolated desire to just know that a file has been created, changed and deleted, but what if you want to watch two files, where you expect the first file to be deleted and then the second? Scenarios can get complex.

I think the current design is quite confusing. Why I do have OnTriggered supporting filter/predicate AND an awaitable but also filterable Wait/WaitAsync?

In my head I have the following mental model:

you have a file system watcher, you can listen to its events, once EnableRaisingEvents = true. Because it is inconvenient to register events for each kind, you have introduced OnTriggered, that registers events, and unregisters them on disposal.

I would expect that OnTriggered is NOT awaitable, so no IAwaitableCallback with custom GetAwaiter pattern, but just I[Async]Disposable, instead I would expect that Wait/WaitAsync moves to the Watcher-layer, so you call fileSystem.Watcher.[METHOD], for instance:

• filesystem.Watcher.AggregateWhile (when you imagine the incoming events as a stream (IEnumerable, IAsyncEnumerable, IObservable), see example)

[teneko]

I have something like the following in mind:

Workflow:

const int timeoutInSeconds = 5 * 1000;
using var changeWatchTask = fileSystem.Watcher.AggregateWhile(
    new List<WatcherChangeTypes?>(),
    (seed, x) => {
        if (x.Path.EndsWith(fileName)) {
            seed.Add(x.ChangeType);
        }
        return seed;
    },
    // count = 3 (Created -> Changed -> Deleted)
    seed => seed.Count < 3,
    timeoutInSeconds,
    TestContext.Current.CancellationToken);

..operate on file system..

var changes = await changeWatchTask; // if timeout exhausts//cancellation kicks in, then throw as usual
changes.FirstOrDefault().ShouldBe(WatcherChangeTypes.Created);
changes.LastOrDefault().ShouldBe(WatcherChangeTypes.Deleted);

AggregateWhile:

public interface IDisposableAggregation<TValue> : IDisposable
{
    TaskAwaiter<TValue> GetAwaiter();
}

public static IDisposableAggregation<TValue> AggregateWhile<TValue>(
    this IWatcherTriggeredHandler watcher,
    TValue seed,
    Func<TValue, ChangeDescription, TValue> aggregateFunc,
    Func<TValue, bool>? whileFunc,
    int? timeout,
    CancellationToken cancellationToken)
{
    ArgumentNullException.ThrowIfNull(watcher);
    ArgumentNullException.ThrowIfNull(aggregateFunc);
    ArgumentNullException.ThrowIfNull(whileFunc);

    var aggregationCompletionSource = new TaskCompletionSource<TValue>(TaskCreationOptions.RunContinuationsAsynchronously);
    var aggregationTask = aggregationCompletionSource.Task;

    Channel<ChangeDescription>? triggerChannel = null;
    CancellationTokenSource? combinedCancellationTokenSource = null;
    IAwaitableCallback<ChangeDescription>? triggerSubscription = null;
    try {
        triggerChannel = Channel.CreateUnbounded<ChangeDescription>(
            new UnboundedChannelOptions {
                SingleReader = true,
                SingleWriter = true,
                AllowSynchronousContinuations = false
            });

        combinedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

        if (timeout.HasValue) {
            combinedCancellationTokenSource.CancelAfter(timeout.Value);
        }

        var combinedCancellationToken = combinedCancellationTokenSource.Token;
        _ = combinedCancellationToken.Register(WhenCancelThenComplete);

        triggerSubscription = watcher.OnTriggered(WhenTriggerThenWrite);

        var aggregation = new DisposableAggregation<TValue>(
            aggregationTask,
            combinedCancellationTokenSource,
            triggerChannel.Writer,
            triggerSubscription);

        _ = AggregateAsync(); // Start aggregation immediately
        return aggregation;

        void WhenCancelThenComplete()
        {
            triggerChannel.Writer.TryComplete();
        }

        async void WhenTriggerThenWrite(ChangeDescription changeDescription)
        {
            try {
                await triggerChannel.Writer.WriteAsync(changeDescription, combinedCancellationToken).ConfigureAwait(false);
            } catch (OperationCanceledException) {
                // Do nothing
            } catch (ChannelClosedException) {
                // Do nothing
            } catch (Exception error) {
                aggregationCompletionSource.TrySetException(error);
                triggerChannel.Writer.TryComplete(error);
            }
        }

        async Task AggregateAsync()
        {
            var currentValue = seed;

            try {
                if (combinedCancellationToken.IsCancellationRequested) {
                    aggregationCompletionSource.TrySetCanceled(combinedCancellationToken);
                    triggerChannel.Writer.TryComplete();
                    return;
                }

                await foreach (var change in triggerChannel.Reader.ReadAllAsync(combinedCancellationToken).ConfigureAwait(false)) {
                    currentValue = aggregateFunc(currentValue, change);

                    if (!whileFunc(currentValue)) {
                        triggerChannel.Writer.TryComplete();
                        break;
                    }
                }

                if (combinedCancellationToken.IsCancellationRequested) {
                    aggregationCompletionSource.TrySetCanceled(combinedCancellationToken);
                } else {
                    aggregationCompletionSource.TrySetResult(currentValue);
                }
            } catch (OperationCanceledException) {
                aggregationCompletionSource.TrySetCanceled(combinedCancellationToken);
            } catch (Exception error) {
                aggregationCompletionSource.TrySetException(error);
                triggerChannel.Writer.TryComplete(error);
            }
        }
    } catch (Exception error) {
        triggerSubscription?.Dispose();
        combinedCancellationTokenSource?.Cancel();
        triggerChannel?.Writer.TryComplete();
        combinedCancellationTokenSource?.Dispose();
        aggregationCompletionSource.TrySetException(error);
        return new CompletedAggregation<TValue>(aggregationTask);
    }
}

private sealed class CompletedAggregation<TValue> : IDisposableAggregation<TValue>
{
    private readonly Task<TValue> _aggregationTask;

    internal CompletedAggregation(Task<TValue> aggregationTask)
    {
        _aggregationTask = aggregationTask;
    }

    public TaskAwaiter<TValue> GetAwaiter() => _aggregationTask.GetAwaiter();

    public void Dispose() { }
}

private sealed class DisposableAggregation<TValue> : IDisposableAggregation<TValue>
{
    private readonly CancellationTokenSource _cancellationTokenSource;
    private readonly ChannelWriter<ChangeDescription> _writer;
    private readonly IDisposable _subscription;
    private readonly Task<TValue> _aggregationTask;
    private int _disposed;

    internal DisposableAggregation(
        Task<TValue> aggregationTask,
        CancellationTokenSource cancellationTokenSource,
        ChannelWriter<ChangeDescription> writer,
        IDisposable subscription)
    {
        _cancellationTokenSource = cancellationTokenSource;
        _writer = writer;
        _subscription = subscription;
        _aggregationTask = aggregationTask;
    }

    public TaskAwaiter<TValue> GetAwaiter()
    {
        return _aggregationTask.GetAwaiter();
    }

    public void Dispose()
    {
        if (Interlocked.Exchange(ref _disposed, 1) != 0) {
            return;
        }

        _subscription.Dispose();
        _cancellationTokenSource.Cancel();
        _ = _writer.TryComplete();
        _cancellationTokenSource.Dispose();
    }
}

[vbreuss]

I created a draft-PR to illustrate what I would implement:

• feat: support async wait handlers #942

1. Then I ask you, where does count contribute to? To the next x events (regardless of filter), or the next x filter match?

In it I added two new (sync and async) Wait/WaitAsync methods without the duplicate filtering, so this should remove the confusion about where the count contributes to.

2. You must put var changeDescriptions = await changeWatcher.WaitAsync(timeout: timeoutInSeconds, count: 3); before "..operate on file system..", otherwise you miss events as they may already have happened.

The main difference after my PR is, that observing the file system starts immediately when you register the wait callback. So you no longer need to use the ExecuteWhileWaiting callbacks, but can simply store the IAwaitableCallback<T> in a variable, then operate on the filesystem and then call Wait or WaitAsync.

3. What about aggregating? In my case I have a super isolated desire to just know that a file has been created, changed and deleted, but what if you want to watch two files, where you expect the first file to be deleted and then the second? Scenarios can get complex.

The two new methods return an array of observed ChangeDescription on the mock file system, so you are able to then filter or work with them in any way you need.

Would this solve your issue, @teneko?

If I go forward with this, I would mark the previous methods as obsolete and would remove them after a transition period in a future major version update.

[teneko]

Can you give me a short motivation into why you would solve it like that?

Similiar I have written a motivation towards AggregateWhile?

What about aggregating? In my case I have a super isolated desire to just know that a file has been created, changed and deleted, but what if you want to watch two files, where you expect the first file to be deleted and then the second? Scenarios can get complex.

the following I assume that count contributes the count when predicate of OnTriggered is resulting into true.

Flaws/misconceptions I can identify:

1. In my opinion you falsely had the tend to introduce WaitAsync with an predicate/count to get to an definite result because OnTriggered is designed as watcher watching on changes applying to the OnTriggered predicate without definite end.
2. count still has an opaque and abstract meaning.. When OnTriggered collect changes (after truthy predicate), then count will just count since start of that collection. So just imagine, you use count on Wait/WaitAsyc, that you use AFTER file changes, where count contributes the collection of changes of OnTriggered.

You want to test file system files/directories, but code using IFIleSystem, can do a lot:

• it can one file either create -> delete once, twice, or more.
• it can one file either create, change once, twice or more, and then delete
• it can one, two or more files either create -> delete once, twice, or more.
• it can one, two or more files either create, change once, twice or more, and then delete
• any operation can mix
• and so much more...

With current means you cannot simply define a simple or potencially aggregating state, nor can you simply define an awaitable definite state.

[teneko]

To be honest it would be sufficient and cover all scenarios if you could just stream your changes via IAsyncEnumerable. Then someone can use Where/Select/Aggregate/Take on its own.

[vbreuss]

Can you give me a short motivation into why you would solve it like that?

I want to keep the interface as simple as possible. What I try to achieve is

• remove the ambiguity of having two separate filters
• avoid having to move the operations on the file system to a callback
  while keeping as close as possible to the existing implementation to reduce the friction/breaking change.

After this change the Wait/WaitAsync methods simply block (or wait asynchronously) until the events in the filesystem that match the filter occurred count times and return the corresponding values. If needed you can wait multiple times on the same object (until you dispose it), so it is much easier to create custom extensions that aggregate some result by waiting multiple times, e.g.:

	public static async Task<TResult> AggregateWhile<TValue, TResult>(this IAwaitableCallback<TValue> source, TResult seed, Func<TResult, TValue, TResult> aggregateFunc, Func<TResult, bool> whileFunc)
	{
		do
		{
			TValue[] value = await source.WaitAsync();
			if (value.Length > 0)
			{
				seed = aggregateFunc(seed, value[0]);
			}
		} while (whileFunc(seed));

		return seed;
	}

To be honest it would be sufficient and cover all scenarios if you could just stream your changes via IAsyncEnumerable. Then someone can use Where/Select/Aggregate/Take on its own.

Also this would be quite simple to implement, I think:

	public static async IAsyncEnumerable<TValue> EnumerateAsync<TValue>(this IAwaitableCallback<TValue> source, [EnumeratorCancellation] CancellationToken cancellationToken = default)
	{
		while (!cancellationToken.IsCancellationRequested)
		{
			var item = await source.WaitAsync(cancellationToken:cancellationToken);
			if (item.Length > 0)
			{
				yield return item[0];
			}
		}
	}

But in most use cases that I know of, I only need to wait for one event to occur, so I want to keep this use case as simple as possible.

[teneko]

Take that please to consideration.. simple, effective and very very open to any kind of scenario..

public static IAsyncEnumerable<ChangeDescription> ToAsyncEnumerable(this IWatcherTriggeredHandler watcher)
{
    ArgumentNullException.ThrowIfNull(watcher);

    return WatchAsyncImpl();

    async IAsyncEnumerable<ChangeDescription> WatchAsyncImpl([EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();

        Channel<ChangeDescription>? triggerChannel = null;
        CancellationTokenRegistration? cancellationTokenRegistration = null;
        IAwaitableCallback<ChangeDescription>? triggerSubscription = null;
        try {
            triggerChannel = Channel.CreateUnbounded<ChangeDescription>(
                new UnboundedChannelOptions {
                    SingleReader = true,
                    SingleWriter = true,
                    AllowSynchronousContinuations = false
                });

            cancellationTokenRegistration = cancellationToken.Register(WhenCancelThenComplete);

            triggerSubscription = watcher.OnTriggered(WhenTriggerThenWrite);

            await foreach (var change in triggerChannel.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) {
                yield return change;
            }

            void WhenCancelThenComplete()
            {
                triggerChannel.Writer.TryComplete();
            }

            async void WhenTriggerThenWrite(ChangeDescription changeDescription)
            {
                try {
                    await triggerChannel.Writer.WriteAsync(changeDescription, cancellationToken).ConfigureAwait(false);
                } catch (OperationCanceledException) {
                    // Do nothing
                } catch (ChannelClosedException) {
                    // Do nothing
                } catch (Exception error) {
                    triggerChannel.Writer.TryComplete(error);
                }
            }
        } finally {
            triggerSubscription?.Dispose();
            triggerChannel?.Writer.TryComplete();

            if (cancellationTokenRegistration is not null) {
                await cancellationTokenRegistration.Value.DisposeAsync();
            }
        }
    }
}

public static IAsyncEnumerable<ChangeDescription> WithTimeout(this IAsyncEnumerable<ChangeDescription> source, int timeout)
{
    ArgumentOutOfRangeException.ThrowIfNegative(timeout);

    return AsyncEnumerableImpl();

    async IAsyncEnumerable<ChangeDescription> AsyncEnumerableImpl(
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();

        CancellationTokenSource? combinedCancellationTokenSource = null;
        try {
            combinedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            combinedCancellationTokenSource.CancelAfter(timeout);

            await foreach (var sourceItem in source.WithCancellation(cancellationToken)) {
                yield return sourceItem;
            }
        } finally {
            combinedCancellationTokenSource?.Dispose();
        }
    }
}

public static async Task<TAccumulate> AggregateWhileAsync<TSource, TAccumulate>(
    this IAsyncEnumerable<TSource> source,
    TAccumulate seed,
    Func<TAccumulate, TSource, TAccumulate> accumulator,
    Func<TAccumulate, bool> whileCondition,
    CancellationToken cancellationToken)
{
    ArgumentNullException.ThrowIfNull(accumulator);
    ArgumentNullException.ThrowIfNull(whileCondition);

    var accumulatedValue = seed;

    await foreach (var currentItem in source.WithCancellation(cancellationToken)) {
        accumulatedValue = accumulator(accumulatedValue, currentItem);

        if (!whileCondition(accumulatedValue)) {
            break;
        }
    }

    return accumulatedValue;
}

[teneko]

I think you really over complicate this.. it is just as I said

You have a file system that fires events.. that gives us the way to observe the file system changes.

you have a file system watcher, you can listen to its events, once EnableRaisingEvents = true. Because it is inconvenient to register events for each kind, you have introduced OnTriggered, that registers events, and unregisters them on disposal.

[teneko]

So at the end you end up with the following workflow

const int timeoutInSeconds = 5 * 1000;
var changeWatchTask = fileSystem.Watcher.ToAsyncEnumerable().WithTimeout(timeoutInSeconds).AggregateWhile(
    new List<WatcherChangeTypes?>(),
    (seed, x) => {
        if (x.Path.EndsWith(fileName)) {
            seed.Add(x.ChangeType);
        }
        return seed;
    },
    // count = 3 (Created -> Changed -> Deleted)
    seed => seed.Count < 3,
    TestContext.Current.CancellationToken);

..operate on file system..

var changes = await changeWatchTask; // if timeout exhausts//cancellation kicks in, then throw as usual
changes.FirstOrDefault().ShouldBe(WatcherChangeTypes.Created);
changes.LastOrDefault().ShouldBe(WatcherChangeTypes.Deleted);

and you can be as creative as you can be at how you define the definite state / while condition.

Important

Keep in mind that as my ToAsyncEnumerable [-> WithTimeout [-> AggregateWhile]] chain is not disposable (not returning IDisposableAggregation) thus OnTriggered subscription will only be disposed if a) successful b) timeout has been specified (over WithTimeout) or c) cancellation (over WithCancellation or passing cancellation token to AggregateWhile) has been specified. Hint: normally you would consume ToAsyncEnumerable() as follow: await using var changeDescriptionEnumerable = watcher.ToAsyncEnumerable().GetAsyncEnumerator() or implictly by await foreach (var changeDescription =watcher.ToAsyncEnumerable() ) { .. } In these cases the enumerator would explictly be disposed including the trigger subscription.

[vbreuss]

I rewrote your workflow, how it would work with my system:

	const int timeoutInSeconds = 5 * 1000;
	var changeWatchTask =
		fileSystem.Watcher.OnTriggered(predicate: x => x.Path.EndsWith(fileName));

	// ..operate on file system..
	
	var changes = await changeWatchTask.WaitAsync(3, timeout: TimeSpan.FromSeconds(timeoutInSeconds), TestContext.Current.CancellationToken);
    changes.FirstOrDefault().ChangeType.ShouldBe(WatcherChangeTypes.Created);
    changes.LastOrDefault().ChangeType.ShouldBe(WatcherChangeTypes.Deleted);

Please note, that IAsyncEnumerable is not supported on .NET Standard 2.0, so an extension like ToAsyncEnumerable() could only be provided as extension method on newer frameworks.

[teneko]

I do not understand why you would narrow possible scenarios with such an opinionated implementation. Please reconsider.

IAsyncEnumerable<> is available for .netstandard2.0, see https://www.nuget.org/packages/Microsoft.Bcl.AsyncInterfaces/11.0.0-preview.1.26104.118#dependencies-body-tab

IAsyncEnumerable<> LINQ extension point is available for .netstandard2.0, see https://www.nuget.org/packages/System.Linq.Async#dependencies-body-tab

Edit: Your proposed solution may just fit into this exact scenario I test for but this is a super super simple case.. but it is in simple terms not scalable.