using Cysharp.Threading.Tasks.Internal; using System; using System.Threading; namespace Cysharp.Threading.Tasks.Linq { public static partial class UniTaskAsyncEnumerable { public static IUniTaskAsyncEnumerable<TSource> Where<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, Boolean> predicate) { Error.ThrowArgumentNullException(source, nameof(source)); Error.ThrowArgumentNullException(predicate, nameof(predicate)); return new Where<TSource>(source, predicate); } public static IUniTaskAsyncEnumerable<TSource> Where<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, Int32, Boolean> predicate) { Error.ThrowArgumentNullException(source, nameof(source)); Error.ThrowArgumentNullException(predicate, nameof(predicate)); return new WhereInt<TSource>(source, predicate); } public static IUniTaskAsyncEnumerable<TSource> WhereAwait<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<Boolean>> predicate) { Error.ThrowArgumentNullException(source, nameof(source)); Error.ThrowArgumentNullException(predicate, nameof(predicate)); return new WhereAwait<TSource>(source, predicate); } public static IUniTaskAsyncEnumerable<TSource> WhereAwait<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, Int32, UniTask<Boolean>> predicate) { Error.ThrowArgumentNullException(source, nameof(source)); Error.ThrowArgumentNullException(predicate, nameof(predicate)); return new WhereIntAwait<TSource>(source, predicate); } public static IUniTaskAsyncEnumerable<TSource> WhereAwaitWithCancellation<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<Boolean>> predicate) { Error.ThrowArgumentNullException(source, nameof(source)); Error.ThrowArgumentNullException(predicate, nameof(predicate)); return new WhereAwaitWithCancellation<TSource>(source, predicate); } public static IUniTaskAsyncEnumerable<TSource> WhereAwaitWithCancellation<TSource>(this IUniTaskAsyncEnumerable<TSource> source, Func<TSource, Int32, CancellationToken, UniTask<Boolean>> predicate) { Error.ThrowArgumentNullException(source, nameof(source)); Error.ThrowArgumentNullException(predicate, nameof(predicate)); return new WhereIntAwaitWithCancellation<TSource>(source, predicate); } } internal sealed class Where<TSource> : IUniTaskAsyncEnumerable<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, bool> predicate; public Where(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, bool> predicate) { this.source = source; this.predicate = predicate; } public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default) { return new _Where(source, predicate, cancellationToken); } sealed class _Where : MoveNextSource, IUniTaskAsyncEnumerator<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, bool> predicate; readonly CancellationToken cancellationToken; int state = -1; IUniTaskAsyncEnumerator<TSource> enumerator; UniTask<bool>.Awaiter awaiter; Action moveNextAction; public _Where(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, bool> predicate, CancellationToken cancellationToken) { this.source = source; this.predicate = predicate; this.cancellationToken = cancellationToken; this.moveNextAction = MoveNext; TaskTracker.TrackActiveTask(this, 3); } public TSource Current { get; private set; } public UniTask<bool> MoveNextAsync() { if (state == -2) return default; completionSource.Reset(); MoveNext(); return new UniTask<bool>(this, completionSource.Version); } void MoveNext() { REPEAT: try { switch (state) { case -1: // init enumerator = source.GetAsyncEnumerator(cancellationToken); goto case 0; case 0: awaiter = enumerator.MoveNextAsync().GetAwaiter(); if (awaiter.IsCompleted) { goto case 1; } else { state = 1; awaiter.UnsafeOnCompleted(moveNextAction); return; } case 1: if (awaiter.GetResult()) { Current = enumerator.Current; if (predicate(Current)) { goto CONTINUE; } else { state = 0; goto REPEAT; } } else { goto DONE; } default: goto DONE; } } catch (Exception ex) { state = -2; completionSource.TrySetException(ex); return; } DONE: state = -2; completionSource.TrySetResult(false); return; CONTINUE: state = 0; completionSource.TrySetResult(true); return; } public UniTask DisposeAsync() { TaskTracker.RemoveTracking(this); return enumerator.DisposeAsync(); } } } internal sealed class WhereInt<TSource> : IUniTaskAsyncEnumerable<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, int, bool> predicate; public WhereInt(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, int, bool> predicate) { this.source = source; this.predicate = predicate; } public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default) { return new _Where(source, predicate, cancellationToken); } sealed class _Where : MoveNextSource, IUniTaskAsyncEnumerator<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, int, bool> predicate; readonly CancellationToken cancellationToken; int state = -1; IUniTaskAsyncEnumerator<TSource> enumerator; UniTask<bool>.Awaiter awaiter; Action moveNextAction; int index; public _Where(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, int, bool> predicate, CancellationToken cancellationToken) { this.source = source; this.predicate = predicate; this.cancellationToken = cancellationToken; this.moveNextAction = MoveNext; TaskTracker.TrackActiveTask(this, 3); } public TSource Current { get; private set; } public UniTask<bool> MoveNextAsync() { if (state == -2) return default; completionSource.Reset(); MoveNext(); return new UniTask<bool>(this, completionSource.Version); } void MoveNext() { REPEAT: try { switch (state) { case -1: // init enumerator = source.GetAsyncEnumerator(cancellationToken); goto case 0; case 0: awaiter = enumerator.MoveNextAsync().GetAwaiter(); if (awaiter.IsCompleted) { goto case 1; } else { state = 1; awaiter.UnsafeOnCompleted(moveNextAction); return; } case 1: if (awaiter.GetResult()) { Current = enumerator.Current; if (predicate(Current, checked(index++))) { goto CONTINUE; } else { state = 0; goto REPEAT; } } else { goto DONE; } default: goto DONE; } } catch (Exception ex) { state = -2; completionSource.TrySetException(ex); return; } DONE: state = -2; completionSource.TrySetResult(false); return; CONTINUE: state = 0; completionSource.TrySetResult(true); return; } public UniTask DisposeAsync() { TaskTracker.RemoveTracking(this); return enumerator.DisposeAsync(); } } } internal sealed class WhereAwait<TSource> : IUniTaskAsyncEnumerable<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, UniTask<bool>> predicate; public WhereAwait(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<bool>> predicate) { this.source = source; this.predicate = predicate; } public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default) { return new _WhereAwait(source, predicate, cancellationToken); } sealed class _WhereAwait : MoveNextSource, IUniTaskAsyncEnumerator<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, UniTask<bool>> predicate; readonly CancellationToken cancellationToken; int state = -1; IUniTaskAsyncEnumerator<TSource> enumerator; UniTask<bool>.Awaiter awaiter; UniTask<bool>.Awaiter awaiter2; Action moveNextAction; public _WhereAwait(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, UniTask<bool>> predicate, CancellationToken cancellationToken) { this.source = source; this.predicate = predicate; this.cancellationToken = cancellationToken; this.moveNextAction = MoveNext; TaskTracker.TrackActiveTask(this, 3); } public TSource Current { get; private set; } public UniTask<bool> MoveNextAsync() { if (state == -2) return default; completionSource.Reset(); MoveNext(); return new UniTask<bool>(this, completionSource.Version); } void MoveNext() { REPEAT: try { switch (state) { case -1: // init enumerator = source.GetAsyncEnumerator(cancellationToken); goto case 0; case 0: awaiter = enumerator.MoveNextAsync().GetAwaiter(); if (awaiter.IsCompleted) { goto case 1; } else { state = 1; awaiter.UnsafeOnCompleted(moveNextAction); return; } case 1: if (awaiter.GetResult()) { Current = enumerator.Current; awaiter2 = predicate(Current).GetAwaiter(); if (awaiter2.IsCompleted) { goto case 2; } else { state = 2; awaiter2.UnsafeOnCompleted(moveNextAction); return; } } else { goto DONE; } case 2: if (awaiter2.GetResult()) { goto CONTINUE; } else { state = 0; goto REPEAT; } default: goto DONE; } } catch (Exception ex) { state = -2; completionSource.TrySetException(ex); return; } DONE: state = -2; completionSource.TrySetResult(false); return; CONTINUE: state = 0; completionSource.TrySetResult(true); return; } public UniTask DisposeAsync() { TaskTracker.RemoveTracking(this); return enumerator.DisposeAsync(); } } } internal sealed class WhereIntAwait<TSource> : IUniTaskAsyncEnumerable<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, int, UniTask<bool>> predicate; public WhereIntAwait(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, int, UniTask<bool>> predicate) { this.source = source; this.predicate = predicate; } public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default) { return new _WhereAwait(source, predicate, cancellationToken); } sealed class _WhereAwait : MoveNextSource, IUniTaskAsyncEnumerator<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, int, UniTask<bool>> predicate; readonly CancellationToken cancellationToken; int state = -1; IUniTaskAsyncEnumerator<TSource> enumerator; UniTask<bool>.Awaiter awaiter; UniTask<bool>.Awaiter awaiter2; Action moveNextAction; int index; public _WhereAwait(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, int, UniTask<bool>> predicate, CancellationToken cancellationToken) { this.source = source; this.predicate = predicate; this.cancellationToken = cancellationToken; this.moveNextAction = MoveNext; TaskTracker.TrackActiveTask(this, 3); } public TSource Current { get; private set; } public UniTask<bool> MoveNextAsync() { if (state == -2) return default; completionSource.Reset(); MoveNext(); return new UniTask<bool>(this, completionSource.Version); } void MoveNext() { REPEAT: try { switch (state) { case -1: // init enumerator = source.GetAsyncEnumerator(cancellationToken); goto case 0; case 0: awaiter = enumerator.MoveNextAsync().GetAwaiter(); if (awaiter.IsCompleted) { goto case 1; } else { state = 1; awaiter.UnsafeOnCompleted(moveNextAction); return; } case 1: if (awaiter.GetResult()) { Current = enumerator.Current; awaiter2 = predicate(Current, checked(index++)).GetAwaiter(); if (awaiter2.IsCompleted) { goto case 2; } else { state = 2; awaiter2.UnsafeOnCompleted(moveNextAction); return; } } else { goto DONE; } case 2: if (awaiter2.GetResult()) { goto CONTINUE; } else { state = 0; goto REPEAT; } default: goto DONE; } } catch (Exception ex) { state = -2; completionSource.TrySetException(ex); return; } DONE: state = -2; completionSource.TrySetResult(false); return; CONTINUE: state = 0; completionSource.TrySetResult(true); return; } public UniTask DisposeAsync() { TaskTracker.RemoveTracking(this); return enumerator.DisposeAsync(); } } } internal sealed class WhereAwaitWithCancellation<TSource> : IUniTaskAsyncEnumerable<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, CancellationToken, UniTask<bool>> predicate; public WhereAwaitWithCancellation(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<bool>> predicate) { this.source = source; this.predicate = predicate; } public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default) { return new _WhereAwaitWithCancellation(source, predicate, cancellationToken); } sealed class _WhereAwaitWithCancellation : MoveNextSource, IUniTaskAsyncEnumerator<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, CancellationToken, UniTask<bool>> predicate; readonly CancellationToken cancellationToken; int state = -1; IUniTaskAsyncEnumerator<TSource> enumerator; UniTask<bool>.Awaiter awaiter; UniTask<bool>.Awaiter awaiter2; Action moveNextAction; public _WhereAwaitWithCancellation(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, CancellationToken, UniTask<bool>> predicate, CancellationToken cancellationToken) { this.source = source; this.predicate = predicate; this.cancellationToken = cancellationToken; this.moveNextAction = MoveNext; TaskTracker.TrackActiveTask(this, 3); } public TSource Current { get; private set; } public UniTask<bool> MoveNextAsync() { if (state == -2) return default; completionSource.Reset(); MoveNext(); return new UniTask<bool>(this, completionSource.Version); } void MoveNext() { REPEAT: try { switch (state) { case -1: // init enumerator = source.GetAsyncEnumerator(cancellationToken); goto case 0; case 0: awaiter = enumerator.MoveNextAsync().GetAwaiter(); if (awaiter.IsCompleted) { goto case 1; } else { state = 1; awaiter.UnsafeOnCompleted(moveNextAction); return; } case 1: if (awaiter.GetResult()) { Current = enumerator.Current; awaiter2 = predicate(Current, cancellationToken).GetAwaiter(); if (awaiter2.IsCompleted) { goto case 2; } else { state = 2; awaiter2.UnsafeOnCompleted(moveNextAction); return; } } else { goto DONE; } case 2: if (awaiter2.GetResult()) { goto CONTINUE; } else { state = 0; goto REPEAT; } default: goto DONE; } } catch (Exception ex) { state = -2; completionSource.TrySetException(ex); return; } DONE: state = -2; completionSource.TrySetResult(false); return; CONTINUE: state = 0; completionSource.TrySetResult(true); return; } public UniTask DisposeAsync() { TaskTracker.RemoveTracking(this); return enumerator.DisposeAsync(); } } } internal sealed class WhereIntAwaitWithCancellation<TSource> : IUniTaskAsyncEnumerable<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, int, CancellationToken, UniTask<bool>> predicate; public WhereIntAwaitWithCancellation(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, int, CancellationToken, UniTask<bool>> predicate) { this.source = source; this.predicate = predicate; } public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default) { return new _WhereAwaitWithCancellation(source, predicate, cancellationToken); } sealed class _WhereAwaitWithCancellation : MoveNextSource, IUniTaskAsyncEnumerator<TSource> { readonly IUniTaskAsyncEnumerable<TSource> source; readonly Func<TSource, int, CancellationToken, UniTask<bool>> predicate; readonly CancellationToken cancellationToken; int state = -1; IUniTaskAsyncEnumerator<TSource> enumerator; UniTask<bool>.Awaiter awaiter; UniTask<bool>.Awaiter awaiter2; Action moveNextAction; int index; public _WhereAwaitWithCancellation(IUniTaskAsyncEnumerable<TSource> source, Func<TSource, int, CancellationToken, UniTask<bool>> predicate, CancellationToken cancellationToken) { this.source = source; this.predicate = predicate; this.cancellationToken = cancellationToken; this.moveNextAction = MoveNext; TaskTracker.TrackActiveTask(this, 3); } public TSource Current { get; private set; } public UniTask<bool> MoveNextAsync() { if (state == -2) return default; completionSource.Reset(); MoveNext(); return new UniTask<bool>(this, completionSource.Version); } void MoveNext() { REPEAT: try { switch (state) { case -1: // init enumerator = source.GetAsyncEnumerator(cancellationToken); goto case 0; case 0: awaiter = enumerator.MoveNextAsync().GetAwaiter(); if (awaiter.IsCompleted) { goto case 1; } else { state = 1; awaiter.UnsafeOnCompleted(moveNextAction); return; } case 1: if (awaiter.GetResult()) { Current = enumerator.Current; awaiter2 = predicate(Current, checked(index++), cancellationToken).GetAwaiter(); if (awaiter2.IsCompleted) { goto case 2; } else { state = 2; awaiter2.UnsafeOnCompleted(moveNextAction); return; } } else { goto DONE; } case 2: if (awaiter2.GetResult()) { goto CONTINUE; } else { state = 0; goto REPEAT; } default: goto DONE; } } catch (Exception ex) { state = -2; completionSource.TrySetException(ex); return; } DONE: state = -2; completionSource.TrySetResult(false); return; CONTINUE: state = 0; completionSource.TrySetResult(true); return; } public UniTask DisposeAsync() { TaskTracker.RemoveTracking(this); return enumerator.DisposeAsync(); } } } }