ModuleCompleter.cs should filter unique values

module/PSParallelPipeline.psd1

@@ -11,7 +11,7 @@
     RootModule        = 'bin/netstandard2.0/PSParallelPipeline.dll'
 
     # Version number of this module.
-    ModuleVersion     = '1.2.4'
+    ModuleVersion     = '1.2.5'
 
     # Supported PSEditions
     # CompatiblePSEditions = @()

src/PSParallelPipeline/Commands/InvokeParallelCommand.cs

@@ -3,7 +3,6 @@
 using System.Management.Automation;
 using System.Management.Automation.Runspaces;
 using System.Threading;
-using PSParallelPipeline.Poly;
 
 namespace PSParallelPipeline.Commands;
 
@@ -80,12 +79,15 @@
             ScriptBlock.GetUsingParameters(this));
 
         _worker = new Worker(poolSettings, workerSettings, _cts.Token);
-        _worker.Run();
     }
 
     protected override void ProcessRecord()
     {
-        Dbg.Assert(_worker is not null);
+        if (_worker is null)
+        {
+            return;
+        }
+
         InputObject.ThrowIfInputObjectIsScriptBlock(this);
 
         try
@@ -103,14 +105,17 @@
         }
         catch (OperationCanceledException exception)
         {
-            CancelAndWait();
+            _worker.WaitForCompletion();
             exception.WriteTimeoutError(this);
         }
     }
 
     protected override void EndProcessing()
     {
-        Dbg.Assert(_worker is not null);
+        if (_worker is null)
+        {
+            return;
+        }
 
         try
         {
@@ -129,7 +134,7 @@
         }
         catch (OperationCanceledException exception)
         {
-            CancelAndWait();
+            _worker.WaitForCompletion();
             exception.WriteTimeoutError(this);
         }
     }

src/PSParallelPipeline/Extensions.cs

@@ -209,14 +209,28 @@ private static string ResolvePath(this PSCmdlet cmdlet, string path)
             .InvokeReturnAsIs();
     }
 
-    internal static Task InvokePowerShellAsync(
+    internal static Task InvokePowerShellAsync<TOut>(
         this PowerShell powerShell,
-        PSDataCollection<PSObject> output)
+        PSDataCollection<TOut> output)
         => Task.Factory.FromAsync(
-            powerShell.BeginInvoke<PSObject, PSObject>(null, output),
+            powerShell.BeginInvoke<PSObject, TOut>(null, output),
             powerShell.EndInvoke);
 
     internal static ConfiguredTaskAwaitable NoContext(this Task task) => task.ConfigureAwait(false);
 
     internal static ConfiguredTaskAwaitable<T> NoContext<T>(this Task<T> task) => task.ConfigureAwait(false);
+
+    public static IEnumerable<TSource> DistinctBy<TSource, TKey>(
+        this IEnumerable<TSource> source,
+        Func<TSource, TKey> keySelector)
+    {
+        HashSet<TKey> seenKeys = [];
+        foreach (TSource element in source)
+        {
+            if (seenKeys.Add(keySelector(element)))
+            {
+                yield return element;
+            }
+        }
+    }
 }

src/PSParallelPipeline/ModuleCompleter.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections;
 using System.Collections.Generic;
+using System.Linq;
 using System.Management.Automation;
 using System.Management.Automation.Language;
 
@@ -15,17 +16,16 @@ public IEnumerable<CompletionResult> CompleteArgument(
         CommandAst commandAst,
         IDictionary fakeBoundParameters)
     {
-        using PowerShell ps = PowerShell
+        using PowerShell powershell = PowerShell
             .Create(RunspaceMode.CurrentRunspace)
             .AddCommand("Get-Module")
             .AddParameter("ListAvailable");
 
-        foreach (PSModuleInfo module in ps.Invoke<PSModuleInfo>())
-        {
-            if (module.Name.StartsWith(wordToComplete, StringComparison.InvariantCultureIgnoreCase))
-            {
-                yield return new CompletionResult(module.Name);
-            }
-        }
+        return powershell
+            .Invoke<PSModuleInfo>()
+            .DistinctBy(e => e.Name)
+            .Where(e => e.Name.StartsWith(wordToComplete, StringComparison.InvariantCultureIgnoreCase))
+            .OrderBy(e => e.Name)
+            .Select(e => new CompletionResult(e.Name));
     }
 }

src/PSParallelPipeline/PSTask.cs

@@ -114,13 +114,13 @@ private PSTask AddUsingStatements(Dictionary<string, object?> usingParams)
 
     private void CompleteTask()
     {
+        _powershell.Dispose();
         if (_canceled)
         {
             _runspace?.Dispose();
             return;
         }
 
-        _powershell.Dispose();
         if (_runspace is not null)
         {
             _pool.PushRunspace(_runspace);
@@ -129,7 +129,7 @@ private void CompleteTask()
 
     internal void Cancel()
     {
-        _powershell.Dispose();
+        _powershell.BeginStop(null, null);
         _canceled = true;
     }
 }

src/PSParallelPipeline/Worker.cs

@@ -8,7 +8,7 @@ namespace PSParallelPipeline;
 
 internal sealed class Worker
 {
-    private Task? _worker;
+    private readonly Task _worker;
 
     private readonly TaskSettings _taskSettings;
 
@@ -31,9 +31,10 @@ internal Worker(
         _taskSettings = taskSettings;
         _streams = new PSOutputStreams(_output);
         _pool = new RunspacePool(poolSettings, _streams, _token);
+        _worker = Task.Run(Start, cancellationToken: _token);
     }
 
-    internal void WaitForCompletion() => _worker?.GetAwaiter().GetResult();
+    internal void WaitForCompletion() => _worker.GetAwaiter().GetResult();
 
     internal void Enqueue(object? input) => _input.Add(input, _token);
 
@@ -43,8 +44,6 @@ internal Worker(
 
     internal IEnumerable<PSOutputData> GetConsumingEnumerable() => _output.GetConsumingEnumerable(_token);
 
-    internal void Run() => _worker = Task.Run(Start, cancellationToken: _token);
-
     private async Task Start()
     {
         List<Task> tasks = new(_pool.MaxRunspaces);
@@ -55,10 +54,7 @@ private async Task Start()
             {
                 if (tasks.Count == tasks.Capacity)
                 {
-                    Task task = await Task
-                        .WhenAny(tasks)
-                        .NoContext();
-
+                    Task task = await Task.WhenAny(tasks).NoContext();
                     tasks.Remove(task);
                     await task.NoContext();
                 }