Merge with Threading 5.10.0

CHANGELOG.md

@@ -1,6 +1,15 @@
 Release Notes
 ====
 
+# 07-15-2024
+<a href="https://www.nuget.org/packages/dotnext/5.8.0">DotNext 5.8.0</a>
+* Added `FirstOrNone` and `LastOrNone` extension methods back from .NEXT 4.x as requested in [247](https://github.com/dotnet/dotNext/issues/247)
+
+<a href="https://www.nuget.org/packages/dotnext.threading/5.10.0">DotNext.Threading 5.10.0</a>
+* Added `TaskQueue<T>` class
+* Added `Completion` optional property to [TaskCompletionPipe&lt;T&gt;](https://dotnet.github.io/dotNext/api/DotNext.Threading.Tasks.TaskCompletionPipe-1.html) that allows to synchronize on full completion of the pipe
+* Added one-shot static methods to [TaskCompletionPipe](https://dotnet.github.io/dotNext/api/DotNext.Threading.Tasks.TaskCompletionPipe.html) to take `IAsyncEnumerable<T>` over tasks as they complete
+
 # 07-09-2024
 <a href="https://www.nuget.org/packages/dotnext.io/5.9.0">DotNext.IO 5.7.1</a>
 * Improved performance of `FileWriter` in some corner cases

README.md

@@ -44,16 +44,15 @@ All these things are implemented in 100% managed code on top of existing .NET AP
 * [NuGet Packages](https://www.nuget.org/profiles/rvsakno)
 
 # What's new
-Release Date: 07-09-2024
+Release Date: 07-15-2024
 
-<a href="https://www.nuget.org/packages/dotnext.io/5.9.0">DotNext.IO 5.7.1</a>
-* Improved performance of `FileWriter` in some corner cases
+<a href="https://www.nuget.org/packages/dotnext/5.8.0">DotNext 5.8.0</a>
+* Added `FirstOrNone` and `LastOrNone` extension methods back from .NEXT 4.x as requested in [247](https://github.com/dotnet/dotNext/issues/247)
 
-<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.7.3">DotNext.Net.Cluster 5.7.3</a>
-* Fixed [244](https://github.com/dotnet/dotNext/issues/244)
-
-<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.7.3">DotNext.AspNetCore.Cluster 5.7.3</a>
-* Fixed [244](https://github.com/dotnet/dotNext/issues/244)
+<a href="https://www.nuget.org/packages/dotnext.threading/5.10.0">DotNext.Threading 5.10.0</a>
+* Added `TaskQueue<T>` class
+* Added `Completion` optional property to [TaskCompletionPipe&lt;T&gt;](https://dotnet.github.io/dotNext/api/DotNext.Threading.Tasks.TaskCompletionPipe-1.html) that allows to synchronize on full completion of the pipe
+* Added one-shot static methods to [TaskCompletionPipe](https://dotnet.github.io/dotNext/api/DotNext.Threading.Tasks.TaskCompletionPipe.html) to take `IAsyncEnumerable<T>` over tasks as they complete
 
 Changelog for previous versions located [here](./CHANGELOG.md).
 

src/DotNext.Tests/Collections/Generic/AsyncEnumerableTests.cs

@@ -35,11 +35,25 @@ public static async Task ForEachTestAsync()
         Equal(3, counter.value);
         counter.value = 0;
 
-        list = new int[] { 1, 2, 10, 11, 15 }.ToAsyncEnumerable();
+        list = new[] { 1, 2, 10, 11, 15 }.ToAsyncEnumerable();
         await list.ForEachAsync(counter.Accept);
         Equal(5, counter.value);
     }
 
+    [Fact]
+    public static async Task ForEachTest1Async()
+    {
+        var list = new List<int> { 1, 10, 20 }.ToAsyncEnumerable();
+        var counter = new CollectionTests.Counter<int>();
+        await list.ForEachAsync(counter.AcceptAsync);
+        Equal(3, counter.value);
+        counter.value = 0;
+
+        list = new[] { 1, 2, 10, 11, 15 }.ToAsyncEnumerable();
+        await list.ForEachAsync(counter.AcceptAsync);
+        Equal(5, counter.value);
+    }
+
     [Fact]
     public static async Task FirstOrNullTestAsync()
     {

src/DotNext.Tests/Collections/Generic/CollectionTests.cs

@@ -1,4 +1,6 @@
 using System.Collections.Concurrent;
+using System.Collections.Immutable;
+using System.Runtime.InteropServices;
 
 namespace DotNext.Collections.Generic;
 
@@ -50,6 +52,21 @@ internal sealed class Counter<T>
         public int value;
 
         public void Accept(T item) => value += 1;
+
+        public ValueTask AcceptAsync(T item, CancellationToken token)
+        {
+            ValueTask task = ValueTask.CompletedTask;
+            try
+            {
+                Accept(item);
+            }
+            catch (Exception e)
+            {
+                task = ValueTask.FromException(e);
+            }
+
+            return task;
+        }
     }
 
     [Fact]
@@ -64,6 +81,20 @@ public static void ForEachTest()
         array2.ForEach(counter.Accept);
         Equal(5, counter.value);
     }
+
+    [Fact]
+    public static async Task ForEachTestAsync()
+    {
+        IList<int> list = new List<int> { 1, 10, 20 };
+        var counter = new Counter<int>();
+        await list.ForEachAsync(counter.AcceptAsync);
+        Equal(3, counter.value);
+        counter.value = 0;
+
+        var array2 = new int[] { 1, 2, 10, 11, 15 };
+        await array2.ForEachAsync(counter.AcceptAsync);
+        Equal(5, counter.value);
+    }
 
     [Fact]
     public static void ElementAtIndex()
@@ -260,4 +291,57 @@ public static void CopyString()
         using var copy = "abcd".Copy();
         Equal("abcd", copy.Memory.ToString());
     }
+
+    [Fact]
+    public static void FirstOrNone()
+    {
+        Equal(5, new[] { 5, 6 }.FirstOrNone());
+        Equal(5, new List<int> { 5, 6 }.FirstOrNone());
+        Equal(5, new LinkedList<int>([5, 6]).FirstOrNone());
+        Equal('5', "56".FirstOrNone());
+        Equal(5, ImmutableArray.Create([5, 6]).FirstOrNone());
+        Equal(5, GetValues().FirstOrNone());
+
+        Equal(Optional<int>.None, Array.Empty<int>().FirstOrNone());
+        Equal(Optional<int>.None, new List<int>().FirstOrNone());
+        Equal(Optional<int>.None, new LinkedList<int>().FirstOrNone());
+        Equal(Optional<char>.None, string.Empty.FirstOrNone());
+        Equal(Optional<int>.None, ImmutableArray<int>.Empty.FirstOrNone());
+        Equal(Optional<int>.None, EmptyEnumerable<int>().FirstOrNone());
+
+        static IEnumerable<int> GetValues()
+        {
+            yield return 5;
+            yield return 6;
+        }
+    }
+
+    [Fact]
+    public static void LastOrNone()
+    {
+        Equal(6, new[] { 5, 6 }.LastOrNone());
+        Equal(6, new List<int> { 5, 6 }.LastOrNone());
+        Equal(6, new LinkedList<int>([5, 6]).LastOrNone());
+        Equal('6', "56".LastOrNone());
+        Equal(6, ImmutableArray.Create([5, 6]).LastOrNone());
+        Equal(6, GetValues().LastOrNone());
+
+        Equal(Optional<int>.None, Array.Empty<int>().LastOrNone());
+        Equal(Optional<int>.None, new List<int>().LastOrNone());
+        Equal(Optional<int>.None, new LinkedList<int>().LastOrNone());
+        Equal(Optional<char>.None, string.Empty.LastOrNone());
+        Equal(Optional<int>.None, ImmutableArray<int>.Empty.LastOrNone());
+        Equal(Optional<int>.None, EmptyEnumerable<int>().LastOrNone());
+
+        static IEnumerable<int> GetValues()
+        {
+            yield return 5;
+            yield return 6;
+        }
+    }
+
+    static IEnumerable<T> EmptyEnumerable<T>()
+    {
+        yield break;
+    }
 }

src/DotNext.Tests/Threading/AsyncBridgeTests.cs

@@ -133,4 +133,49 @@ public static async Task Interruption()
         var e = await ThrowsAsync<PendingTaskInterruptedException>(Func.Constant(task));
         Equal(interruptionReason, e.Reason);
     }
+
+    [Fact]
+    public static void CompletedTaskAsToken()
+    {
+        var token = Task.CompletedTask.AsCancellationToken();
+        True(token.IsCancellationRequested);
+
+        token = Task.CompletedTask.AsCancellationToken(out var diposeSource);
+        True(token.IsCancellationRequested);
+        False(diposeSource());
+    }
+
+    [Fact]
+    public static async Task TaskAsToken()
+    {
+        var source = new TaskCompletionSource();
+        var token = source.Task.AsCancellationToken();
+        False(token.IsCancellationRequested);
+
+        source.SetResult();
+        await token.WaitAsync();
+    }
+
+    [Fact]
+    public static void DisposeTaskTokenBeforeCompletion()
+    {
+        var source = new TaskCompletionSource();
+        var token = source.Task.AsCancellationToken(out var disposeTokenSource);
+        False(token.IsCancellationRequested);
+
+        True(disposeTokenSource());
+        source.SetResult();
+    }
+
+    [Fact]
+    public static async Task DisposeTaskTokenAfterCompletion()
+    {
+        var source = new TaskCompletionSource();
+        var token = source.Task.AsCancellationToken(out var disposeTokenSource);
+        False(token.IsCancellationRequested);
+
+        source.SetResult();
+        await token.WaitAsync();
+        False(disposeTokenSource());
+    }
 }

src/DotNext.Tests/Threading/AsyncTriggerTests.cs

@@ -56,6 +56,22 @@ public static async Task SignalAndWait()
 
         await task2;
     }
+
+    [Fact]
+    public static async Task SignalAndWaitWithTimeout()
+    {
+        using var trigger = new AsyncTrigger();
+
+        var task1 = trigger.WaitAsync();
+        var task2 = trigger.SignalAndWaitAsync(false, true, DefaultTimeout);
+
+        await task1;
+        False(task2.IsCompleted);
+
+        True(trigger.Signal());
+
+        True(await task2);
+    }
 
     [Fact]
     public static async Task SignalEmptyQueue()

src/DotNext.Tests/Threading/Tasks/TaskCompletionPipeTests.cs

@@ -158,4 +158,56 @@ public static async Task WrongIteratorVersion()
         pipe.Add(Task.FromResult(42));
         False(await enumerator.MoveNextAsync());
     }
+
+    [Fact]
+    public static async Task CompletedTaskGroupToCollection()
+    {
+        await foreach (var t in TaskCompletionPipe.Create([Task.CompletedTask, Task.CompletedTask]))
+        {
+            True(t.IsCompleted);
+        }
+    }
+
+    [Fact]
+    public static async Task TaskGroupToCollection()
+    {
+        var source1 = new TaskCompletionSource<int>();
+        var source2 = new TaskCompletionSource<int>();
+        await using var consumer = TaskCompletionPipe.GetConsumer([source1.Task, source2.Task]).GetAsyncEnumerator();
+
+        source1.SetResult(42);
+        True(await consumer.MoveNextAsync());
+        Equal(42, consumer.Current);
+
+        source2.SetResult(43);
+        True(await consumer.MoveNextAsync());
+        Equal(43, consumer.Current);
+
+        False(await consumer.MoveNextAsync());
+    }
+
+    [Fact]
+    public static async Task CompletionTask()
+    {
+        var pipe = new TaskCompletionPipe<Task> { IsCompletionTaskSupported = true };
+        True(pipe.IsCompletionTaskSupported);
+
+        var source1 = new TaskCompletionSource();
+        var source2 = new TaskCompletionSource();
+
+        pipe.Add([source1.Task, source2.Task], complete: true);
+
+        source1.SetResult();
+        source2.SetResult();
+
+        await pipe.Completion.WaitAsync(DefaultTimeout);
+
+        var count = 0;
+        while (pipe.TryRead(out _))
+        {
+            count++;
+        }
+
+        Equal(2, count);
+    }
 }