append prepend concat

master
neuecc 2020-05-10 03:50:29 +09:00
parent 716decd199
commit af6dbd8868
7 changed files with 1129 additions and 2335 deletions

View File

@ -1,775 +0,0 @@
namespace Cysharp.Threading.Tasks.Linq
{
internal sealed class Append
{
}
}

View File

@ -0,0 +1,913 @@
using Cysharp.Threading.Tasks.Internal;
using System;
using System.Threading;
namespace Cysharp.Threading.Tasks.Linq
{
public static partial class UniTaskAsyncEnumerable
{
public static IUniTaskAsyncEnumerable<TSource> Append<TSource>(this IUniTaskAsyncEnumerable<TSource> source, TSource element)
{
Error.ThrowArgumentNullException(source, nameof(source));
return new AppendPrepend<TSource>(source, element, true);
}
public static IUniTaskAsyncEnumerable<TSource> Prepend<TSource>(this IUniTaskAsyncEnumerable<TSource> source, TSource element)
{
Error.ThrowArgumentNullException(source, nameof(source));
return new AppendPrepend<TSource>(source, element, true);
}
}
internal sealed class AppendPrepend<TSource> : IUniTaskAsyncEnumerable<TSource>
{
readonly IUniTaskAsyncEnumerable<TSource> source;
readonly TSource element;
readonly bool append; // or prepend
public AppendPrepend(IUniTaskAsyncEnumerable<TSource> source, TSource element, bool append)
{
this.source = source;
this.element = element;
this.append = append;
}
public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default)
{
return new Enumerator(source, element, append, cancellationToken);
}
sealed class Enumerator : MoveNextSource, IUniTaskAsyncEnumerator<TSource>
{
enum State : byte
{
None,
RequirePrepend,
RequireAppend,
Completed
}
static readonly Action<object> MoveNextCoreDelegate = MoveNextCore;
readonly IUniTaskAsyncEnumerable<TSource> source;
readonly TSource element;
CancellationToken cancellationToken;
State state;
IUniTaskAsyncEnumerator<TSource> enumerator;
UniTask<bool>.Awaiter awaiter;
public Enumerator(IUniTaskAsyncEnumerable<TSource> source, TSource element, bool append, CancellationToken cancellationToken)
{
this.source = source;
this.element = element;
this.state = append ? State.RequireAppend : State.RequirePrepend;
this.cancellationToken = cancellationToken;
}
public TSource Current { get; private set; }
public UniTask<bool> MoveNextAsync()
{
cancellationToken.ThrowIfCancellationRequested();
completionSource.Reset();
if (enumerator == null)
{
if (state == State.RequireAppend)
{
Current = element;
state = State.None;
return CompletedTasks.True;
}
enumerator = source.GetAsyncEnumerator(cancellationToken);
}
if (state == State.Completed)
{
return CompletedTasks.False;
}
awaiter = enumerator.MoveNextAsync().GetAwaiter();
if (awaiter.IsCompleted)
{
MoveNextCoreDelegate(this);
}
else
{
awaiter.SourceOnCompleted(MoveNextCoreDelegate, this);
}
return new UniTask<bool>(this, completionSource.Version);
}
static void MoveNextCore(object state)
{
var self = (Enumerator)state;
if (self.awaiter.GetResult())
{
self.Current = self.enumerator.Current;
}
else
{
if (self.state == State.RequireAppend)
{
self.state = State.Completed;
self.Current = self.element;
}
else
{
self.state = State.Completed;
self.completionSource.TrySetResult(false);
return;
}
}
self.completionSource.TrySetResult(true);
}
public UniTask DisposeAsync()
{
if (enumerator != null)
{
return enumerator.DisposeAsync();
}
return default;
}
}
}
}

View File

@ -153,6 +153,8 @@ namespace Cysharp.Threading.Tasks.Linq
public abstract class AsyncEnumeratorAwaitSelectorBase<TSource, TResult, TAwait> : MoveNextSource, IUniTaskAsyncEnumerator<TResult>
{
static readonly Action<object> moveNextCallbackDelegate = MoveNextCallBack;
static readonly Action<object> setCurrentCallbackDelegate = SetCurrentCallBack;
readonly IUniTaskAsyncEnumerable<TSource> source;
protected CancellationToken cancellationToken;
@ -274,7 +276,7 @@ namespace Cysharp.Threading.Tasks.Linq
}
else
{
resultAwaiter.SourceOnCompleted(SetCurrentCallBack, this); // TODO:cache
resultAwaiter.SourceOnCompleted(setCurrentCallbackDelegate, this);
result = default;
return false;
}

View File

@ -1,775 +1,151 @@
namespace Cysharp.Threading.Tasks.Linq
using Cysharp.Threading.Tasks.Internal;
using System;
using System.Threading;
namespace Cysharp.Threading.Tasks.Linq
{
internal sealed class Concat
public static partial class UniTaskAsyncEnumerable
{
public static IUniTaskAsyncEnumerable<TSource> Concat<TSource>(this IUniTaskAsyncEnumerable<TSource> first, IUniTaskAsyncEnumerable<TSource> second)
{
Error.ThrowArgumentNullException(first, nameof(first));
Error.ThrowArgumentNullException(second, nameof(second));
return new Concat<TSource>(first, second);
}
}
}
internal sealed class Concat<TSource> : IUniTaskAsyncEnumerable<TSource>
{
readonly IUniTaskAsyncEnumerable<TSource> first;
readonly IUniTaskAsyncEnumerable<TSource> second;
public Concat(IUniTaskAsyncEnumerable<TSource> first, IUniTaskAsyncEnumerable<TSource> second)
{
this.first = first;
this.second = second;
}
public IUniTaskAsyncEnumerator<TSource> GetAsyncEnumerator(CancellationToken cancellationToken = default)
{
return new Enumerator(first, second, cancellationToken);
}
sealed class Enumerator : MoveNextSource, IUniTaskAsyncEnumerator<TSource>
{
static readonly Action<object> MoveNextCoreDelegate = MoveNextCore;
enum IteratingState
{
IteratingFirst,
IteratingSecond,
Complete
}
readonly IUniTaskAsyncEnumerable<TSource> first;
readonly IUniTaskAsyncEnumerable<TSource> second;
CancellationToken cancellationToken;
IteratingState iteratingState;
IUniTaskAsyncEnumerator<TSource> enumerator;
UniTask<bool>.Awaiter awaiter;
public Enumerator(IUniTaskAsyncEnumerable<TSource> first, IUniTaskAsyncEnumerable<TSource> second, CancellationToken cancellationToken)
{
this.first = first;
this.second = second;
this.cancellationToken = cancellationToken;
this.iteratingState = IteratingState.IteratingFirst;
}
public TSource Current { get; private set; }
public UniTask<bool> MoveNextAsync()
{
cancellationToken.ThrowIfCancellationRequested();
if (iteratingState == IteratingState.Complete) return CompletedTasks.False;
completionSource.Reset();
StartIterate();
return new UniTask<bool>(this, completionSource.Version);
}
void StartIterate()
{
if (enumerator == null)
{
if (iteratingState == IteratingState.IteratingFirst)
{
enumerator = first.GetAsyncEnumerator(cancellationToken);
}
else if (iteratingState == IteratingState.IteratingSecond)
{
enumerator = second.GetAsyncEnumerator(cancellationToken);
}
}
awaiter = enumerator.MoveNextAsync().GetAwaiter();
if (awaiter.IsCompleted)
{
MoveNextCoreDelegate(this);
}
else
{
awaiter.SourceOnCompleted(MoveNextCoreDelegate, this);
}
}
static void MoveNextCore(object state)
{
var self = (Enumerator)state;
if (self.awaiter.GetResult())
{
self.Current = self.enumerator.Current;
self.completionSource.TrySetResult(true);
}
else
{
if (self.iteratingState == IteratingState.IteratingFirst)
{
self.RunSecondAfterDisposeAsync().Forget();
return;
}
self.iteratingState = IteratingState.Complete;
self.completionSource.TrySetResult(false);
}
}
async UniTaskVoid RunSecondAfterDisposeAsync()
{
try
{
await enumerator.DisposeAsync();
enumerator = null;
awaiter = default;
iteratingState = IteratingState.IteratingSecond;
}
catch (Exception ex)
{
completionSource.TrySetException(ex);
}
StartIterate();
}
public UniTask DisposeAsync()
{
if (enumerator != null)
{
return enumerator.DisposeAsync();
}
return default;
}
}
}
}

View File

@ -1,775 +0,0 @@
namespace Cysharp.Threading.Tasks.Linq
{
internal sealed class Prepend
{
}
}

View File

@ -23,21 +23,12 @@ namespace ___Dummy
public static IUniTaskAsyncEnumerable<TSource> Append<TSource>(this IUniTaskAsyncEnumerable<TSource> source, TSource element)
{
throw new NotImplementedException();
}
public static IUniTaskAsyncEnumerable<TResult> Cast<TResult>(this IUniTaskAsyncEnumerable<Object> source)
{
throw new NotImplementedException();
}
public static IUniTaskAsyncEnumerable<TSource> Concat<TSource>(this IUniTaskAsyncEnumerable<TSource> first, IUniTaskAsyncEnumerable<TSource> second)
{
throw new NotImplementedException();
}
@ -333,10 +324,7 @@ namespace ___Dummy
throw new NotImplementedException();
}
public static IUniTaskAsyncEnumerable<TSource> Prepend<TSource>(this IUniTaskAsyncEnumerable<TSource> source, TSource element)
{
throw new NotImplementedException();
}
public static IUniTaskAsyncEnumerable<TSource> Reverse<TSource>(this IUniTaskAsyncEnumerable<TSource> source)
{

View File

@ -0,0 +1,65 @@
using Cysharp.Threading.Tasks;
using Cysharp.Threading.Tasks.Linq;
using FluentAssertions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Threading.Tasks;
using Xunit;
namespace NetCoreTests.Linq
{
public class Concat
{
[Theory]
[InlineData(0, 0)]
[InlineData(0, 1)]
[InlineData(0, 2)]
[InlineData(0, 10)]
public async Task Append(int start, int count)
{
var xs = await Enumerable.Range(start, count).ToUniTaskAsyncEnumerable().Append(99).ToArrayAsync();
var ys = Enumerable.Range(start, count).Append(99).ToArray();
xs.Should().BeEquivalentTo(ys);
}
[Theory]
[InlineData(0, 0)]
[InlineData(0, 1)]
[InlineData(0, 2)]
[InlineData(0, 10)]
public async Task Prepend(int start, int count)
{
var xs = await Enumerable.Range(start, count).ToUniTaskAsyncEnumerable().Prepend(99).ToArrayAsync();
var ys = Enumerable.Range(start, count).Prepend(99).ToArray();
xs.Should().BeEquivalentTo(ys);
}
public static IEnumerable<object[]> array1 = new object[][]
{
new object[] { (0, 0), (0, 0) }, // empty + empty
new object[] { (0, 1), (0, 0) }, // 1 + empty
new object[] { (0, 0), (0, 1) }, // empty + 1
new object[] { (0, 5), (0, 0) }, // 5 + empty
new object[] { (0, 0), (0, 5) }, // empty + 5
new object[] { (0, 5), (0, 5) }, // 5 + 5
};
[Theory]
[MemberData(nameof(array1))]
public async Task ConcatTest((int, int) left, (int, int) right)
{
var l = Enumerable.Range(left.Item1, left.Item2);
var r = Enumerable.Range(right.Item1, right.Item2);
var xs = await l.ToUniTaskAsyncEnumerable().Concat(r.ToUniTaskAsyncEnumerable()).ToArrayAsync();
var ys = l.Concat(r).ToArray();
xs.Should().BeEquivalentTo(ys);
}
}
}