using System; using System.Collections.Concurrent; using System.Management.Automation; using System.Threading; using System.Threading.Tasks; namespace LANCommander.SDK.PowerShell; /// /// Base class for PowerShell cmdlets that need to execute async code. /// Based on PowerShell-OpenAuthenticode AsyncPSCmdlet. /// Override BeginProcessingAsync, ProcessRecordAsync, and/or EndProcessingAsync; WriteObject, WriteError, etc. /// can be called from async code and are marshalled to the pipeline thread. /// public abstract class AsyncCmdlet : PSCmdlet, IDisposable { private enum PipelineType { Output, OutputEnumerate, Error, Warning, Verbose, Debug, Information, Progress, ShouldProcess, } private readonly CancellationTokenSource _cancelSource = new(); private BlockingCollection<(object?, PipelineType)>? _currentOutPipe; private BlockingCollection? _currentReplyPipe; /// /// Gets the cancellation token for the current operation. Canceled when the cmdlet is stopped. /// protected CancellationToken CancellationToken => _cancelSource.Token; /// /// Override to perform async startup. Default implementation returns a completed task. /// protected override void BeginProcessing() { SessionState.PSVariable.Set("LANCommander.SDK.PSHostUI", Host.UI); RunBlockInAsync(BeginProcessingAsync); } /// /// Override to perform async startup. /// protected virtual Task BeginProcessingAsync() => Task.CompletedTask; /// /// Processes a single record by running ProcessRecordAsync and consuming pipeline output on the pipeline thread. /// protected override void ProcessRecord() => RunBlockInAsync(() => ProcessRecordAsync(CancellationToken)); /// /// Override to implement async record processing. /// protected abstract Task ProcessRecordAsync(CancellationToken cancellationToken); /// /// Override to perform async cleanup. Default implementation returns a completed task. /// protected override void EndProcessing() => RunBlockInAsync(EndProcessingAsync); /// /// Override to perform async cleanup. /// protected virtual Task EndProcessingAsync() => Task.CompletedTask; /// /// Called when the cmdlet is stopping. Cancels the cancellation token. /// protected override void StopProcessing() { _cancelSource.Cancel(); base.StopProcessing(); } private void RunBlockInAsync(Func task) { using var outPipe = new BlockingCollection<(object?, PipelineType)>(); using var replyPipe = new BlockingCollection(); var blockTask = Task.Run(async () => { try { _currentOutPipe = outPipe; _currentReplyPipe = replyPipe; await task(); } finally { _currentOutPipe = null; _currentReplyPipe = null; outPipe.CompleteAdding(); replyPipe.CompleteAdding(); } }); try { foreach (var (data, pipelineType) in outPipe.GetConsumingEnumerable(_cancelSource.Token)) { switch (pipelineType) { case PipelineType.Output: base.WriteObject(data); break; case PipelineType.OutputEnumerate: base.WriteObject(data, true); break; case PipelineType.Error: base.WriteError((ErrorRecord)data!); break; case PipelineType.Warning: base.WriteWarning((string)data!); break; case PipelineType.Verbose: base.WriteVerbose((string)data!); break; case PipelineType.Debug: base.WriteDebug((string)data!); break; case PipelineType.Information: base.WriteInformation((InformationRecord)data!); break; case PipelineType.Progress: base.WriteProgress((ProgressRecord)data!); break; case PipelineType.ShouldProcess: var (target, action) = (ValueTuple)data!; var res = base.ShouldProcess(target, action); replyPipe.Add(res, _cancelSource.Token); break; } } } catch (OperationCanceledException) { // Expected when StopProcessing cancels } try { blockTask.GetAwaiter().GetResult(); } catch (Exception ex) { base.WriteError(new ErrorRecord(ex, "ProcessRecordError", ErrorCategory.NotSpecified, null)); } } /// /// Writes an object to the pipeline. Safe to call from async code; marshalled to the pipeline thread. /// public new void WriteObject(object? sendToPipeline) => WriteObject(sendToPipeline, false); /// /// Writes an object to the pipeline. Safe to call from async code; marshalled to the pipeline thread. /// public new void WriteObject(object? sendToPipeline, bool enumerateCollection) { ThrowIfStopped(); _currentOutPipe?.Add((sendToPipeline, enumerateCollection ? PipelineType.OutputEnumerate : PipelineType.Output)); } /// /// Writes an error record. Safe to call from async code; marshalled to the pipeline thread. /// public new void WriteError(ErrorRecord errorRecord) { ThrowIfStopped(); _currentOutPipe?.Add((errorRecord, PipelineType.Error)); } /// /// Writes a warning. Safe to call from async code; marshalled to the pipeline thread. /// public new void WriteWarning(string message) { ThrowIfStopped(); _currentOutPipe?.Add((message, PipelineType.Warning)); } /// /// Writes verbose output. Safe to call from async code; marshalled to the pipeline thread. /// public new void WriteVerbose(string message) { ThrowIfStopped(); _currentOutPipe?.Add((message, PipelineType.Verbose)); } /// /// Writes debug output. Safe to call from async code; marshalled to the pipeline thread. /// public new void WriteDebug(string message) { ThrowIfStopped(); _currentOutPipe?.Add((message, PipelineType.Debug)); } /// /// Writes an information record. Safe to call from async code; marshalled to the pipeline thread. /// public new void WriteInformation(InformationRecord informationRecord) { ThrowIfStopped(); _currentOutPipe?.Add((informationRecord, PipelineType.Information)); } /// /// Writes a progress record. Safe to call from async code; marshalled to the pipeline thread. /// public new void WriteProgress(ProgressRecord progressRecord) { ThrowIfStopped(); _currentOutPipe?.Add((progressRecord, PipelineType.Progress)); } /// /// Confirms an operation with the user. Safe to call from async code; blocks until the pipeline thread returns the result. /// public new bool ShouldProcess(string target, string action) { ThrowIfStopped(); _currentOutPipe?.Add(((target, action), PipelineType.ShouldProcess)); return (bool)_currentReplyPipe?.Take(CancellationToken)!; } private void ThrowIfStopped() { if (_cancelSource.IsCancellationRequested) throw new PipelineStoppedException(); } /// /// Disposes the cancellation source. /// protected virtual void Dispose(bool disposing) { if (disposing) _cancelSource.Dispose(); } /// public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } }