44 invoke parallel should throw when passing a function not found

module/PSParallelPipeline.psd1

@@ -11,7 +11,7 @@
     RootModule        = 'bin/netstandard2.0/PSParallelPipeline.dll'
 
     # Version number of this module.
-    ModuleVersion     = '1.2.1'
+    ModuleVersion     = '1.2.2'
 
     # Supported PSEditions
     # CompatiblePSEditions = @()
@@ -77,7 +77,7 @@
     VariablesToExport = @()
 
     # Aliases to export from this module, for best performance, do not use wildcards and do not delete the entry, use an empty array if there are no aliases to export.
-    AliasesToExport   = @('parallel')
+    AliasesToExport   = @('parallel', 'asparallel')
 
     # DSC resources to export from this module
     # DscResourcesToExport = @()

src/PSParallelPipeline/Commands/InvokeParallelCommand.cs

@@ -2,17 +2,20 @@
 using System.Collections;
 using System.Management.Automation;
 using System.Management.Automation.Runspaces;
+using System.Threading;
 using PSParallelPipeline.Poly;
 
 namespace PSParallelPipeline.Commands;
 
 [Cmdlet(VerbsLifecycle.Invoke, "Parallel")]
-[Alias("parallel")]
+[Alias("parallel", "asparallel")]
 [OutputType(typeof(object))]
 public sealed class InvokeParallelCommand : PSCmdlet, IDisposable
 {
     private Worker? _worker;
 
+    private readonly CancellationTokenSource _cts = new();
+
     [Parameter(Position = 0, Mandatory = true)]
     public ScriptBlock ScriptBlock { get; set; } = null!;
 
@@ -46,6 +49,11 @@
 
     protected override void BeginProcessing()
     {
+        if (TimeoutSeconds > 0)
+        {
+            _cts.CancelAfter(TimeSpan.FromSeconds(TimeoutSeconds));
+        }
+
         InitialSessionState iss = InitialSessionState
             .CreateDefault2()
             .AddFunctions(Functions, this)
@@ -55,48 +63,42 @@
         {
             MaxRunspaces = ThrottleLimit,
             UseNewRunspace = UseNewRunspace,
-            InitialSessionState = iss,
-            UsingStatements = ScriptBlock.GetUsingParameters(this)
+            InitialSessionState = iss
         };
 
-        _worker = new Worker(poolSettings);
-
-        if (TimeoutSeconds > 0)
+        TaskSettings workerSettings = new()
         {
-            _worker.CancelAfter(TimeSpan.FromSeconds(TimeoutSeconds));
-        }
+            Script = ScriptBlock.ToString(),
+            UsingStatements = ScriptBlock.GetUsingParameters(this)
+        };
 
+        _worker = new Worker(poolSettings, workerSettings, _cts.Token);
         _worker.Run();
     }
 
     protected override void ProcessRecord()
     {
         Dbg.Assert(_worker is not null);
-        this.ThrowIfInputObjectIsScriptBlock(InputObject);
+        InputObject.ThrowIfInputObjectIsScriptBlock(this);
 
         try
         {
-            _worker.Enqueue(InputObject, ScriptBlock);
+            _worker.Enqueue(InputObject);
             while (_worker.TryTake(out PSOutputData data))
             {
                 ProcessOutput(data);
             }
         }
         catch (Exception _) when (_ is PipelineStoppedException or FlowControlException)
         {
-            _worker.Cancel();
-            _worker.Wait();
+            CancelAndWait();
             throw;
         }
         catch (OperationCanceledException exception)
         {
             _worker.Wait();
             exception.WriteTimeoutError(this);
         }
-        catch (Exception exception)
-        {
-            exception.WriteUnspecifiedError(this);
-        }
     }
 
     protected override void EndProcessing()
@@ -110,23 +112,19 @@
             {
                 ProcessOutput(data);
             }
+
             _worker.Wait();
         }
         catch (Exception _) when (_ is PipelineStoppedException or FlowControlException)
         {
-            _worker.Cancel();
-            _worker.Wait();
+            CancelAndWait();
             throw;
         }
         catch (OperationCanceledException exception)
         {
             _worker.Wait();
             exception.WriteTimeoutError(this);
         }
-        catch (Exception exception)
-        {
-            exception.WriteUnspecifiedError(this);
-        }
     }
 
     private void ProcessOutput(PSOutputData data)
@@ -166,11 +164,18 @@
         }
     }
 
-    protected override void StopProcessing() => _worker?.Cancel();
+    private void CancelAndWait()
+    {
+        _cts.Cancel();
+        _worker?.Wait();
+    }
+
+    protected override void StopProcessing() => CancelAndWait();
 
     public void Dispose()
     {
         _worker?.Dispose();
+        _cts.Dispose();
         GC.SuppressFinalize(this);
     }
 }

src/PSParallelPipeline/ExceptionHelpers.cs → src/PSParallelPipeline/ExceptionHelper.cs

@@ -3,7 +3,7 @@
 
 namespace PSParallelPipeline;
 
-internal static class ExceptionHelpers
+internal static class ExceptionHelper
 {
     private const string _notsupported =
         "Passed-in script block variables are not supported, and can result in undefined behavior.";
@@ -15,17 +15,33 @@ internal static void WriteTimeoutError(this Exception exception, PSCmdlet cmdlet
             ErrorCategory.OperationTimeout,
             cmdlet));
 
-    internal static void WriteUnspecifiedError(this Exception exception, PSCmdlet cmdlet) =>
-        cmdlet.WriteError(new ErrorRecord(
-            exception, "UnspecifiedCmdletError", ErrorCategory.NotSpecified, cmdlet));
-
     internal static PSOutputData CreateProcessingTaskError(this Exception exception, object context) =>
         PSOutputData.WriteError(new ErrorRecord(
             exception, "ProcessingTask", ErrorCategory.NotSpecified, context));
 
+    internal static void ThrowFunctionNotFoundError(
+        this CommandNotFoundException exception,
+        Cmdlet cmdlet,
+        string function) =>
+        cmdlet.ThrowTerminatingError(new ErrorRecord(
+            exception, "FunctionNotFound", ErrorCategory.ObjectNotFound, function));
+
     private static bool ValueIsNotScriptBlock(object? value) =>
         value is not ScriptBlock and not PSObject { BaseObject: ScriptBlock };
 
+    internal static CommandInfo ThrowIfFunctionNotFoundError(
+        this CommandInfo? command,
+        string function)
+    {
+        if (command is not null)
+        {
+            return command;
+        }
+
+        throw new CommandNotFoundException(
+            $"The function with name '{function}' could not be found.");
+    }
+
     internal static void ThrowIfVariableIsScriptBlock(this PSCmdlet cmdlet, object? value)
     {
         if (ValueIsNotScriptBlock(value))
@@ -40,7 +56,7 @@ internal static void ThrowIfVariableIsScriptBlock(this PSCmdlet cmdlet, object?
             value));
     }
 
-    internal static void ThrowIfInputObjectIsScriptBlock(this PSCmdlet cmdlet, object? value)
+    internal static void ThrowIfInputObjectIsScriptBlock(this object? value, PSCmdlet cmdlet)
     {
         if (ValueIsNotScriptBlock(value))
         {

src/PSParallelPipeline/Extensions.cs

@@ -20,18 +20,22 @@ internal static InitialSessionState AddFunctions(
         {
             foreach (string function in functionsToAdd)
             {
-                CommandInfo? commandInfo = cmdlet
-                    .InvokeCommand
-                    .GetCommand(function, CommandTypes.Function);
-
-                if (commandInfo is null)
+                try
                 {
-                    continue;
+                    CommandInfo commandInfo = cmdlet
+                        .InvokeCommand
+                        .GetCommand(function, CommandTypes.Function)
+                        .ThrowIfFunctionNotFoundError(function);
+
+                    initialSessionState.Commands.Add(
+                        new SessionStateFunctionEntry(
+                            name: function,
+                            definition: commandInfo.Definition));
+                }
+                catch (CommandNotFoundException exception)
+                {
+                    exception.ThrowFunctionNotFoundError(cmdlet, function);
                 }
-
-                initialSessionState.Commands.Add(new SessionStateFunctionEntry(
-                    name: function,
-                    definition: commandInfo.Definition));
             }
         }
 

src/PSParallelPipeline/PSOutputStreams.cs

@@ -1,13 +1,12 @@
 using System;
 using System.Collections.Concurrent;
 using System.Management.Automation;
-using System.Threading;
 
 namespace PSParallelPipeline;
 
 internal sealed class PSOutputStreams : IDisposable
 {
-    private BlockingCollection<PSOutputData> OutputPipe { get => _worker.OutputPipe; }
+    private BlockingCollection<PSOutputData> Output { get; }
 
     internal PSDataCollection<PSObject> Success { get; } = [];
 
@@ -23,17 +22,15 @@ internal sealed class PSOutputStreams : IDisposable
 
     internal PSDataCollection<WarningRecord> Warning { get; } = [];
 
-    private readonly Worker _worker;
-
-    internal PSOutputStreams(Worker worker)
+    internal PSOutputStreams(BlockingCollection<PSOutputData> output)
     {
-        _worker = worker;
-        SetStreamHandlers();
+        Output = output;
+        SetHandlers();
     }
 
-    internal void AddOutput(PSOutputData data) => OutputPipe.Add(data);
+    internal void AddOutput(PSOutputData data) => Output.Add(data);
 
-    private void SetStreamHandlers()
+    private void SetHandlers()
     {
         Success.DataAdding += (s, e) =>
             AddOutput(PSOutputData.WriteObject(e.ItemAdded));
@@ -66,7 +63,6 @@ public void Dispose()
         Progress.Dispose();
         Verbose.Dispose();
         Warning.Dispose();
-        OutputPipe.Dispose();
         GC.SuppressFinalize(this);
     }
 }