/********************************************************************++
Copyright (c) Microsoft Corporation. All rights reserved.
--********************************************************************/
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.ComponentModel;
using System.Linq;
using System.Management.Automation.Internal;
using System.Management.Automation.Remoting;
using System.Management.Automation.Runspaces;
using System.Management.Automation.Tracing;
using System.Threading;
using Dbg = System.Management.Automation.Diagnostics;
// Stops compiler from warning about unknown warnings
#pragma warning disable 1634, 1691
namespace System.Management.Automation
{
#region Workflow Hosting API
///
/// Class that will serve as the API for hosting and executing
/// workflows in PowerShell. This class will have a behavior
/// similar to how the Runspace and PowerShell APIs behave in
/// the remoting scenario. The objects on the client side act
/// as proxies to the real objects on the server
///
public sealed class PSJobProxy : Job2
{
#region Constructors
///
/// Internal constructor
///
///
/// Success status of the command execution.
///
public override string StatusMessage
{
get { return _remoteJobStatusMessage; }
}
///
/// Indicates that more data is available in this
/// result object for reading.
///
public override bool HasMoreData
{
get
{
// moreData is initially set to true, and it
// will remain so until the async result
// object has completed execution.
if (_moreData && IsFinishedState(JobStateInfo.State))
{
bool atleastOneChildHasMoreData = ChildJobs.Any(t => t.HasMoreData);
bool parentHasMoreData = (CollectionHasMoreData(Output)
|| CollectionHasMoreData(Error)
|| CollectionHasMoreData(Verbose)
|| CollectionHasMoreData(Debug)
|| CollectionHasMoreData(Warning)
|| CollectionHasMoreData(Progress));
_moreData = parentHasMoreData || atleastOneChildHasMoreData;
}
return _moreData;
}
}
///
/// This is the location string from the remote job.
///
public override string Location
{
get { return _remoteJobLocation; }
}
#endregion
#region Overrides of Job2
///
/// start a job. The job will be started with the parameters
/// specified in StartParameters
///
/// Thrown if the job
/// is already running, if there is no runspace or runspace pool
/// assigned.
/// Thrown if the job is
/// otherwise started, finished, or suspended.
/// Thrown if job is
/// disposed.
public override void StartJob()
{
StartJob(null, null, null);
}
///
/// Start a job asynchronously
///
/// When a job is started all the data in the
/// job streams from a previous invocation will be cleared
public override void StartJobAsync()
{
StartJobAsync(null, null, null);
}
///
/// Stop a job synchronously. In order to be consistent, this method
/// should be used in place of StopJob which was introduced in the
/// v2 Job API
///
/// Thrown if job is blocked.
/// Thrown if job is disposed.
public override void StopJob()
{
try
{
// The call is asynchronous because Receive-Job is called
// in the same invocation. Return once the job state changed
// event is received, and the job state is updated to a
// Finished state.
if (ShouldQueueOperation())
_pendingOperations.Enqueue(QueueOperation.Stop);
else
DoStopAsync();
Finished.WaitOne();
}
catch (Exception error)
{
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "StopJob", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
throw;
}
}
///
/// Stop a job asynchronously
///
public override void StopJobAsync()
{
#pragma warning disable 56500
try
{
if (ShouldQueueOperation())
_pendingOperations.Enqueue(QueueOperation.Stop);
else
DoStopAsync();
}
catch (Exception error)
{
// Exception transferred using event arguments.
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "StopJobAsync", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
OnStopJobCompleted(new AsyncCompletedEventArgs(error, false, null));
}
#pragma warning restore 56500
}
///
/// StopJob
///
///
/// StopJobAsync
///
///
/// Suspend a job
///
/// Throws if the job is not in
/// a running or suspended state.
/// Thrown if job is
/// disposed.
public override void SuspendJob()
{
try
{
// The call is asynchronous because Receive-Job is called
// in the same invocation. Return once the job state changed
// event is received, and the job state is updated to a
// Finished or Suspended state.
if (ShouldQueueOperation())
_pendingOperations.Enqueue(QueueOperation.Suspend);
else
DoSuspendAsync();
JobSuspendedOrFinished.WaitOne();
}
catch (Exception error)
{
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "SuspendJob", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
throw;
}
}
///
/// Asynchronously suspend a job
///
public override void SuspendJobAsync()
{
#pragma warning disable 56500
try
{
if (ShouldQueueOperation())
_pendingOperations.Enqueue(QueueOperation.Suspend);
else
DoSuspendAsync();
}
catch (Exception error)
{
// Exception transferred using event arguments.
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "SuspendJobAsync", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
OnSuspendJobCompleted(new AsyncCompletedEventArgs(error, false, null));
}
#pragma warning restore 56500
}
///
/// SuspendJob
///
///
/// SuspendJobAsync
///
///
/// Resume a suspended job
///
/// Throws if the job
/// is not in a suspended or running state.
/// Thrown if job is
/// disposed.
public override void ResumeJob()
{
try
{
// The call is asynchronous because Receive-Job is called
// in the same invocation. Return once the job state changed
// event is received, and the job state is updated to a
// Finished or Running state.
if (ShouldQueueOperation())
_pendingOperations.Enqueue(QueueOperation.Resume);
else
DoResumeAsync();
JobRunningOrFinished.WaitOne();
}
catch (Exception error)
{
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "ResumeJob", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
throw;
}
}
///
/// Resume a suspended job asynchronously.
///
public override void ResumeJobAsync()
{
#pragma warning disable 56500
try
{
if (ShouldQueueOperation())
_pendingOperations.Enqueue(QueueOperation.Resume);
else
DoResumeAsync();
}
catch (Exception error)
{
// Exception transferred using event arguments.
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "ResumeJobAsync", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
OnResumeJobCompleted(new AsyncCompletedEventArgs(error, false, null));
}
#pragma warning restore 56500
}
///
/// Unblock a blocked job
///
/// Unblock job is not supported on PSJobProxy.
public override void UnblockJob()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyUnblockJobNotSupported);
}
///
/// Unblock a blocked job asynchronously
///
public override void UnblockJobAsync()
{
OnUnblockJobCompleted(new AsyncCompletedEventArgs(PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyUnblockJobNotSupported), false, null));
}
#endregion
#region Additional State Management Methods
///
/// Start execution of the workflow with the
/// specified input. This input will serve as
/// input to the underlying pipeline
///
/// input)
{
StartJobAsync(null, null, input);
}
///
/// Start execution of the job with the
/// specified input. This input will serve as
/// input to the underlying pipeline
///
/// Not sure if this method is needed. This has
/// been added just to be in sync with the PowerShell
/// APIs
public void StartJob(PSDataCollection input)
{
StartJob(null, null, input);
}
///
/// Start execution of the workflow with the
/// specified input. This input will serve as
/// input to the underlying pipeline.
/// Because the number of child jobs is unknown before starting
/// the job, delegates may be indicated to ensure that no events will be missed
/// after the child job is created if data begins streaming back immediately.
///
/// dataAdded, EventHandler stateChanged, PSDataCollection input)
{
try
{
// The call is asynchronous because Receive-Job is called
// in the same invocation. Return once the job object is returned and
// used to initialize the proxy.
DoStartAsync(dataAdded, stateChanged, input);
_jobInitialiedWaitHandle.WaitOne();
}
catch (Exception error)
{
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "StartJob", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
throw;
}
}
///
/// Start asynchronous execution of the workflow with the
/// specified input. This input will serve as
/// input to the underlying pipeline.
/// Because the number of child jobs is unknown before starting
/// the job, delegates may be indicated to ensure that no events will be missed
/// after the child job is created if data begins streaming back immediately.
///
/// dataAdded, EventHandler stateChanged, PSDataCollection input)
{
#pragma warning disable 56500
try
{
DoStartAsync(dataAdded, stateChanged, input);
}
catch (Exception error)
{
// Exception transferred using event arguments.
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "StartJobAsync", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
OnStartJobCompleted(new AsyncCompletedEventArgs(error, false, null));
}
#pragma warning restore 56500
}
///
/// Removes the job.
/// If remoteJob is true, the job output that has been transferred to this
/// client object will be preserved.
///
/// Thrown if the job is not in
/// a completed state.
public void RemoveJob(bool removeRemoteJob, bool force)
{
if (!removeRemoteJob)
{
Dispose();
return;
}
lock (SyncRoot)
{
AssertNotDisposed();
}
try
{
DoRemove(force);
lock (SyncRoot)
{
// If _removeCalled has not been set to true, do not wait.
// There should be an exception during DoRemove in most cases when
// this is true.
if (!_removeCalled)
{
Diagnostics.Assert(false, "remove called is false after calling remove. No exception was thrown.");
return;
}
}
RemoveComplete.WaitOne();
}
catch (Exception error)
{
CommandProcessorBase.CheckForSevereException(error);
_tracer.WriteMessage(ClassNameTrace, "RemoveJob", _remoteJobInstanceId, this, "Error", null);
_tracer.TraceException(error);
throw;
}
}
///
/// Removes the job.
/// If remoteJob is true, the job output that has been transferred to this
/// client object will be preserved.
///
/// Thrown if the job is not in
/// a completed state.
public void RemoveJob(bool removeRemoteJob)
{
RemoveJob(removeRemoteJob, false);
}
///
/// Removes the job on the remote server.
/// The job output that has been transferred to this client object will be
/// preserved.
///
///
/// Removes the job on the remote server.
/// The job output that has been transferred to this client object will be
/// preserved.
///
///
/// This event should be raised whenever the asynchronous removal of
/// a server side job is completed.
///
public event EventHandler RemoveJobCompleted;
///
/// Method to raise the event when removing a
/// server side job is completed
///
/// handler = RemoveJobCompleted;
#pragma warning disable 56500
try
{
if (handler != null)
{
handler(this, eventArgs);
}
}
catch (Exception exception)
{
// errors in the handlers are not errors in the operation
// silently ignore them
CommandProcessorBase.CheckForSevereException(exception);
_tracer.TraceException(exception);
}
#pragma warning restore 56500
}
#endregion Additional State Management Methods
#region Additional Properties
///
/// If set, the remote job will be removed when it has been completed and the data has been received.
/// This can only be set prior to a job being started.
///
public bool RemoveRemoteJobOnCompletion
{
get
{
return _removeRemoteJobOnCompletion;
}
set
{
AssertChangesCanBeAccepted();
_removeRemoteJobOnCompletion = value;
}
}
private bool _removeRemoteJobOnCompletion;
///
/// The instance ID of the remote job that this proxy interacts with.
///
public Guid RemoteJobInstanceId
{
get { return _remoteJobInstanceId; }
}
///
/// Runspace in which this job will be executed
///
/// At any point of time only a runspace or a
/// runspacepool may be specified
public Runspace Runspace
{
get
{
return _runspace;
}
set
{
if (value == null)
{
throw PSTraceSource.NewArgumentNullException("value");
}
lock (SyncRoot)
{
AssertChangesCanBeAccepted();
_runspacePool = null;
_runspace = value;
}
}
}
///
/// RunspacePool in which this job will be executed
///
public RunspacePool RunspacePool
{
get
{
return _runspacePool;
}
set
{
if (value == null)
{
throw PSTraceSource.NewArgumentNullException("value");
}
lock (SyncRoot)
{
AssertChangesCanBeAccepted();
_runspace = null;
_runspacePool = value;
}
}
}
#endregion Additional Properties
#region Proxy Factory Static Methods
///
/// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
///
/// A collection of job proxies that represent the jobs collected based on the filter.
public static ICollection Create(Runspace runspace, Hashtable filter,
EventHandler dataAdded, EventHandler stateChanged)
{
return Create(runspace, filter, dataAdded, stateChanged, true);
}
///
/// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
///
/// A collection of job proxies that represent the jobs collected based on the filter.
public static ICollection Create(Runspace runspace, Hashtable filter, bool receiveImmediately)
{
return Create(runspace, filter, null, null, receiveImmediately);
}
///
/// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
///
/// A collection of job proxies that represent the jobs collected based on the filter.
public static ICollection Create(Runspace runspace, Hashtable filter)
{
return Create(runspace, filter, null, null, true);
}
///
/// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
///
/// A collection of job proxies that represent the jobs collected based on the filter.
public static ICollection Create(Runspace runspace)
{
return Create(runspace, null, null, null, true);
}
///
/// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
///
/// A collection of job proxies that represent the jobs collected based on the filter.
public static ICollection Create(RunspacePool runspacePool, Hashtable filter,
EventHandler dataAdded, EventHandler stateChanged)
{
return Create(runspacePool, filter, dataAdded, stateChanged, true);
}
///
/// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
///
/// A collection of job proxies that represent the jobs collected based on the filter.
public static ICollection Create(RunspacePool runspacePool, Hashtable filter, bool receiveImmediately)
{
return Create(runspacePool, filter, null, null, receiveImmediately);
}
///
/// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
///
/// A collection of job proxies that represent the jobs collected based on the filter.
public static ICollection Create(RunspacePool runspacePool, Hashtable filter)
{
return Create(runspacePool, filter, null, null, true);
}
///
/// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
///
/// A collection of job proxies that represent the jobs collected based on the filter.
public static ICollection Create(RunspacePool runspacePool)
{
return Create(runspacePool, null, null, null, true);
}
private static ICollection Create(RunspacePool runspacePool, Hashtable filter,
EventHandler dataAdded, EventHandler stateChanged, bool connectImmediately)
{
if (runspacePool == null)
throw new PSArgumentNullException("runspacePool");
return Create(null, runspacePool, filter, dataAdded, stateChanged, connectImmediately);
}
private static ICollection Create(Runspace runspace, Hashtable filter,
EventHandler dataAdded, EventHandler stateChanged, bool connectImmediately)
{
if (runspace == null)
throw new PSArgumentNullException("runspace");
return Create(runspace, null, filter, dataAdded, stateChanged, connectImmediately);
}
private static ICollection Create(Runspace runspace, RunspacePool runspacePool, Hashtable filter,
EventHandler dataAdded, EventHandler stateChanged, bool connectImmediately)
{
// run command
Collection jobResults;
using (PowerShell ps = PowerShell.Create())
{
Dbg.Assert(runspacePool == null && runspace != null ||
runspace == null && runspacePool != null, "Either a runspace or a runspacepool should be used, not both.");
if (runspacePool == null)
{
ps.Runspace = runspace;
}
else
{
ps.RunspacePool = runspacePool;
}
ps.AddCommand("Get-Job");
if (filter != null) ps.AddParameter("Filter", filter);
jobResults = ps.Invoke();
}
Collection jobs = new Collection();
foreach (var deserializedJob in jobResults)
{
if (!Deserializer.IsDeserializedInstanceOfType(deserializedJob, typeof(Job)))
{
// Do not create proxies if jobs are live.
continue;
}
string command = String.Empty;
if (!TryGetJobPropertyValue(deserializedJob, "Command", out command))
{
Dbg.Assert(false, "Job object did not contain command when creating proxy.");
}
PSJobProxy job = new PSJobProxy(command);
job.InitializeExistingJobProxy(deserializedJob, runspace, runspacePool);
// Events will be registered and states will be set by ReceiveJob when it is called.
// This may be later, if connectImmediately is set to false.
job._receiveIsValidCall = true;
if (connectImmediately)
{
// This will make receive invalid in the future, sets the state to running.
job.ReceiveJob(dataAdded, stateChanged);
}
else
{
Dbg.Assert(dataAdded == null, "DataAdded cannot be specified if not connecting immediately");
Dbg.Assert(stateChanged == null, "StateChanged cannot be specified if not connecting immediately");
}
jobs.Add(job);
}
return jobs;
}
#endregion Proxy Factory Static Methods
///
/// Will begin streaming data for a job object created by the "create" method that is in a not started state.
///
public void ReceiveJob()
{
ReceiveJob(null, null);
}
///
/// Will begin streaming data for a job object created by the "create" method that is in a not started state.
///
/// dataAdded, EventHandler stateChanged)
{
lock (SyncRoot)
{
if (!_receiveIsValidCall)
{
// Receive may only be called once, and is only a valid call after a proxy job has been created in
// a non-streaming state.
throw PSTraceSource.NewInvalidOperationException(PowerShellStrings.JobProxyReceiveInvalid);
}
Dbg.Assert(JobStateInfo.State == JobState.NotStarted, "ReceiveJob is only valid on a not started PSJobProxy");
Dbg.Assert(_receivePowerShell.Commands.Commands.Count == 0, "ReceiveJob is only valid if internal pipeline is unused.");
Dbg.Assert(_receivePowerShell.InvocationStateInfo.State == PSInvocationState.NotStarted, "ReceiveJob is only valid if internal pipeline has not been started");
Dbg.Assert(_dataAddedHandler == null, "DataAdded cannot be specified before calling ReceiveJob");
Dbg.Assert(_stateChangedHandler == null, "StateChanged cannot be specified before calling ReceiveJob");
_receiveIsValidCall = false;
_dataAddedHandler = dataAdded;
_stateChangedHandler = stateChanged;
}
RegisterChildEvents();
// Set job states after events have been registered.
// All jobs begin in the running state. States will be updated when state changed
// event information is gotten by Receive-Job. See DataAddedToOutput.
ValidateAndDoSetJobState(JobState.Running);
foreach (PSChildJobProxy child in ChildJobs)
{
Dbg.Assert(child != null, "Child should always be PSChildJobProxy");
child.DoSetJobState(JobState.Running);
}
AssignRunspaceOrRunspacePool(_receivePowerShell);
AddReceiveJobCommandToPowerShell(_receivePowerShell, false);
_receivePowerShell.AddParameter("InstanceId", _remoteJobInstanceId);
_receivePowerShell.BeginInvoke((PSDataCollection)null, _receivePowerShellOutput, null,
CleanupReceivePowerShell, null);
}
#region Internal Methods
internal void InitializeJobProxy(PSCommand command, Runspace runspace, RunspacePool runspacePool)
{
Dbg.Assert(command != null, "Command cannot be null");
_tracer.WriteMessage(ClassNameTrace, "InitializeJobProxy", _remoteJobInstanceId, this,
"Initializing Job Proxy.", null);
_pscommand = command.Clone();
var paramCol = new CommandParameterCollection();
foreach (var parameter in _pscommand.Commands[0].Parameters)
{
paramCol.Add(parameter);
}
// The StartParameters may be edited or overwritten by the user.
// The parameters will need to be added back to the command object before
// execution.
StartParameters = new List { paramCol };
_pscommand.Commands[0].Parameters.Clear();
CommonInit(runspace, runspacePool);
}
internal void InitializeExistingJobProxy(PSObject o, Runspace runspace, RunspacePool runspacePool)
{
Dbg.Assert(o != null, "deserialized job cannot be null");
_tracer.WriteMessage(ClassNameTrace, "InitializeExistingJobProxy", _remoteJobInstanceId, this,
"Initializing job proxy for existing job.", null);
_pscommand = null;
_startCalled = true;
_jobInitialized = true;
CommonInit(runspace, runspacePool);
PopulateJobProperties(o);
// Do not set child job states, or subscribe to child events. This will
// be done in ReceiveJob whenever it is called.
// We need to wait until then because event handlers may not be specified until then.
// Construct the parent job's start parameters specific for SM+
List psParamCollection = new List();
object psPrivateMetadata = null;
foreach (PSChildJobProxy job in ChildJobs)
{
// Each child job should have StartParameters.Count = 1.
if (job.StartParameters.Count == 0) continue;
Hashtable childJobCol = new Hashtable();
foreach (CommandParameter p in job.StartParameters[0])
{
if (psPrivateMetadata == null && p.Name.Equals("PSPrivateMetadata", StringComparison.OrdinalIgnoreCase))
{
psPrivateMetadata = p.Value;
}
childJobCol.Add(p.Name, p.Value);
}
psParamCollection.Add(childJobCol);
}
CommandParameterCollection newStartParameters = new CommandParameterCollection();
newStartParameters.Add(new CommandParameter("PSParameterCollection", psParamCollection));
if (psPrivateMetadata != null)
{
newStartParameters.Add(new CommandParameter("PSPrivateMetadata", psPrivateMetadata));
}
StartParameters.Add(newStartParameters);
}
private void CommonInit(Runspace runspace, RunspacePool runspacePool)
{
_runspacePool = runspacePool;
_runspace = runspace;
_receivePowerShell.InvocationStateChanged += ReceivePowerShellInvocationStateChanged;
// Output buffer other than the PSJobProxy output is needed.
// If the proxy's output buffer is used, data added there will raise events
// to the consumer before we have re-distributed the output.
_receivePowerShellOutput.DataAdded += DataAddedToOutput;
_receivePowerShell.Streams.Error.DataAdded += DataAddedToError;
_receivePowerShell.Streams.Debug.DataAdded += DataAddedToDebug;
_receivePowerShell.Streams.Verbose.DataAdded += DataAddedToVerbose;
_receivePowerShell.Streams.Warning.DataAdded += DataAddedToWarning;
_receivePowerShell.Streams.Progress.DataAdded += DataAddedToProgress;
_receivePowerShell.Streams.Information.DataAdded += DataAddedToInformation;
}
///
/// Helper to do error checking for getting a property of type T from a PSobject.
///
/// (PSObject o, string propertyName, out T propertyValue)
{
propertyValue = default(T);
PSPropertyInfo propertyInfo = o.Properties[propertyName];
if (propertyInfo == null || !(propertyInfo.Value is T)) return false;
propertyValue = (T)propertyInfo.Value;
return true;
}
#endregion Internal Methods
#region Async helpers
private enum ActionType
{
Remove
}
private class AsyncCompleteContainer
{
internal AsyncCompletedEventArgs EventArgs;
internal ActionType Action;
}
private delegate void JobActionWorkerDelegate(AsyncOperation asyncOp, ActionType action);
private void JobActionWorker(AsyncOperation asyncOp, ActionType action)
{
Exception exception = null;
#pragma warning disable 56500
try
{
switch (action)
{
case ActionType.Remove:
DoRemove(asyncOp.UserSuppliedState);
break;
}
}
catch (Exception e)
{
// Called on a background thread, need to include any exception in
// event arguments.
exception = e;
}
#pragma warning restore 56500
var eventArgs = new AsyncCompletedEventArgs(exception, false, asyncOp.UserSuppliedState);
var container = new AsyncCompleteContainer { EventArgs = eventArgs, Action = action };
// End the task. The asyncOp object is responsible
// for marshaling the call.
asyncOp.PostOperationCompleted(JobActionAsyncCompleted, container);
}
private void JobActionAsyncCompleted(object operationState)
{
var container = operationState as AsyncCompleteContainer;
Dbg.Assert(container != null, "AsyncComplete Container is null; not passed properly.");
_tracer.WriteMessage(ClassNameTrace, "JobActionAsyncCompleted", _remoteJobInstanceId, this,
"Async operation {0} completed", container.Action.ToString());
switch (container.Action)
{
case ActionType.Remove:
if (container.EventArgs.Error == null) RemoveComplete.WaitOne();
OnRemoveJobCompleted(container.EventArgs);
break;
}
}
#endregion Async helpers
#region Private Methods
private bool ShouldQueueOperation()
{
bool queueThisOperation;
lock (_inProgressSyncObject)
{
if (!_inProgress)
{
_inProgress = true;
queueThisOperation = false;
}
else
{
queueThisOperation = true;
}
}
return queueThisOperation;
}
private void ProcessQueue()
{
bool queueWorkItem = false;
lock (_inProgressSyncObject)
{
if (!_pendingOperations.IsEmpty && !_workerCreated)
{
queueWorkItem = true;
_workerCreated = true;
_inProgress = true;
}
else
_inProgress = false;
}
if (queueWorkItem)
{
ThreadPool.QueueUserWorkItem(ProcessQueueWorker);
}
}
private void ProcessQueueWorker(object state)
{
// If this operation has been queued to the thread pool, processing rights
// already belong here. Release only when finished.
while (true)
{
QueueOperation nextOperation;
lock (_inProgressSyncObject)
{
if (!_pendingOperations.TryDequeue(out nextOperation))
{
_inProgress = false;
_workerCreated = false;
break;
}
}
switch (nextOperation)
{
case QueueOperation.Stop:
#pragma warning disable 56500
try
{
DoStopAsync();
Finished.WaitOne();
}
catch (Exception e)
{
// Trasfer exception via event arguments.
OnStopJobCompleted(new AsyncCompletedEventArgs(e, false, null));
}
break;
case QueueOperation.Suspend:
try
{
DoSuspendAsync();
JobSuspendedOrFinished.WaitOne();
}
catch (Exception e)
{
// Transfer exception via event arguments.
OnSuspendJobCompleted(new AsyncCompletedEventArgs(e, false, null));
}
break;
case QueueOperation.Resume:
try
{
DoResumeAsync();
JobRunningOrFinished.WaitOne();
}
catch (Exception e)
{
// Transfer exception via event arguments.
OnResumeJobCompleted(new AsyncCompletedEventArgs(e, false, null));
}
#pragma warning restore 56500
break;
}
}
}
///
/// Checks if there is more data in the specified collection
///
/// Type of the collection
/// true if the collection has more data
private static bool CollectionHasMoreData(PSDataCollection collection)
{
return (collection.IsOpen || collection.Count > 0);
}
///
/// Worker method which starts the job
///
private void DoStartAsync(EventHandler dataAdded, EventHandler stateChanged, PSDataCollection input)
{
// Checks disposed, and whether the start operation is valid.
AssertJobCanBeStartedAndSetStateToRunning();
Dbg.Assert(_inProgress == false, "Start should always be able to obtain processing rights if it is a valid call.");
lock (_inProgressSyncObject) { _inProgress = true; }
// Set input and handlers. Should only be done after we've confirmed that this
// call to StartJob should proceed, which isn't until now.
lock (SyncRoot)
{
Dbg.Assert(_dataAddedHandler == null, "DataAdded cannot be specified before StartJob is called");
Dbg.Assert(_stateChangedHandler == null, "StateChanged cannot be specified before StartJob is called");
_dataAddedHandler = dataAdded;
_stateChangedHandler = stateChanged;
}
_tracer.WriteMessage(ClassNameTrace, "DoStartAsync", _remoteJobInstanceId, this,
"Starting command invocation.", null);
s_structuredTracer.BeginProxyJobExecution(InstanceId);
DoStartPrepare();
_receivePowerShell.BeginInvoke(input, _receivePowerShellOutput, null, CleanupReceivePowerShell, null);
}
private void DoStartPrepare()
{
// Either a runspace or a runspacepool has to be set
// before the job can be started
if (_runspacePool == null && _runspace == null)
{
throw PSTraceSource.NewInvalidOperationException(PowerShellStrings.RunspaceAndRunspacePoolNull);
}
AssignRunspaceOrRunspacePool(_receivePowerShell);
// set the parameters. the user may have changed them.
bool found = false;
if (StartParameters != null && StartParameters.Count > 0)
{
foreach (var parameter in StartParameters[0])
{
_pscommand.Commands[0].Parameters.Add(parameter);
if (string.Compare(parameter.Name, "AsJob", StringComparison.OrdinalIgnoreCase) == 0)
{
if (!(parameter.Value is bool) || !(bool)parameter.Value)
{
// If the AsJob Parameter has been passed and explicitly set to false, we should
// not proceed with the operation. This is an error.
throw PSTraceSource.NewInvalidOperationException(PowerShellStrings.JobProxyAsJobMustBeTrue);
}
found = true;
}
}
}
// If the user has overwritten the parameter collection, we should add AsJob back again.
if (!found)
{
_pscommand.Commands[0].Parameters.Add("AsJob", true);
}
// set the commands for the powershell
_receivePowerShell.Commands = _pscommand;
// Setup receive-job data streaming.
// Pipe the invocation with as job to Receive-Job, because we do not yet have the job's
// instance ID.
// Use the WriteJob parameter to cause Receive-Job to output the job object.
AddReceiveJobCommandToPowerShell(_receivePowerShell, true);
}
///
/// Worker method which stops the job
///
private void DoStopAsync()
{
if (!AssertStopJobIsValidAndSetToStopping())
{
OnStopJobCompleted(new AsyncCompletedEventArgs(null, false, null));
return;
}
_receivePowerShell.Stop();
_receivePowerShell.Commands.Clear();
_receivePowerShell.GenerateNewInstanceId();
_receivePowerShell.AddCommand("Stop-Job").AddParameter("InstanceId", _remoteJobInstanceId).AddParameter("PassThru");
AddReceiveJobCommandToPowerShell(_receivePowerShell, false);
_receivePowerShell.BeginInvoke((PSDataCollection)null, _receivePowerShellOutput, null, CleanupReceivePowerShell, null);
}
///
/// Worker method which suspends the job
///
private void DoSuspendAsync()
{
if (!AssertSuspendJobIsValidAndSetToSuspending())
{
OnSuspendJobCompleted(new AsyncCompletedEventArgs(null, false, null));
return;
}
_receivePowerShell.Stop();
_receivePowerShell.Commands.Clear();
_receivePowerShell.GenerateNewInstanceId();
_receivePowerShell.AddCommand("Suspend-Job").AddParameter("InstanceId", _remoteJobInstanceId).AddParameter("Wait");
AddReceiveJobCommandToPowerShell(_receivePowerShell, false);
_receivePowerShell.BeginInvoke((PSDataCollection)null, _receivePowerShellOutput, null, CleanupReceivePowerShell, null);
}
///
/// Worker method to resume the job
///
private void DoResumeAsync()
{
AssertResumeJobIsValidAndSetToRunning();
// If the job state was not set to running, the job should not be resumed. (True if job is in finished state.)
if (JobStateInfo.State != JobState.Running)
{
OnResumeJobCompleted(new AsyncCompletedEventArgs(null, false, null));
return;
}
_receivePowerShell.Stop();
_receivePowerShell.Commands.Clear();
_receivePowerShell.GenerateNewInstanceId();
_receivePowerShell.AddCommand("Resume-Job").AddParameter("InstanceId", _remoteJobInstanceId).AddParameter("Wait");
AddReceiveJobCommandToPowerShell(_receivePowerShell, false);
_receivePowerShell.BeginInvoke((PSDataCollection)null, _receivePowerShellOutput, null, CleanupReceivePowerShell, null);
}
///
/// Worker method to remove the remote job object
///
///
/// Assigns either a runspace or runspacepool to the specified powershell
/// instance
///
///
/// Event handler for InvocationStateChanged on the powershell
/// object running receive-job
///
///
/// Assigns job properties and creates child job tree.
///
/// ().Where(job => !(job.BaseObject is string)))
{
Guid childJobInstanceId;
if (!TryGetJobPropertyValue(job, "InstanceId", out childJobInstanceId))
{
Dbg.Assert(false, "ChildJobs should be serialized to include InstanceID, cannot interact with them otherwise.");
continue;
}
var childProxyJob = new PSChildJobProxy(Command, job); // All have the same workflow name.
_childJobsMapping.Add(childJobInstanceId, childProxyJob);
// Set DataAddedCount for each child to match parent.
// This may otherwise not be available to set until data arrives.
childProxyJob.Output.DataAddedCount = Output.DataAddedCount;
childProxyJob.Error.DataAddedCount = Error.DataAddedCount;
childProxyJob.Progress.DataAddedCount = Progress.DataAddedCount;
childProxyJob.Warning.DataAddedCount = Warning.DataAddedCount;
childProxyJob.Verbose.DataAddedCount = Verbose.DataAddedCount;
childProxyJob.Debug.DataAddedCount = Debug.DataAddedCount;
childProxyJob.Information.DataAddedCount = Information.DataAddedCount;
// Now see if there are start parameters we can populate.
PSObject childJobStartParametersObject;
if (TryGetJobPropertyValue(job, "StartParameters", out childJobStartParametersObject))
{
PopulateStartParametersOnChild(childJobStartParametersObject, childProxyJob);
}
ChildJobs.Add(childProxyJob);
}
}
private void PopulateStartParametersOnChild(PSObject childJobStartParametersObject, PSChildJobProxy childProxyJob)
{
ArrayList childJobStartParameters = childJobStartParametersObject.BaseObject as ArrayList;
if (childJobStartParameters != null)
{
List listComParCol = new List();
// check childjobstartparameters--was a List
foreach (PSObject paramCollection in childJobStartParameters.Cast().Where(paramCollection => !(paramCollection.BaseObject is string)))
{
ArrayList parameterCollection = paramCollection.BaseObject as ArrayList;
if (parameterCollection != null)
{
CommandParameterCollection newComParCol = new CommandParameterCollection();
foreach (PSObject deserializedCommandParameter in parameterCollection.Cast().Where(deserializedCommandParameter => !(deserializedCommandParameter.BaseObject is string)))
{
string parameterName;
object parameterValue;
if (TryGetJobPropertyValue(deserializedCommandParameter, "Name", out parameterName) && TryGetJobPropertyValue(deserializedCommandParameter, "Value", out parameterValue))
{
CommandParameter cp = new CommandParameter(parameterName, parameterValue);
newComParCol.Add(cp);
}
}
listComParCol.Add(newComParCol);
}
}
childProxyJob.StartParameters = listComParCol;
}
}
private void RegisterChildEvents()
{
if (_childEventsRegistered) return;
lock (SyncRoot)
{
if (_childEventsRegistered) return;
_childEventsRegistered = true;
// if the data added handler was provided, use it to subscribe to child events.
if (_dataAddedHandler != null)
{
foreach (PSChildJobProxy job in ChildJobs)
{
Dbg.Assert(job != null, "ChildJob that is not child job proxy exists");
job.JobDataAdded += _dataAddedHandler;
}
}
// we need to register for our handler to compute parent proxy's state
// after the passed in handlers are registered. This will ensure that
// our handler is invoked last
foreach (var job in ChildJobs)
{
job.StateChanged += HandleChildProxyJobStateChanged;
}
// likewise for state changed
if (_stateChangedHandler != null)
{
foreach (PSChildJobProxy job in ChildJobs)
{
Dbg.Assert(job != null, "ChildJob that is not child job proxy exists");
job.StateChanged += _stateChangedHandler;
}
}
}
}
private void UnregisterChildEvents()
{
lock (SyncRoot)
{
// Check inside the lock only, if the lock is held by the register method
// the childEventsRegistered may be altered before the lock is released.
if (!_childEventsRegistered) return;
// if the data added handler was provided, use it to unsubscribe from child events.
if (_dataAddedHandler != null)
{
foreach (PSChildJobProxy job in ChildJobs)
{
Dbg.Assert(job != null, "ChildJob that is not child job proxy exists");
job.JobDataAdded -= _dataAddedHandler;
}
}
// likewise for state changed
if (_stateChangedHandler != null)
{
foreach (PSChildJobProxy job in ChildJobs)
{
Dbg.Assert(job != null, "ChildJob that is not child job proxy exists");
job.StateChanged -= _stateChangedHandler;
}
}
foreach (var job in ChildJobs)
{
job.StateChanged -= HandleChildProxyJobStateChanged;
}
_childEventsRegistered = false;
}
}
private void HandleChildProxyJobStateChanged(object sender, JobStateEventArgs e)
{
JobState computedJobState;
if (!ContainerParentJob.ComputeJobStateFromChildJobStates(ClassNameTrace, e, ref _blockedChildJobsCount,
ref _suspendedChildJobsCount,
ref _suspendingChildJobsCount,
ref _finishedChildJobsCount,
ref _failedChildJobsCount,
ref _stoppedChildJobsCount,
ChildJobs.Count,
out computedJobState))
return;
if (computedJobState == JobState.Suspending) return; // Ignor for proxy job
_tracer.WriteMessage(ClassNameTrace, "HandleChildProxyJobStateChanged", Guid.Empty, this,
"storing job state to {0}", computedJobState.ToString());
// we need to store the state and set it only when invocation
// state changed for _receivePowerShell is received. This is to
// enable consumers from disposing the proxy job on its state
// changed handler
_computedJobState = computedJobState;
}
///
/// Check if changes to the jobs properties can be accepted
///
private void AssertChangesCanBeAccepted()
{
lock (SyncRoot)
{
AssertNotDisposed();
if (JobStateInfo.State != JobState.NotStarted)
{
throw new InvalidJobStateException(JobStateInfo.State);
}
}
}
private void AssertJobCanBeStartedAndSetStateToRunning()
{
lock (SyncRoot)
{
AssertNotDisposed();
// only a job which is not started or which is in a terminal
// state can be restarted
if (JobStateInfo.State != JobState.NotStarted && !IsFinishedState(JobStateInfo.State))
{
throw new InvalidJobStateException(JobStateInfo.State, StringUtil.Format(PowerShellStrings.JobCannotBeStartedWhenRunning));
}
if (_startCalled)
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.JobCanBeStartedOnce);
}
_startCalled = true;
}
_tracer.WriteMessage(ClassNameTrace, "AssertJobCanBeStartedAndSetStateToRunning", _remoteJobInstanceId, this,
"Setting job state to running", null);
ValidateAndDoSetJobState(JobState.Running);
}
private bool AssertStopJobIsValidAndSetToStopping()
{
lock (SyncRoot)
{
AssertNotDisposed();
if (JobStateInfo.State == JobState.NotStarted)
{
throw new InvalidJobStateException(JobState.NotStarted);
}
if (IsFinishedState(JobStateInfo.State))
return false;
}
DoSetJobState(JobState.Stopping);
return true;
}
private bool AssertSuspendJobIsValidAndSetToSuspending()
{
lock (SyncRoot)
{
AssertNotDisposed();
// Valid states: Running, Suspended, Suspending
if (JobStateInfo.State != JobState.Suspended
&& JobStateInfo.State != JobState.Suspending
&& JobStateInfo.State != JobState.Running)
{
throw new InvalidJobStateException(JobStateInfo.State);
}
}
if (JobStateInfo.State != JobState.Running) return false;
DoSetJobState(JobState.Suspending);
return true;
}
private void AssertResumeJobIsValidAndSetToRunning()
{
lock (SyncRoot)
{
AssertNotDisposed();
// Valid states: Running, Suspended, Suspending
if (JobStateInfo.State != JobState.Suspended
&& JobStateInfo.State != JobState.Suspending
&& JobStateInfo.State != JobState.Running)
{
throw new InvalidJobStateException(JobStateInfo.State);
}
}
if (JobStateInfo.State != JobState.Running) ValidateAndDoSetJobState(JobState.Running);
foreach (PSChildJobProxy job in ChildJobs)
{
Dbg.Assert(job != null, "child job is not PSChildJobProxy");
if (!IsFinishedState(job.JobStateInfo.State))
job.DoSetJobState(JobState.Running);
}
}
private void AssertRemoveJobIsValid()
{
if (JobStateInfo.State == JobState.NotStarted || _remoteJobInstanceId == Guid.Empty)
{
throw new InvalidJobStateException(JobStateInfo.State);
}
}
///
/// Assert if the object is not yet disposed and if so
/// throw an ObjectDisposedException
///
/// Thrown if
/// the object has already been disposed
/// Method is not thread-safe. Caller has to
/// ensure thread safety
private new void AssertNotDisposed()
{
if (_isDisposed)
{
throw PSTraceSource.NewObjectDisposedException("PSJobProxy");
}
}
private void ValidateAndDoSetJobState(JobState state, Exception reason = null)
{
// Do not set the state to running if the job is in the process of suspending or stopping.
// This can happen if Stop or Suspend is called during job initialization.
if ((_previousState == JobState.Stopping || _previousState == JobState.Suspending) && state == JobState.Running)
return;
DoSetJobState(state, reason);
}
private void DoSetJobState(JobState state, Exception reason = null)
{
// If the object is disposed, the user does not care about state.
if (_isDisposed) return;
lock (SyncRoot)
{
if (_previousState == state) return; // Do not set the state if the desired state is accurate.
_previousState = state;
}
#if DEBUG
switch (JobStateInfo.State)
{
case JobState.Running:
Dbg.Assert(state != JobState.NotStarted && state != JobState.Blocked,
"JobProxy invalid state transition from Running.");
break;
case JobState.Stopping:
Dbg.Assert(
state == JobState.Stopped || state == JobState.Failed || state == JobState.Completed,
"JobProxy invalid state transition from Stopping.");
break;
case JobState.Stopped:
Dbg.Assert(false, "JobProxy should never transition after Stopped state.");
break;
case JobState.Suspending:
Dbg.Assert(
state == JobState.Suspended || state == JobState.Completed || state == JobState.Failed ||
state == JobState.Stopped || state == JobState.Stopping,
"JobProxy invalid state transition from Suspending.");
break;
case JobState.Suspended:
Dbg.Assert(
state == JobState.Running || state == JobState.Stopping || state == JobState.Stopped,
"JobProxy invalid state transition from Suspended.");
break;
case JobState.Failed:
Dbg.Assert(false, "JobProxy should never transition after Failed state.");
break;
case JobState.Completed:
Dbg.Assert(false, "JobProxy should never transition after Completed state.");
break;
case JobState.Disconnected:
Dbg.Assert(state != JobState.NotStarted && state != JobState.Blocked,
"JobProxy invalid state transition from Disconnected.");
break;
case JobState.Blocked:
Dbg.Assert(false, "JobProxy should never be in a blocked state");
break;
default:
break;
}
#endif
try
{
_tracer.WriteMessage(ClassNameTrace, "DoSetJobState", _remoteJobInstanceId, this,
"BEGIN Set job state to {0} and call event handlers", state.ToString());
s_structuredTracer.EndProxyJobExecution(InstanceId);
s_structuredTracer.BeginProxyJobEventHandler(InstanceId);
SetJobState(state, reason);
s_structuredTracer.EndProxyJobEventHandler(InstanceId);
_tracer.WriteMessage(ClassNameTrace, "DoSetJobState", _remoteJobInstanceId, this,
"END Set job state to {0} and call event handlers", state.ToString());
}
catch (ObjectDisposedException)
{
_tracer.WriteMessage(ClassNameTrace, "DoSetJobState", _remoteJobInstanceId, this,
"Caught object disposed exception", null);
}
}
#endregion Private Methods
#region Private methods for Output management
private T GetRecord(object sender)
{
lock (SyncRoot)
{
var collection = sender as PSDataCollection;
Dbg.Assert(collection != null, "Sender cannot be null");
T data = collection.ReadAndRemoveAt0();
Dbg.Assert(data != null, "DataAdded should be raised for each object added");
return data;
}
}
private void DataAddedToOutput(object sender, DataAddedEventArgs e)
{
var newObject = GetRecord(sender);
if (!_jobInitialized)
{
// If the job is not initialized, then this should be the first object received after
// invoking a command. This condition should be met when StartJob is called.
// Note: These events will always come in sequence, So this is not thread safe.
_jobInitialized = true;
// Because we have added the AsJob parameter we expect
// the command to write out a job, if it does not
// the command fails.)
// If the command writes an error before any output, the proxy will use this
// as an indication of a failed command execution.
if (!Deserializer.IsDeserializedInstanceOfType(newObject, typeof(Job)))
{
_tracer.WriteMessage(ClassNameTrace, "DataAddedToOutput", _remoteJobInstanceId, this,
"Setting job state to failed. Command did not return a job object.",
null);
Exception reason = (_receivePowerShell.Streams.Error.Count == 0 ||
_receivePowerShell.Streams.Error[0].Exception == null)
? PSTraceSource.NewNotSupportedException(PowerShellStrings.CommandDoesNotWriteJob)
: _receivePowerShell.Streams.Error[0].Exception;
DoSetJobState(JobState.Failed, reason);
_jobInitialiedWaitHandle.Set();
OnStartJobCompleted(new AsyncCompletedEventArgs(reason, false, null));
return;
}
PopulateJobProperties(newObject);
RegisterChildEvents();
// Child Job States must be set after events are subscribed.
// All jobs start in the running state. Receive-Job will return
// a persistent job state event for each child job sometime after
// the job object is returned. This will update the job state of
// the child in a subsequent call to DataAddedToOutput.
foreach (PSChildJobProxy child in ChildJobs)
{
Dbg.Assert(child != null, "Child should always be PSChildJobProxy");
child.DoSetJobState(JobState.Running);
}
// Release any thread waiting start.
_jobInitialiedWaitHandle.Set();
_tracer.WriteMessage(ClassNameTrace, "DataAddedToOutput", Guid.Empty, this,
"BEGIN Invoke StartJobCompleted event", null);
OnStartJobCompleted(new AsyncCompletedEventArgs(null, false, null));
_tracer.WriteMessage(ClassNameTrace, "DataAddedToOutput", Guid.Empty, this,
"END Invoke StartJobCompleted event", null);
ProcessQueue();
return;
}
if (newObject.Properties[RemotingConstants.EventObject] != null)
{
PSPropertyInfo guidProperty = newObject.Properties[RemotingConstants.SourceJobInstanceId];
Guid sourceJobId = guidProperty != null ? (Guid)guidProperty.Value : Guid.Empty;
if (guidProperty == null || sourceJobId == Guid.Empty)
{
Diagnostics.Assert(false, "We should not get guidProperty as null or an empty source job id in non interop scenarios");
return;
}
if (!_childJobsMapping.ContainsKey(sourceJobId))
{
if (sourceJobId != _remoteJobInstanceId)
{
Diagnostics.Assert(false,
"We should not get an unidentified source job id in non interop scenarios");
}
return;
}
// If the event args are job state, the event is a Job's StateChanged event.
var jobStateEventArgs = newObject.BaseObject as JobStateEventArgs ??
Microsoft.PowerShell.DeserializingTypeConverter.RehydrateJobStateEventArgs(newObject);
if (jobStateEventArgs != null)
{
// If the child job sent the event, then change the proxy child's state.
// If the current job raised the event, the state change will come when the
// receive call returns.
_tracer.WriteMessage(ClassNameTrace, "DataAddedToOutput", Guid.Empty, this,
"Updating child job {0} state to {1} ", sourceJobId.ToString(),
jobStateEventArgs.JobStateInfo.State.ToString());
((PSChildJobProxy)_childJobsMapping[sourceJobId]).DoSetJobState(
jobStateEventArgs.JobStateInfo.State, jobStateEventArgs.JobStateInfo.Reason);
_tracer.WriteMessage(ClassNameTrace, "DataAddedToOutput", Guid.Empty, this,
"Finished updating child job {0} state to {1} ", sourceJobId.ToString(),
jobStateEventArgs.JobStateInfo.State.ToString());
}
return;
}
SortOutputObject(newObject);
}
private void SortOutputObject(PSObject newObject)
{
PSPropertyInfo guidProperty = newObject.Properties[RemotingConstants.SourceJobInstanceId];
Guid sourceJobId = guidProperty != null ? (Guid)guidProperty.Value : Guid.Empty;
if (guidProperty == null || sourceJobId == Guid.Empty || !_childJobsMapping.ContainsKey(sourceJobId))
{
// If there is no child job matching the source, add it to the parent's collection so that it is not lost.
Output.Add(newObject);
return;
}
// make sure the tag reflects client side job.
newObject.Properties.Remove(RemotingConstants.SourceJobInstanceId);
newObject.Properties.Add(new PSNoteProperty(RemotingConstants.SourceJobInstanceId,
((PSChildJobProxy)_childJobsMapping[sourceJobId]).InstanceId));
((PSChildJobProxy)_childJobsMapping[sourceJobId]).Output.Add(newObject);
}
private void DataAddedToError(object sender, DataAddedEventArgs e)
{
var newError = GetRecord(sender);
Dbg.Assert(newError != null, "DataAdded should be raised once for each piece of data.");
SortError(newError);
}
private void SortError(ErrorRecord record)
{
// var newJobError = record as RemotingErrorRecord;
// if (newJobError == null || !_childJobsMapping.ContainsKey(newJobError.OriginInfo.InstanceID))
Guid id = Guid.Empty;
string computerName = string.Empty;
if (record.ErrorDetails != null)
{
record.ErrorDetails.RecommendedAction = RemoveIdentifierInformation(record.ErrorDetails.RecommendedAction, out id, out computerName);
}
if (id == Guid.Empty || !_childJobsMapping.ContainsKey(id))
{
// If there is no child job matching the source, add it to the parent's collection so that it is not lost.
WriteError(record);
return;
}
// make sure the tag reflects client side job.
// newJobError.OriginInfo.InstanceID = ((PSChildJobProxy)_childJobsMapping[newJobError.OriginInfo.InstanceID]).InstanceId;
var oi = new OriginInfo(null, Guid.Empty, ((PSChildJobProxy)_childJobsMapping[id]).InstanceId);
((PSChildJobProxy)_childJobsMapping[id]).WriteError(new RemotingErrorRecord(record, oi));
}
private void DataAddedToProgress(object sender, DataAddedEventArgs e)
{
var newRecord = GetRecord(sender);
Dbg.Assert(newRecord != null, "DataAdded should be raised once for each piece of data.");
SortProgress(newRecord);
}
private void SortProgress(ProgressRecord newRecord)
{
// var newJobRecord = newRecord as RemotingProgressRecord;
// if (newJobRecord == null || !_childJobsMapping.ContainsKey(newJobRecord.OriginInfo.InstanceID))
Guid id;
string computerName;
newRecord.CurrentOperation = RemoveIdentifierInformation(newRecord.CurrentOperation, out id, out computerName);
if (id == Guid.Empty || !_childJobsMapping.ContainsKey(id))
{
// If there is no child job matching the source, add it to the parent's collection so that it is not lost.
WriteProgress(newRecord);
return;
}
// make sure the tag reflects client side job.
// newJobRecord.OriginInfo.InstanceID = ((PSChildJobProxy)_childJobsMapping[newJobRecord.OriginInfo.InstanceID]).InstanceId;
var oi = new OriginInfo(computerName, Guid.Empty, ((PSChildJobProxy)_childJobsMapping[id]).InstanceId);
((PSChildJobProxy)_childJobsMapping[id]).WriteProgress(new RemotingProgressRecord(newRecord, oi));
}
private void DataAddedToDebug(object sender, DataAddedEventArgs e)
{
var record = GetRecord(sender);
Dbg.Assert(record != null, "DataAdded should be raised once for each piece of data.");
SortDebug(record);
}
private void SortDebug(DebugRecord record)
{
Guid remoteJobInstanceId;
string computerName;
string message = RemoveIdentifierInformation(record.Message, out remoteJobInstanceId, out computerName);
if (remoteJobInstanceId == Guid.Empty || !_childJobsMapping.ContainsKey(remoteJobInstanceId))
{
// If there is no child job matching the source, add it to the parent's collection so that it is not lost.
WriteDebug(message);
return;
}
// make sure the tag reflects client side job.
var originInfo = new OriginInfo(computerName, Guid.Empty, ((PSChildJobProxy)_childJobsMapping[remoteJobInstanceId]).InstanceId);
((PSChildJobProxy)_childJobsMapping[remoteJobInstanceId]).Debug.Add(new RemotingDebugRecord(message, originInfo));
}
private void DataAddedToWarning(object sender, DataAddedEventArgs e)
{
var record = GetRecord(sender);
Dbg.Assert(record != null, "DataAdded should be raised once for each piece of data.");
SortWarning(record);
}
private void SortWarning(WarningRecord record)
{
Guid remoteJobInstanceId;
string computerName;
string message = RemoveIdentifierInformation(record.Message, out remoteJobInstanceId, out computerName);
if (remoteJobInstanceId == Guid.Empty || !_childJobsMapping.ContainsKey(remoteJobInstanceId))
{
// If there is no child job matching the source, add it to the parent's collection so that it is not lost.
WriteWarning(message);
return;
}
// make sure the tag reflects client side job.
var originInfo = new OriginInfo(computerName, Guid.Empty, ((PSChildJobProxy)_childJobsMapping[remoteJobInstanceId]).InstanceId);
((PSChildJobProxy)_childJobsMapping[remoteJobInstanceId]).Warning.Add(new RemotingWarningRecord(message, originInfo));
}
private void DataAddedToVerbose(object sender, DataAddedEventArgs e)
{
var record = GetRecord(sender);
Dbg.Assert(record != null, "DataAdded should be raised once for each piece of data.");
SortVerbose(record);
}
private void SortVerbose(VerboseRecord record)
{
Guid remoteJobInstanceId;
string computerName;
string message = RemoveIdentifierInformation(record.Message, out remoteJobInstanceId, out computerName);
if (remoteJobInstanceId == Guid.Empty || !_childJobsMapping.ContainsKey(remoteJobInstanceId))
{
// If there is no child job matching the source, add it to the parent's collection so that it is not lost.
WriteVerbose(message);
return;
}
// make sure the tag reflects client side job.
var originInfo = new OriginInfo(computerName, Guid.Empty, ((PSChildJobProxy)_childJobsMapping[remoteJobInstanceId]).InstanceId);
((PSChildJobProxy)_childJobsMapping[remoteJobInstanceId]).Verbose.Add(new RemotingVerboseRecord(message, originInfo));
}
private void DataAddedToInformation(object sender, DataAddedEventArgs e)
{
var record = GetRecord(sender);
Dbg.Assert(record != null, "DataAdded should be raised once for each piece of data.");
SortInformation(record);
}
private void SortInformation(InformationRecord record)
{
Guid remoteJobInstanceId;
string computerName;
record.Source = RemoveIdentifierInformation(record.Source, out remoteJobInstanceId, out computerName);
if (remoteJobInstanceId == Guid.Empty || !_childJobsMapping.ContainsKey(remoteJobInstanceId))
{
// If there is no child job matching the source, add it to the parent's collection so that it is not lost.
WriteInformation(record);
return;
}
// make sure the tag reflects client side job.
var originInfo = new OriginInfo(computerName, Guid.Empty, ((PSChildJobProxy)_childJobsMapping[remoteJobInstanceId]).InstanceId);
((PSChildJobProxy)_childJobsMapping[remoteJobInstanceId]).Information.Add(new RemotingInformationRecord(record, originInfo));
}
private static string RemoveIdentifierInformation(string message, out Guid jobInstanceId, out string computerName)
{
jobInstanceId = Guid.Empty;
computerName = string.Empty;
if (!string.IsNullOrEmpty(message))
{
string[] parts = message.Split(Utils.Separators.Colon, 3);
if (parts.Length == 3)
{
if (!Guid.TryParse(parts[0], out jobInstanceId))
jobInstanceId = Guid.Empty;
computerName = parts[1];
return parts[2];
}
}
return message;
}
#endregion Output management
#region IDisposable Overrides
///
/// Dispose all managed resources
///
/// _pendingOperations = new ConcurrentQueue();
private ManualResetEvent _removeComplete;
private ManualResetEvent RemoveComplete
{
get
{
if (_removeComplete == null)
{
lock (SyncRoot)
{
if (_removeComplete == null)
{
// this assert is required so that a wait handle
// is not created after the object is disposed
// which will result in a leak
AssertNotDisposed();
_removeComplete = new ManualResetEvent(false);
}
}
}
return _removeComplete;
}
}
private ManualResetEvent _jobRunningOrFinished;
private ManualResetEvent JobRunningOrFinished
{
get
{
if (_jobRunningOrFinished == null)
{
lock (SyncRoot)
{
if (_jobRunningOrFinished == null)
{
// this assert is required so that a wait handle
// is not created after the object is disposed
// which will result in a leak
AssertNotDisposed();
_jobRunningOrFinished = new ManualResetEvent(false);
}
}
}
return _jobRunningOrFinished;
}
}
private readonly ManualResetEvent _jobInitialiedWaitHandle = new ManualResetEvent(false);
private ManualResetEvent _jobSuspendedOrFinished;
private ManualResetEvent JobSuspendedOrFinished
{
get
{
if (_jobSuspendedOrFinished == null)
{
lock (SyncRoot)
{
if (_jobSuspendedOrFinished == null)
{
// this assert is required so that a wait handle
// is not created after the object is disposed
// which will result in a leak
AssertNotDisposed();
_jobSuspendedOrFinished = new ManualResetEvent(false);
}
}
}
return _jobSuspendedOrFinished;
}
}
private PSCommand _pscommand;
private Runspace _runspace;
private RunspacePool _runspacePool;
private EventHandler _dataAddedHandler;
private EventHandler _stateChangedHandler;
private const String ResBaseName = "PowerShellStrings";
private Guid _remoteJobInstanceId = Guid.Empty;
private string _remoteJobStatusMessage = String.Empty;
private string _remoteJobLocation = String.Empty;
private readonly Hashtable _childJobsMapping = new Hashtable();
private readonly PowerShell _receivePowerShell = PowerShell.Create();
private readonly PSDataCollection _receivePowerShellOutput = new PSDataCollection();
private bool _moreData = true;
private JobState _previousState = JobState.NotStarted;
private JobState _computedJobState;
private readonly PowerShellTraceSource _tracer = PowerShellTraceSourceFactory.GetTraceSource();
private static Tracer s_structuredTracer = new Tracer();
private bool _jobInitialized;
private bool _removeCalled;
private bool _startCalled;
private bool _receiveIsValidCall;
private bool _isDisposed;
private bool _remoteJobRemoved;
private bool _childEventsRegistered;
private object _inProgressSyncObject = new object();
private bool _inProgress = false;
private bool _workerCreated = false;
private const string ClassNameTrace = "PSJobProxy";
// count of number of child jobs which have finished
private int _finishedChildJobsCount = 0;
// count of number of child jobs which are blocked
private int _blockedChildJobsCount = 0;
// count of number of child jobs which are suspended
private int _suspendedChildJobsCount = 0;
// count of number of child jobs which are suspending
private int _suspendingChildJobsCount = 0;
// count of number of child jobs which failed
private int _failedChildJobsCount = 0;
// count of number of child jobs which stopped
private int _stoppedChildJobsCount = 0;
#endregion Private Members
}
///
/// Job class used for children of PSJobProxy jobs.
///
public sealed class PSChildJobProxy : Job2
{
internal PSChildJobProxy(string command, PSObject o)
: base(command)
{
PSJobProxy.TryGetJobPropertyValue(o, "StatusMessage", out _statusMessage);
PSJobProxy.TryGetJobPropertyValue(o, "Location", out _location);
string name;
PSJobProxy.TryGetJobPropertyValue(o, "Name", out name);
Name = name;
Output.DataAdded += OutputAdded;
Error.DataAdded += ErrorAdded;
Warning.DataAdded += WarningAdded;
Verbose.DataAdded += VerboseAdded;
Progress.DataAdded += ProgressAdded;
Debug.DataAdded += DebugAdded;
Information.DataAdded += InformationAdded;
}
internal void AssignDisconnectedState()
{
DoSetJobState(JobState.Disconnected);
}
///
/// Method to raise the event when this job has data added.
///
///
/// Dispose
///
///
/// Not supported
///
public override void StartJob()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// Not supported
///
public override void StartJobAsync()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// StopJob
///
///
/// Not supported
///
public override void StopJobAsync()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// StopJobAsync
///
///
/// Not supported
///
public override void SuspendJob()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// Not supported
///
public override void SuspendJobAsync()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// SuspendJob
///
///
/// SuspendJobAsync
///
///
/// Not supported
///
public override void ResumeJob()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// Not supported
///
public override void ResumeJobAsync()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// Not supported
///
public override void UnblockJob()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// Not supported
///
public override void UnblockJobAsync()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
///
/// Not supported
///
public override void StopJob()
{
throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyChildJobControlNotSupported);
}
#endregion Control method overrides
#region Public properties
///
/// This event will be raised whenever data has been added to one of the job object's
/// 6 collections. The event arguments include the job itself, the data type, indicating
/// which collection has data added, and the index.
///
public event EventHandler JobDataAdded;
///
/// Status message
///
public override string StatusMessage
{
get { return _statusMessage; }
}
///
/// Indicates the job has or can have more data on one or more data collection
///
public override bool HasMoreData
{
get
{
return (Output.IsOpen || Output.Count > 0
|| Error.IsOpen || Error.Count > 0
|| Verbose.IsOpen || Verbose.Count > 0
|| Debug.IsOpen || Debug.Count > 0
|| Warning.IsOpen || Warning.Count > 0
|| Progress.IsOpen || Progress.Count > 0
|| Information.IsOpen || Information.Count > 0
);
}
}
///
/// The location of the job
///
public override string Location
{
get { return _location; }
}
#endregion Public properties
#region Private members
private string _statusMessage;
private string _location;
private readonly PowerShellTraceSource _tracer = PowerShellTraceSourceFactory.GetTraceSource();
private readonly object _syncObject = new object();
private bool _disposed = false;
private const string ClassNameTrace = "PSChildJobProxy";
#endregion
}
///
/// Event arguments that indicate data has been added to a child job.
///
public sealed class JobDataAddedEventArgs : EventArgs
{
///
/// Constructor
///
///
/// The job that contains the PSDataCollection which is the sender.
///
public Job SourceJob { get; }
///
/// Identifies the type of the sending collection as one of the six collections
/// associated with a job.
/// If data type = output, sender is PSDataCollection of PSObject, Error is of ErrorRecord, etc.
///
public PowerShellStreamType DataType { get; }
///
/// Index at which the data is added.
///
public int Index { get; }
}
///
/// Job data is added to one of these streams. Each
/// type of data implies a different type of object.
///
public enum PowerShellStreamType
{
///
/// PSObject
///
Input = 0,
///
/// PSObject
///
Output = 1,
///
/// ErrorRecord
///
Error = 2,
///
/// WarningRecord
///
Warning = 3,
///
/// VerboseRecord
///
Verbose = 4,
///
/// DebugRecord
///
Debug = 5,
///
/// ProgressRecord
///
Progress = 6,
///
/// InformationRecord
///
Information = 7
}
#endregion Workflow Hosting API
}