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);
}
}