+semver:minor - Merge pull request #33 from thomhurst/feature/Paralle…

…lAsyncProcessorTests

ParallelAsyncProcessorTests

EnumerableAsyncProcessor.UnitTests/ParallelAsyncProcessorTests.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading.Tasks;
+using EnumerableAsyncProcessor.Builders;
+using NUnit.Framework;
+
+namespace EnumerableAsyncProcessor.UnitTests;
+
+[FixtureLifeCycle(LifeCycle.InstancePerTestCase)]
+[Parallelizable(ParallelScope.All)]
+public class ParallelAsyncProcessorTests
+{
+    [Test]
+    public async Task Test()
+    {
+        var stopwatch = Stopwatch.StartNew();
+
+        var processor = AsyncProcessorBuilder
+            .WithExecutionCount(500)
+            .ForEachAsync( () => Task.Delay(100))
+            .ProcessInParallel();
+
+        await processor;
+
+        stopwatch.Stop();
+
+        var completedTasks = processor.GetEnumerableTasks().Count(x => x.IsCompleted);
+
+        Assert.That(completedTasks, Is.EqualTo(500));
+        Assert.That(stopwatch.Elapsed.Milliseconds, Is.EqualTo(100).Within(1000));
+    }
+}

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultParallelAsyncProcessor_1.cs

@@ -1,8 +1,15 @@
-namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors;
+using EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract;
 
-public class ResultParallelAsyncProcessor<TOutput> : ResultRateLimitedParallelAsyncProcessor<TOutput>
+namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors;
+
+public class ResultParallelAsyncProcessor<TOutput> : ResultAbstractAsyncProcessor<TOutput>
 {
-    internal ResultParallelAsyncProcessor(int count, Func<Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource) : base(count, taskSelector, -1, cancellationTokenSource)
+    internal ResultParallelAsyncProcessor(int count, Func<Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource) : base(count, taskSelector, cancellationTokenSource)
+    {
+    }
+
+    internal override Task Process()
     {
+        return Task.WhenAll(EnumerableTaskCompletionSources.Select(ProcessItem));
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultParallelAsyncProcessor_2.cs

@@ -1,10 +1,16 @@
 using System.Collections.Immutable;
+using EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract;
 
 namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors;
 
-public class ResultParallelAsyncProcessor<TInput, TOutput> : ResultRateLimitedParallelAsyncProcessor<TInput, TOutput>
+public class ResultParallelAsyncProcessor<TInput, TOutput> : ResultAbstractAsyncProcessor<TInput, TOutput>
 {
-    internal ResultParallelAsyncProcessor(ImmutableList<TInput> items, Func<TInput, Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource) : base(items, taskSelector, -1, cancellationTokenSource)
+    internal ResultParallelAsyncProcessor(ImmutableList<TInput> items, Func<TInput, Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource) : base(items, taskSelector, cancellationTokenSource)
     {
     }
+
+    internal override Task Process()
+    {
+        return Task.WhenAll(ItemisedTaskCompletionSourceContainers.Select(ProcessItem));
+    }
 }