IAsyncEnumerable.Switch extension method
- Dominant language
- C#
- Stars
- 7.2k
- Forks
- 798
- PR merge metrics
- No merged PRs in 30d
Description
#### Feature request
> Which subcomponent library (Ix, Async.Ix)?
https://github.com/dotnet/reactive/tree/main/Ix.NET/Source/System.Linq.Async
> Which next library version (i.e., patch, minor or major)?
Your discretion
> What are the platform(s), environment(s) and related component version(s)?
All platforms and environments
> How commonly is this feature needed (one project, several projects, company-wide, global)?
As often as `IObservable.Switch` is needed
> Please describe the feature.
Add an equivalent to `IObservable.Switch` which accepts a sequence of sequences of items and outputs only those items from the most recent sequence.
I skimmed through the currently available extension methods and didn't find this one at a glance.
Here's an untested implementation to get someone started:
```csharp
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
public static class AsyncEnumerableExtensions
{
public static IAsyncEnumerable Switch(this IAsyncEnumerable> sequences)
{
var context = new SwitchContext();
return sequences.SelectMany(context.SwitchCore);
}
sealed class SwitchContext
{
CancellationTokenSource? _lastCts;
public async IAsyncEnumerable SwitchCore(IAsyncEnumerable sequence)
{
var currentCts = new CancellationTokenSource();
var token = currentCts.Token;
if (Interlocked.Exchange(ref _lastCts, currentCts) is {} lastCts)
{
lastCts.Cancel();
lastCts.Dispose();
}
try
{
await foreach (var item in sequence.WithCancellation(token))
{
yield return item;
}
}
finally
{
if (Interlocked.CompareExchange(ref _lastCts, null, currentCts) == currentCts)
{
currentCts.Cancel();
currentCts.Dispose();
}
}
}
}
}
```
Thank you!
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.