AsyncEnumerable
Provides extension methods for IAsyncEnumerable<T> interface.
using DotNext.Buffers;
using DotNext.Patterns;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
namespace DotNext.Collections.Generic
{
[NullableContext(1)]
[Nullable(0)]
public static class AsyncEnumerable
{
[DebuggerDisplay("Count = 0")]
private sealed class EmptyEnumerator<T> : IAsyncEnumerator<T>, IAsyncDisposable, IAsyncEnumerable<T>, ISingleton<EmptyEnumerator<T>>
{
public static EmptyEnumerator<T> Instance { get; } = new EmptyEnumerator<T>();
[DebuggerBrowsable(DebuggerBrowsableState.Never)]
public T Current {
get {
throw new InvalidOperationException();
}
}
private EmptyEnumerator()
{
}
ValueTask<bool> IAsyncEnumerator<T>.MoveNextAsync()
{
return new ValueTask<bool>(false);
}
IAsyncEnumerator<T> IAsyncEnumerable<T>.GetAsyncEnumerator(CancellationToken cancellationToken)
{
return this;
}
ValueTask IAsyncDisposable.DisposeAsync()
{
return ValueTask.CompletedTask;
}
}
private sealed class NotNullEnumerable<T> : IAsyncEnumerable<T> where T : class
{
private sealed class Enumerator : IAsyncEnumerator<T>, IAsyncDisposable
{
private readonly IAsyncEnumerator<T> enumerator;
private T current;
public T Current {
get {
T val = current;
if (val == null)
throw new InvalidOperationException();
return val;
}
}
internal Enumerator(IAsyncEnumerable<T> enumerable, CancellationToken token)
{
enumerator = enumerable.GetAsyncEnumerator(token);
}
[AsyncStateMachine(typeof(NotNullEnumerable<>.Enumerator.<MoveNextAsync>d__5))]
public ValueTask<bool> MoveNextAsync()
{
<MoveNextAsync>d__5 stateMachine = default(<MoveNextAsync>d__5);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<bool>.Create();
stateMachine.<>4__this = this;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
public ValueTask DisposeAsync()
{
current = null;
return enumerator.DisposeAsync();
}
}
private readonly IAsyncEnumerable<T> enumerable;
internal NotNullEnumerable(IAsyncEnumerable<T> enumerable)
{
this.enumerable = enumerable;
}
public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token)
{
return new Enumerator(enumerable, token);
}
}
internal sealed class Proxy<T> : IAsyncEnumerable<T>
{
internal sealed class Enumerator : Disposable, IAsyncEnumerator<T>, IAsyncDisposable
{
private readonly IEnumerator<T> enumerator;
private readonly CancellationToken token;
public T Current => enumerator.Current;
internal Enumerator(IEnumerable<T> enumerable, CancellationToken token)
{
enumerator = enumerable.GetEnumerator();
this.token = token;
}
public ValueTask<bool> MoveNextAsync()
{
if (!token.IsCancellationRequested)
return new ValueTask<bool>(enumerator.MoveNext());
return ValueTask.FromCanceled<bool>(token);
}
protected override void Dispose(bool disposing)
{
if (disposing)
enumerator.Dispose();
base.Dispose(disposing);
}
public new ValueTask DisposeAsync()
{
return base.DisposeAsync();
}
}
private readonly IEnumerable<T> enumerable;
internal Proxy(IEnumerable<T> enumerable)
{
this.enumerable = enumerable;
}
public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token)
{
return new Enumerator(enumerable, token);
}
}
private sealed class ThrowingEnumerator<T> : IAsyncEnumerator<T>, IAsyncDisposable, IAsyncEnumerable<T>
{
[CompilerGenerated]
private Exception <exception>P = exception;
[DebuggerBrowsable(DebuggerBrowsableState.Never)]
public T Current {
get {
throw new InvalidOperationException();
}
}
public ThrowingEnumerator(Exception exception)
{
}
ValueTask<bool> IAsyncEnumerator<T>.MoveNextAsync()
{
return ValueTask.FromException<bool>(<exception>P);
}
IAsyncEnumerator<T> IAsyncEnumerable<T>.GetAsyncEnumerator(CancellationToken cancellationToken)
{
return this;
}
ValueTask IAsyncDisposable.DisposeAsync()
{
return ValueTask.CompletedTask;
}
}
[AsyncStateMachine(typeof(<CopyAsync>d__0<>))]
[return: Nullable(new byte[] {
1,
0,
1
})]
public static Task<MemoryOwner<T>> CopyAsync<[Nullable(2)] T>(this IAsyncEnumerable<T> enumerable, int sizeHint = 0, [Nullable(new byte[] {
2,
1
})] MemoryAllocator<T> allocator = null, CancellationToken token = default(CancellationToken))
{
<CopyAsync>d__0<T> stateMachine = default(<CopyAsync>d__0<T>);
stateMachine.<>t__builder = AsyncTaskMethodBuilder<MemoryOwner<T>>.Create();
stateMachine.enumerable = enumerable;
stateMachine.sizeHint = sizeHint;
stateMachine.allocator = allocator;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<ForEachAsync>d__1<>))]
public static ValueTask ForEachAsync<[Nullable(2)] T>(this IAsyncEnumerable<T> collection, Action<T> action, CancellationToken token = default(CancellationToken))
{
<ForEachAsync>d__1<T> stateMachine = default(<ForEachAsync>d__1<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder.Create();
stateMachine.collection = collection;
stateMachine.action = action;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<ForEachAsync>d__2<>))]
public static ValueTask ForEachAsync<[Nullable(2)] T>(this IAsyncEnumerable<T> collection, Func<T, CancellationToken, ValueTask> action, CancellationToken token = default(CancellationToken))
{
<ForEachAsync>d__2<T> stateMachine = default(<ForEachAsync>d__2<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder.Create();
stateMachine.collection = collection;
stateMachine.action = action;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[NullableContext(0)]
[AsyncStateMachine(typeof(<FirstOrNullAsync>d__3<>))]
public static ValueTask<T?> FirstOrNullAsync<T>([Nullable(new byte[] {
1,
0
})] this IAsyncEnumerable<T> seq, CancellationToken token = default(CancellationToken)) where T : struct
{
<FirstOrNullAsync>d__3<T> stateMachine = default(<FirstOrNullAsync>d__3<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<T?>.Create();
stateMachine.seq = seq;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[NullableContext(0)]
[AsyncStateMachine(typeof(<LastOrNullAsync>d__4<>))]
public static ValueTask<T?> LastOrNullAsync<T>([Nullable(new byte[] {
1,
0
})] this IAsyncEnumerable<T> seq, CancellationToken token = default(CancellationToken)) where T : struct
{
<LastOrNullAsync>d__4<T> stateMachine = default(<LastOrNullAsync>d__4<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<T?>.Create();
stateMachine.seq = seq;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<FirstOrNoneAsync>d__5<>))]
[return: Nullable(new byte[] {
0,
0,
1
})]
public static ValueTask<Optional<T>> FirstOrNoneAsync<[Nullable(2)] T>(this IAsyncEnumerable<T> seq, CancellationToken token = default(CancellationToken))
{
<FirstOrNoneAsync>d__5<T> stateMachine = default(<FirstOrNoneAsync>d__5<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<Optional<T>>.Create();
stateMachine.seq = seq;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<LastOrNoneAsync>d__6<>))]
[return: Nullable(new byte[] {
0,
0,
1
})]
public static ValueTask<Optional<T>> LastOrNoneAsync<[Nullable(2)] T>(this IAsyncEnumerable<T> seq, CancellationToken token = default(CancellationToken))
{
<LastOrNoneAsync>d__6<T> stateMachine = default(<LastOrNoneAsync>d__6<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<Optional<T>>.Create();
stateMachine.seq = seq;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<FirstOrNoneAsync>d__7<>))]
[return: Nullable(new byte[] {
0,
0,
1
})]
public static ValueTask<Optional<T>> FirstOrNoneAsync<[Nullable(2)] T>(this IAsyncEnumerable<T> seq, Predicate<T> filter, CancellationToken token = default(CancellationToken))
{
<FirstOrNoneAsync>d__7<T> stateMachine = default(<FirstOrNoneAsync>d__7<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<Optional<T>>.Create();
stateMachine.seq = seq;
stateMachine.filter = filter;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[NullableContext(0)]
[AsyncStateMachine(typeof(<SkipAsync>d__8<>))]
public static ValueTask<bool> SkipAsync<[Nullable(2)] T>([Nullable(1)] this IAsyncEnumerator<T> enumerator, int count)
{
<SkipAsync>d__8<T> stateMachine = default(<SkipAsync>d__8<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<bool>.Create();
stateMachine.enumerator = enumerator;
stateMachine.count = count;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
[AsyncStateMachine(typeof(<ElementAtAsync>d__9<>))]
[return: Nullable(new byte[] {
0,
0,
1
})]
public static ValueTask<Optional<T>> ElementAtAsync<[Nullable(2)] T>(this IAsyncEnumerable<T> collection, int index, CancellationToken token = default(CancellationToken))
{
<ElementAtAsync>d__9<T> stateMachine = default(<ElementAtAsync>d__9<T>);
stateMachine.<>t__builder = AsyncValueTaskMethodBuilder<Optional<T>>.Create();
stateMachine.collection = collection;
stateMachine.index = index;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
public static IAsyncEnumerable<T> SkipNulls<T>([Nullable(new byte[] {
1,
2
})] this IAsyncEnumerable<T> collection) where T : class
{
return new NotNullEnumerable<T>(collection);
}
[AsyncStateMachine(typeof(<ToArrayAsync>d__11<>))]
public static Task<T[]> ToArrayAsync<[Nullable(2)] T>(this IAsyncEnumerable<T> collection, int initialCapacity = 10, [Nullable(new byte[] {
2,
1
})] MemoryAllocator<T> allocator = null, CancellationToken token = default(CancellationToken))
{
<ToArrayAsync>d__11<T> stateMachine = default(<ToArrayAsync>d__11<T>);
stateMachine.<>t__builder = AsyncTaskMethodBuilder<T[]>.Create();
stateMachine.collection = collection;
stateMachine.initialCapacity = initialCapacity;
stateMachine.allocator = allocator;
stateMachine.token = token;
stateMachine.<>1__state = -1;
stateMachine.<>t__builder.Start(ref stateMachine);
return stateMachine.<>t__builder.Task;
}
public static IAsyncEnumerable<T> Empty<[Nullable(2)] T>()
{
return EmptyEnumerator<T>.Instance;
}
public static IAsyncEnumerable<T> Throw<[Nullable(2)] T>(Exception e)
{
ArgumentNullException.ThrowIfNull((object)e, "e");
return new ThrowingEnumerator<T>(e);
}
}
}