Add CombineLatest
parent
21dc83c641
commit
2b87cadba3
|
@ -93,10 +93,16 @@ namespace NetCoreSandbox
|
||||||
|
|
||||||
var channel = Channel.CreateSingleConsumerUnbounded<int>();
|
var channel = Channel.CreateSingleConsumerUnbounded<int>();
|
||||||
|
|
||||||
|
// Observable.Range(1,10).CombineLatest(
|
||||||
|
|
||||||
|
var cts = new CancellationTokenSource();
|
||||||
|
|
||||||
|
var token = cts.Token;
|
||||||
|
|
||||||
|
FooAsync(token).ForEachAsync(x => { }, token);
|
||||||
|
|
||||||
|
|
||||||
|
// Observable.Range(1,10).CombineLatest(
|
||||||
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -319,5 +319,86 @@ namespace NetCoreTests.Linq
|
||||||
await Assert.ThrowsAsync<UniTaskTestException>(async () => await ys);
|
await Assert.ThrowsAsync<UniTaskTestException>(async () => await ys);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task CombineLatestOK()
|
||||||
|
{
|
||||||
|
var a = new AsyncReactiveProperty<int>(0);
|
||||||
|
var b = new AsyncReactiveProperty<int>(0);
|
||||||
|
|
||||||
|
var list = new List<(int, int)>();
|
||||||
|
var complete = a.WithoutCurrent().CombineLatest(b.WithoutCurrent(), (x, y) => (x, y)).ForEachAsync(x => list.Add(x));
|
||||||
|
|
||||||
|
list.Count.Should().Be(0);
|
||||||
|
|
||||||
|
a.Value = 10;
|
||||||
|
list.Count.Should().Be(0);
|
||||||
|
|
||||||
|
a.Value = 20;
|
||||||
|
list.Count.Should().Be(0);
|
||||||
|
|
||||||
|
b.Value = 1;
|
||||||
|
list.Count.Should().Be(1);
|
||||||
|
|
||||||
|
list[0].Should().Be((20, 1));
|
||||||
|
|
||||||
|
a.Value = 30;
|
||||||
|
list.Last().Should().Be((30, 1));
|
||||||
|
|
||||||
|
b.Value = 2;
|
||||||
|
list.Last().Should().Be((30, 2));
|
||||||
|
|
||||||
|
a.Dispose();
|
||||||
|
b.Value = 3;
|
||||||
|
list.Last().Should().Be((30, 3));
|
||||||
|
|
||||||
|
b.Dispose();
|
||||||
|
|
||||||
|
await complete;
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task CombineLatestLong()
|
||||||
|
{
|
||||||
|
var a = UniTaskAsyncEnumerable.Range(1, 100000);
|
||||||
|
var b = new AsyncReactiveProperty<int>(0);
|
||||||
|
|
||||||
|
var list = new List<(int, int)>();
|
||||||
|
var complete = a.CombineLatest(b.WithoutCurrent(), (x, y) => (x, y)).ForEachAsync(x => list.Add(x));
|
||||||
|
|
||||||
|
b.Value = 1;
|
||||||
|
|
||||||
|
list[0].Should().Be((100000, 1));
|
||||||
|
|
||||||
|
b.Dispose();
|
||||||
|
|
||||||
|
await complete;
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task CombineLatestError()
|
||||||
|
{
|
||||||
|
var a = new AsyncReactiveProperty<int>(0);
|
||||||
|
var b = new AsyncReactiveProperty<int>(0);
|
||||||
|
|
||||||
|
var list = new List<(int, int)>();
|
||||||
|
var complete = a.WithoutCurrent()
|
||||||
|
.Select(x => { if (x == 0) { throw new MyException(); } return x; })
|
||||||
|
.CombineLatest(b.WithoutCurrent(), (x, y) => (x, y)).ForEachAsync(x => list.Add(x));
|
||||||
|
|
||||||
|
|
||||||
|
a.Value = 10;
|
||||||
|
b.Value = 1;
|
||||||
|
list.Last().Should().Be((10, 1));
|
||||||
|
|
||||||
|
a.Value = 0;
|
||||||
|
|
||||||
|
await Assert.ThrowsAsync<MyException>(async () => await complete);
|
||||||
|
}
|
||||||
|
|
||||||
|
class MyException : Exception
|
||||||
|
{
|
||||||
|
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
File diff suppressed because it is too large
Load Diff
|
@ -0,0 +1,11 @@
|
||||||
|
fileFormatVersion: 2
|
||||||
|
guid: 6cb07f6e88287e34d9b9301a572284a5
|
||||||
|
MonoImporter:
|
||||||
|
externalObjects: {}
|
||||||
|
serializedVersion: 2
|
||||||
|
defaultReferences: []
|
||||||
|
executionOrder: 0
|
||||||
|
icon: {instanceID: 0}
|
||||||
|
userData:
|
||||||
|
assetBundleName:
|
||||||
|
assetBundleVariant:
|
|
@ -0,0 +1,212 @@
|
||||||
|
<#@ template debug="false" hostspecific="false" language="C#" #>
|
||||||
|
<#@ assembly name="System.Core" #>
|
||||||
|
<#@ import namespace="System.Linq" #>
|
||||||
|
<#@ import namespace="System.Text" #>
|
||||||
|
<#@ import namespace="System.Collections.Generic" #>
|
||||||
|
<#@ output extension=".cs" #>
|
||||||
|
<#
|
||||||
|
var tMax = 15;
|
||||||
|
Func<int, string> typeArgs = x => string.Join(", ", Enumerable.Range(1, x).Select(x => $"T{x}")) + ", TResult";
|
||||||
|
Func<int, string> paramArgs = x => string.Join(", ", Enumerable.Range(1, x).Select(x => $"IUniTaskAsyncEnumerable<T{x}> source{x}"));
|
||||||
|
Func<int, string> parameters = x => string.Join(", ", Enumerable.Range(1, x).Select(x => $"source{x}"));
|
||||||
|
|
||||||
|
|
||||||
|
#>
|
||||||
|
using Cysharp.Threading.Tasks.Internal;
|
||||||
|
using System;
|
||||||
|
using System.Threading;
|
||||||
|
|
||||||
|
namespace Cysharp.Threading.Tasks.Linq
|
||||||
|
{
|
||||||
|
public static partial class UniTaskAsyncEnumerable
|
||||||
|
{
|
||||||
|
<# for(var i = 1; i <= tMax; i++) { #>
|
||||||
|
public static IUniTaskAsyncEnumerable<TResult> CombineLatest<<#= typeArgs(i) #>>(this <#= paramArgs(i) #>, Func<<#= typeArgs(i) #>> resultSelector)
|
||||||
|
{
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
Error.ThrowArgumentNullException(source<#= j #>, nameof(source<#= j #>));
|
||||||
|
<# } #>
|
||||||
|
Error.ThrowArgumentNullException(resultSelector, nameof(resultSelector));
|
||||||
|
|
||||||
|
return new CombineLatest<<#= typeArgs(i) #>>(<#= parameters(i) #>, resultSelector);
|
||||||
|
}
|
||||||
|
|
||||||
|
<# } #>
|
||||||
|
}
|
||||||
|
|
||||||
|
<# for(var i = 1; i <= tMax; i++) { #>
|
||||||
|
internal class CombineLatest<<#= typeArgs(i) #>> : IUniTaskAsyncEnumerable<TResult>
|
||||||
|
{
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
readonly IUniTaskAsyncEnumerable<T<#= j #>> source<#= j #>;
|
||||||
|
<# } #>
|
||||||
|
readonly Func<<#= typeArgs(i) #>> resultSelector;
|
||||||
|
|
||||||
|
public CombineLatest(<#= paramArgs(i) #>, Func<<#= typeArgs(i) #>> resultSelector)
|
||||||
|
{
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
this.source<#= j #> = source<#= j #>;
|
||||||
|
<# } #>
|
||||||
|
this.resultSelector = resultSelector;
|
||||||
|
}
|
||||||
|
|
||||||
|
public IUniTaskAsyncEnumerator<TResult> GetAsyncEnumerator(CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
return new _CombineLatest(<#= parameters(i) #>, resultSelector, cancellationToken);
|
||||||
|
}
|
||||||
|
|
||||||
|
class _CombineLatest : MoveNextSource, IUniTaskAsyncEnumerator<TResult>
|
||||||
|
{
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
static readonly Action<object> Completed<#= j #>Delegate = Completed<#= j #>;
|
||||||
|
<# } #>
|
||||||
|
const int CompleteCount = <#= i #>;
|
||||||
|
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
readonly IUniTaskAsyncEnumerable<T<#= j #>> source<#= j #>;
|
||||||
|
<# } #>
|
||||||
|
readonly Func<<#= typeArgs(i) #>> resultSelector;
|
||||||
|
CancellationToken cancellationToken;
|
||||||
|
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
IUniTaskAsyncEnumerator<T<#= j #>> enumerator<#= j #>;
|
||||||
|
UniTask<bool>.Awaiter awaiter<#= j #>;
|
||||||
|
bool hasCurrent<#= j #>;
|
||||||
|
bool running<#= j #>;
|
||||||
|
T<#= j #> current<#= j #>;
|
||||||
|
|
||||||
|
<# } #>
|
||||||
|
|
||||||
|
int completedCount;
|
||||||
|
bool syncRunning;
|
||||||
|
TResult result;
|
||||||
|
|
||||||
|
public _CombineLatest(<#= paramArgs(i) #>, Func<<#= typeArgs(i) #>> resultSelector, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
this.source<#= j #> = source<#= j #>;
|
||||||
|
<# } #>
|
||||||
|
this.resultSelector = resultSelector;
|
||||||
|
this.cancellationToken = cancellationToken;
|
||||||
|
}
|
||||||
|
|
||||||
|
public TResult Current => result;
|
||||||
|
|
||||||
|
public UniTask<bool> MoveNextAsync()
|
||||||
|
{
|
||||||
|
cancellationToken.ThrowIfCancellationRequested();
|
||||||
|
if (completedCount == CompleteCount) return CompletedTasks.False;
|
||||||
|
|
||||||
|
if (enumerator1 == null)
|
||||||
|
{
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
enumerator<#= j #> = source<#= j #>.GetAsyncEnumerator(cancellationToken);
|
||||||
|
<# } #>
|
||||||
|
}
|
||||||
|
|
||||||
|
completionSource.Reset();
|
||||||
|
|
||||||
|
AGAIN:
|
||||||
|
syncRunning = true;
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
if (!running<#= j #>)
|
||||||
|
{
|
||||||
|
running<#= j #> = true;
|
||||||
|
awaiter<#= j #> = enumerator<#= j #>.MoveNextAsync().GetAwaiter();
|
||||||
|
awaiter<#= j #>.SourceOnCompleted(Completed<#= j #>Delegate, this);
|
||||||
|
}
|
||||||
|
<# } #>
|
||||||
|
|
||||||
|
if (<#= string.Join(" || ", Enumerable.Range(1, i).Select(x => $"!running{x}")) #>)
|
||||||
|
{
|
||||||
|
goto AGAIN;
|
||||||
|
}
|
||||||
|
syncRunning = false;
|
||||||
|
|
||||||
|
return new UniTask<bool>(this, completionSource.Version);
|
||||||
|
}
|
||||||
|
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
static void Completed<#= j #>(object state)
|
||||||
|
{
|
||||||
|
var self = (_CombineLatest)state;
|
||||||
|
self.running<#= j #> = false;
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
if (self.awaiter<#= j #>.GetResult())
|
||||||
|
{
|
||||||
|
self.hasCurrent<#= j #> = true;
|
||||||
|
self.current<#= j #> = self.enumerator<#= j #>.Current;
|
||||||
|
goto SUCCESS;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
if (Interlocked.Increment(ref self.completedCount) == CompleteCount)
|
||||||
|
{
|
||||||
|
goto COMPLETE;
|
||||||
|
}
|
||||||
|
self.running<#= j #> = true; // as complete.
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
self.completedCount = CompleteCount;
|
||||||
|
self.completionSource.TrySetException(ex);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
SUCCESS:
|
||||||
|
if (!self.TrySetResult())
|
||||||
|
{
|
||||||
|
if (self.syncRunning) return;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
self.awaiter<#= j #> = self.enumerator<#= j #>.MoveNextAsync().GetAwaiter();
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
self.completedCount = CompleteCount;
|
||||||
|
self.completionSource.TrySetException(ex);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
self.awaiter<#= j #>.SourceOnCompleted(Completed<#= j #>Delegate, self);
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
COMPLETE:
|
||||||
|
self.completionSource.TrySetResult(false);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
<# } #>
|
||||||
|
|
||||||
|
bool TrySetResult()
|
||||||
|
{
|
||||||
|
if (<#= string.Join(" && ", Enumerable.Range(1, i).Select(x => $"hasCurrent{x}")) #>)
|
||||||
|
{
|
||||||
|
result = resultSelector(<#= string.Join(", ", Enumerable.Range(1, i).Select(x => $"current{x}")) #>);
|
||||||
|
completionSource.TrySetResult(true);
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public async UniTask DisposeAsync()
|
||||||
|
{
|
||||||
|
<# for(var j = 1; j <= i; j++) { #>
|
||||||
|
if (enumerator<#= j #> != null)
|
||||||
|
{
|
||||||
|
await enumerator<#= j #>.DisposeAsync();
|
||||||
|
}
|
||||||
|
<# } #>
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
<# } #>
|
||||||
|
}
|
|
@ -0,0 +1,7 @@
|
||||||
|
fileFormatVersion: 2
|
||||||
|
guid: b1b8cfa9d17af814a971ee2224aaaaa2
|
||||||
|
DefaultImporter:
|
||||||
|
externalObjects: {}
|
||||||
|
userData:
|
||||||
|
assetBundleName:
|
||||||
|
assetBundleVariant:
|
|
@ -275,6 +275,8 @@ public class SandboxMain : MonoBehaviour
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
okButton.OnClickAsAsyncEnumerable().ForEachAsync(_ =>
|
okButton.OnClickAsAsyncEnumerable().ForEachAsync(_ =>
|
||||||
{
|
{
|
||||||
|
|
Loading…
Reference in New Issue