|
- 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();
- }
- }
- }
- }
|