AlphaaeGameStudio/TEngine
1
0
Fork 0
mirror of https://github.com/Alex-Rachel/TEngine.git synced 2025-08-14 16:51:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

更新Demo

Browse Source 
更新Demo
This commit is contained in:
ALEXTANG
2023-12-20 12:14:27 +08:00
parent f64bb37706
commit 17a5d7425c
847 changed files with 106095 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

232
UnityProject/Packages/UniTask/Runtime/Linq/Merge.cs Normal file
View File

@@ -0,0 +1,232 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using Cysharp.Threading.Tasks.Internal;
namespace Cysharp.Threading.Tasks.Linq
{
public static partial class UniTaskAsyncEnumerable
{
public static IUniTaskAsyncEnumerable<T> Merge<T>(this IUniTaskAsyncEnumerable<T> first, IUniTaskAsyncEnumerable<T> second)
{
Error.ThrowArgumentNullException(first, nameof(first));
Error.ThrowArgumentNullException(second, nameof(second));
return new Merge<T>(new [] { first, second });
}
public static IUniTaskAsyncEnumerable<T> Merge<T>(this IUniTaskAsyncEnumerable<T> first, IUniTaskAsyncEnumerable<T> second, IUniTaskAsyncEnumerable<T> third)
{
Error.ThrowArgumentNullException(first, nameof(first));
Error.ThrowArgumentNullException(second, nameof(second));
Error.ThrowArgumentNullException(third, nameof(third));
return new Merge<T>(new[] { first, second, third });
}
public static IUniTaskAsyncEnumerable<T> Merge<T>(this IEnumerable<IUniTaskAsyncEnumerable<T>> sources)
{
return new Merge<T>(sources.ToArray());
}
public static IUniTaskAsyncEnumerable<T> Merge<T>(params IUniTaskAsyncEnumerable<T>[] sources)
{
return new Merge<T>(sources);
}
}
internal sealed class Merge<T> : IUniTaskAsyncEnumerable<T>
{
readonly IUniTaskAsyncEnumerable<T>[] sources;
public Merge(IUniTaskAsyncEnumerable<T>[] sources)
{
if (sources.Length <= 0)
{
Error.ThrowArgumentException("No source async enumerable to merge");
}
this.sources = sources;
}
public IUniTaskAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
=> new _Merge(sources, cancellationToken);
enum MergeSourceState
{
Pending,
Running,
Completed,
}
sealed class _Merge : MoveNextSource, IUniTaskAsyncEnumerator<T>
{
static readonly Action<object> GetResultAtAction = GetResultAt;
readonly int length;
readonly IUniTaskAsyncEnumerator<T>[] enumerators;
readonly MergeSourceState[] states;
readonly Queue<(T, Exception, bool)> queuedResult = new Queue<(T, Exception, bool)>();
readonly CancellationToken cancellationToken;
int moveNextCompleted;
public T Current { get; private set; }
public _Merge(IUniTaskAsyncEnumerable<T>[] sources, CancellationToken cancellationToken)
{
this.cancellationToken = cancellationToken;
length = sources.Length;
states = ArrayPool<MergeSourceState>.Shared.Rent(length);
enumerators = ArrayPool<IUniTaskAsyncEnumerator<T>>.Shared.Rent(length);
for (var i = 0; i < length; i++)
{
enumerators[i] = sources[i].GetAsyncEnumerator(cancellationToken);
states[i] = (int)MergeSourceState.Pending;;
}
}
public UniTask<bool> MoveNextAsync()
{
cancellationToken.ThrowIfCancellationRequested();
completionSource.Reset();
Interlocked.Exchange(ref moveNextCompleted, 0);
if (HasQueuedResult() && Interlocked.CompareExchange(ref moveNextCompleted, 1, 0) == 0)
{
(T, Exception, bool) value;
lock (states)
{
value = queuedResult.Dequeue();
}
var resultValue = value.Item1;
var exception = value.Item2;
var hasNext = value.Item3;
if (exception != null)
{
completionSource.TrySetException(exception);
}
else
{
Current = resultValue;
completionSource.TrySetResult(hasNext);
}
return new UniTask<bool>(this, completionSource.Version);
}
for (var i = 0; i < length; i++)
{
lock (states)
{
if (states[i] == MergeSourceState.Pending)
{
states[i] = MergeSourceState.Running;
}
else
{
continue;
}
}
var awaiter = enumerators[i].MoveNextAsync().GetAwaiter();
if (awaiter.IsCompleted)
{
GetResultAt(i, awaiter);
}
else
{
awaiter.SourceOnCompleted(GetResultAtAction, StateTuple.Create(this, i, awaiter));
}
}
return new UniTask<bool>(this, completionSource.Version);
}
public async UniTask DisposeAsync()
{
for (var i = 0; i < length; i++)
{
await enumerators[i].DisposeAsync();
}
ArrayPool<MergeSourceState>.Shared.Return(states, true);
ArrayPool<IUniTaskAsyncEnumerator<T>>.Shared.Return(enumerators, true);
}
static void GetResultAt(object state)
{
using (var tuple = (StateTuple<_Merge, int, UniTask<bool>.Awaiter>)state)
{
tuple.Item1.GetResultAt(tuple.Item2, tuple.Item3);
}
}
void GetResultAt(int index, UniTask<bool>.Awaiter awaiter)
{
bool hasNext;
bool completedAll;
try
{
hasNext = awaiter.GetResult();
}
catch (Exception ex)
{
if (Interlocked.CompareExchange(ref moveNextCompleted, 1, 0) == 0)
{
completionSource.TrySetException(ex);
}
else
{
lock (states)
{
queuedResult.Enqueue((default, ex, default));
}
}
return;
}
lock (states)
{
states[index] = hasNext ? MergeSourceState.Pending : MergeSourceState.Completed;
completedAll = !hasNext && IsCompletedAll();
}
if (hasNext || completedAll)
{
if (Interlocked.CompareExchange(ref moveNextCompleted, 1, 0) == 0)
{
Current = enumerators[index].Current;
completionSource.TrySetResult(!completedAll);
}
else
{
lock (states)
{
queuedResult.Enqueue((enumerators[index].Current, null, !completedAll));
}
}
}
}
bool HasQueuedResult()
{
lock (states)
{
return queuedResult.Count > 0;
}
}
bool IsCompletedAll()
{
lock (states)
{
for (var i = 0; i < length; i++)
{
if (states[i] != MergeSourceState.Completed)
{
return false;
}
}
}
return true;
}
}
}
}