Job.cs

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Diagnostics.CodeAnalysis;
using System.Management.Automation.Language;
using System.Management.Automation.Remoting.Internal;
using System.Management.Automation.Tracing;
using System.Management.Automation.Remoting;
using System.Management.Automation.Runspaces;
using System.Management.Automation.Internal;
using System.Management.Automation.Host;
using System.Runtime.Serialization;
using System.Threading;
using System.Text;
using Microsoft.PowerShell.Commands;
using Dbg = System.Management.Automation.Diagnostics;

// Stops compiler from warning about unknown warnings
#pragma warning disable 1634, 1691

namespace System.Management.Automation
{
    /// <summary>
    /// Enumeration for job status values. Indicates the status
    /// of the result object
    /// </summary>
    public enum JobState
    {
        /// <summary>
        /// Execution of command in job not started
        /// </summary>
        NotStarted = 0,

        /// <summary>
        /// execution of command in progress
        /// </summary>
        Running = 1,

        /// <summary>
        /// execution of command completed in all
        /// computernames/runspaces
        /// </summary>
        Completed = 2,

        /// <summary>
        /// An error was encountered when trying to executed
        /// command in one or more computernames/runspaces
        /// </summary>
        Failed = 3,

        /// <summary>
        /// Command execution is cancelled (stopped) in one or more
        /// computernames/runspaces.
        /// </summary>
        Stopped = 4,

        /// <summary>
        /// Command execution is blocked (on user input host calls etc)
        /// </summary>
        Blocked = 5,

        /// <summary>
        /// The job has been suspended
        /// </summary>
        Suspended = 6,

        /// <summary>
        /// The job is a remote job and has been disconnected from the server.
        /// </summary>
        Disconnected = 7,

        /// <summary>
        /// Suspend is in progress
        /// </summary>
        Suspending = 8,

        /// <summary>
        /// Stop is in progress
        /// </summary>
        Stopping = 9,

        /// <summary>
        /// Script execution is halted in a debugger stop.
        /// </summary>
        AtBreakpoint = 10
    }

    /// <summary>
    /// Defines exception which is thrown when state of the PSJob is different
    /// from the expected state.
    /// </summary>
    [Serializable]
    public class InvalidJobStateException : SystemException
    {
        /// <summary>
        /// Creates a new instance of InvalidPSJobStateException class.
        /// </summary>
        public InvalidJobStateException()
            : base
        (
            PSRemotingErrorInvariants.FormatResourceString
            (
                RemotingErrorIdStrings.InvalidJobStateGeneral
            )
        )
        {
        }

        /// <summary>
        /// Creates a new instance of InvalidPSJobStateException class.
        /// </summary>
        /// <param name="message">
        /// The error message that explains the reason for the exception.
        /// </param>
        public InvalidJobStateException(string message)
            : base(message)
        {
        }

        /// <summary>
        /// Creates a new instance of InvalidPSJobStateException class.
        /// </summary>
        /// <param name="message">
        /// The error message that explains the reason for the exception.
        /// </param>
        /// <param name="innerException">
        /// The exception that is the cause of the current exception.
        /// </param>
        public InvalidJobStateException(string message, Exception innerException)
            : base(message, innerException)
        {
        }

        /// <summary>
        /// Creates a new instance of InvalidJobStateException class.
        /// </summary>
        /// <param name="currentState">
        /// The Job State at the time of the error.
        /// </param>
        /// <param name="actionMessage">
        /// An additional message that gives more information about the error. Used
        /// for context after a generalized error message.
        /// </param>
        public InvalidJobStateException(JobState currentState, string actionMessage)
            : base
        (
            PSRemotingErrorInvariants.FormatResourceString
            (
                RemotingErrorIdStrings.InvalidJobStateSpecific, currentState, actionMessage
            )
        )
        {
            _currState = currentState;
        }

        /// <summary>
        /// Initializes a new instance of the InvalidPSJobStateException and defines value of
        /// CurrentState.
        /// </summary>
        /// <param name="currentState">Current state of powershell</param>
        internal InvalidJobStateException(JobState currentState)
            : base
        (
            PSRemotingErrorInvariants.FormatResourceString
            (
                RemotingErrorIdStrings.InvalidJobStateGeneral
            )
        )
        {
            _currState = currentState;
        }

        #region ISerializable Members

        // No need to implement GetObjectData
        // if all fields are static or [NonSerialized]

        /// <summary>
        /// Initializes a new instance of the InvalidPSJobStateException
        /// class with serialized data.
        /// </summary>
        /// <param name="info">
        /// The <see cref="SerializationInfo"/> that holds the serialized object
        /// data about the exception being thrown.
        /// </param>
        /// <param name="context">
        /// The <see cref="StreamingContext"/> that contains contextual information
        /// about the source or destination.
        /// </param>
        protected
        InvalidJobStateException(SerializationInfo info, StreamingContext context)
            : base(info, context)
        {
        }

        #endregion

        /// <summary>
        /// Gets CurrentState of the Job
        /// </summary>
        public JobState CurrentState
        {
            get
            {
                return _currState;
            }
        }

        /// <summary>
        /// State of job when exception was thrown.
        /// </summary>
        [NonSerialized]
        private JobState _currState = 0;
    }

    /// <summary>
    /// Type which has information about JobState and Exception
    /// ,if any, associated with JobState
    /// </summary>
    public sealed class JobStateInfo
    {
        #region constructors

        /// <summary>
        /// Constructor for state changes not resulting from an error.
        /// </summary>
        /// <param name="state">Execution state</param>
        public JobStateInfo(JobState state)
            : this(state, null)
        {
        }

        /// <summary>
        /// Constructor for state changes with an optional error.
        /// </summary>
        /// <param name="state">The new state.</param>
        /// <param name="reason">A non-null exception if the state change was
        /// caused by an error,otherwise; null.
        /// </param>
        public JobStateInfo(JobState state, Exception reason)
        {
            State = state;
            Reason = reason;
        }

        /// <summary>
        /// Copy constructor to support cloning
        /// </summary>
        /// <param name="jobStateInfo">source information</param>
        /// <throws>
        /// ArgumentNullException when <paramref name="jobStateInfo"/> is null.
        /// </throws>
        internal JobStateInfo(JobStateInfo jobStateInfo)
        {
            State = jobStateInfo.State;
            Reason = jobStateInfo.Reason;
        }

        #endregion constructors

        #region public_properties

        /// <summary>
        /// The state of the job.
        /// </summary>
        /// <remarks>
        /// This value indicates the state of the job .
        /// </remarks>
        public JobState State { get; }

        /// <summary>
        /// The reason for the state change, if caused by an error.
        /// </summary>
        /// <remarks>
        /// The value of this property is non-null if the state
        /// changed due to an error. Otherwise, the value of this
        /// property is null.
        /// </remarks>
        public Exception Reason { get; }

        #endregion public_properties

        /// <summary>
        /// override for ToString()
        /// </summary>
        /// <returns></returns>
        public override string ToString()
        {
            return State.ToString();
        }

        /// <summary>
        /// Clones this object
        /// </summary>
        /// <returns>Cloned object</returns>
        internal JobStateInfo Clone()
        {
            return new JobStateInfo(this);
        }

        #region private_fields

        #endregion private_fields
    }

    /// <summary>
    /// Event arguments passed to JobStateEvent handlers
    /// <see cref="Job.StateChanged"/> event.
    /// </summary>
    public sealed class JobStateEventArgs : EventArgs
    {
        #region constructors

        /// <summary>
        /// Constructor of JobStateEventArgs
        /// </summary>
        /// <param name="jobStateInfo">The current state of the job.</param>
        public JobStateEventArgs(JobStateInfo jobStateInfo)
            : this(jobStateInfo, null)
        {
        }

        /// <summary>
        /// Constructor of JobStateEventArgs
        /// </summary>
        /// <param name="jobStateInfo">The current state of the job.</param>
        /// <param name="previousJobStateInfo">The previous state of the job.</param>
        public JobStateEventArgs(JobStateInfo jobStateInfo, JobStateInfo previousJobStateInfo)
        {
            if (jobStateInfo == null)
            {
                throw PSTraceSource.NewArgumentNullException("jobStateInfo");
            }
            JobStateInfo = jobStateInfo;
            PreviousJobStateInfo = previousJobStateInfo;
        }

        #endregion constructors

        #region public_properties

        /// <summary>
        /// Info about the current state of the job.
        /// </summary>
        public JobStateInfo JobStateInfo { get; }

        /// <summary>
        /// Info about the previous state of the job.
        /// </summary>
        public JobStateInfo PreviousJobStateInfo { get; }

        #endregion public_properties
    }

    /// <summary>
    /// Object that must be created by PowerShell to allow reuse of an ID for a job.
    /// Also allows setting of the Instance Id so that jobs may be recreated.
    /// </summary>
    public sealed class JobIdentifier
    {
        internal JobIdentifier(int id, Guid instanceId)
        {
            if (id <= 0)
                PSTraceSource.NewArgumentException("id", RemotingErrorIdStrings.JobSessionIdLessThanOne, id);
            Id = id;
            InstanceId = instanceId;
        }

        internal int Id { get; private set; }
        internal Guid InstanceId { get; private set; }
    }

    /// <summary>
    /// Interface to expose a job debugger.
    /// </summary>
    public interface IJobDebugger
    {
        /// <summary>
        /// Job Debugger
        /// </summary>
        Debugger Debugger
        {
            get;
        }

        /// <summary>
        /// True if job is running asynchronously.
        /// </summary>
        bool IsAsync
        {
            get;
            set;
        }
    }

    /// <summary>
    /// Represents a command running in background. A job object can internally
    /// contain many child job objects.
    /// </summary>
    public abstract class Job : IDisposable
    {
        #region Constructor

        /// <summary>
        /// Default constructor
        /// </summary>
        protected Job()
        {
            Id = System.Threading.Interlocked.Increment(ref s_jobIdSeed);
        }

        /// <summary>
        /// Creates an instance of this class
        /// </summary>
        /// <param name="command">Command invoked by this job object</param>
        protected Job(string command)
            : this()
        {
            Command = command;
            _name = AutoGenerateJobName();
        }

        /// <summary>
        /// Creates an instance of this class
        /// </summary>
        /// <param name="command">Command invoked by this job object</param>
        /// <param name="name">Friendly name for the job object</param>
        protected Job(string command, string name)
            : this(command)
        {
            if (!String.IsNullOrEmpty(name))
            {
                _name = name;
            }
        }

        /// <summary>
        /// Creates an instance of this class
        /// </summary>
        /// <param name="command">Command invoked by this job object</param>
        /// <param name="name">Friendly name for the job object</param>
        /// <param name="childJobs">Child jobs of this job object</param>
        protected Job(string command, string name, IList<Job> childJobs)
            : this(command, name)
        {
            _childJobs = childJobs;
        }

        /// <summary>
        /// Creates an instance of this class
        /// </summary>
        /// <param name="command">Command invoked by this job object</param>
        /// <param name="name">Friendly name for the job object</param>
        /// <param name="token">Id and InstanceId pair to be used for this job object</param>
        /// <remarks>The JobIdentifier is a token that must be issued by PowerShell to allow
        /// reuse of the Id. This is the only way to set either Id or instance Id.</remarks>
        protected Job(string command, string name, JobIdentifier token)
        {
            if (token == null)
                throw PSTraceSource.NewArgumentNullException("token", RemotingErrorIdStrings.JobIdentifierNull);
            if (token.Id > s_jobIdSeed)
            {
                throw PSTraceSource.NewArgumentException("token", RemotingErrorIdStrings.JobIdNotYetAssigned, token.Id);
            }

            Command = command;

            Id = token.Id;
            InstanceId = token.InstanceId;

            if (!String.IsNullOrEmpty(name))
            {
                _name = name;
            }
            else
            {
                _name = AutoGenerateJobName();
            }
        }

        /// <summary>
        /// Creates an instance of this class
        /// </summary>
        /// <param name="command">Command invoked by this job object</param>
        /// <param name="name">Friendly name for the job object</param>
        /// <param name="instanceId">InstanceId to be used for this job object</param>
        /// <remarks>The InstanceId may need to be set to maintain job identity across
        /// instances of the process.</remarks>
        protected Job(string command, string name, Guid instanceId)
            : this(command, name)
        {
            InstanceId = instanceId;
        }

        internal static string GetCommandTextFromInvocationInfo(InvocationInfo invocationInfo)
        {
            if (invocationInfo == null)
            {
                return null;
            }

            IScriptExtent scriptExtent = invocationInfo.ScriptPosition;
            if ((scriptExtent != null) && (scriptExtent.StartScriptPosition != null) && !string.IsNullOrWhiteSpace(scriptExtent.StartScriptPosition.Line))
            {
                Dbg.Assert(scriptExtent.StartScriptPosition.ColumnNumber > 0, "Column numbers start at 1");
                Dbg.Assert(scriptExtent.StartScriptPosition.ColumnNumber <= scriptExtent.StartScriptPosition.Line.Length, "Column numbers are not greater than the length of a line");
                return scriptExtent.StartScriptPosition.Line.Substring(scriptExtent.StartScriptPosition.ColumnNumber - 1).Trim();
            }

            return invocationInfo.InvocationName;
        }

        #endregion Constructor

        #region Private Members

        private ManualResetEvent _finished = new ManualResetEvent(false);

        private string _name;
        private IList<Job> _childJobs;
        internal readonly object syncObject = new object();   // object used for synchronization
        //ISSUE: Should Result be public property
        private PSDataCollection<PSStreamObject> _results = new PSDataCollection<PSStreamObject>();
        private bool _resultsOwner = true;
        private PSDataCollection<ErrorRecord> _error = new PSDataCollection<ErrorRecord>();
        private bool _errorOwner = true;
        private PSDataCollection<ProgressRecord> _progress = new PSDataCollection<ProgressRecord>();
        private bool _progressOwner = true;
        private PSDataCollection<VerboseRecord> _verbose = new PSDataCollection<VerboseRecord>();
        private bool _verboseOwner = true;
        private PSDataCollection<WarningRecord> _warning = new PSDataCollection<WarningRecord>();
        private bool _warningOwner = true;
        private PSDataCollection<DebugRecord> _debug = new PSDataCollection<DebugRecord>();
        private bool _debugOwner = true;
        private PSDataCollection<InformationRecord> _information = new PSDataCollection<InformationRecord>();
        private bool _informationOwner = true;
        private PSDataCollection<PSObject> _output = new PSDataCollection<PSObject>();
        private bool _outputOwner = true;

        /// <summary>
        /// Static variable which is incremented to generate id
        /// </summary>
        private static int s_jobIdSeed = 0;

        private string _jobTypeName = String.Empty;

        #endregion Private Members

        #region Job Properties

        /// <summary>
        /// Command Invoked by this Job
        /// </summary>
        public string Command { get; }

        /// <summary>
        /// Status of the command execution.
        /// </summary>
        public JobStateInfo JobStateInfo { get; private set; } = new JobStateInfo(JobState.NotStarted);

        /// <summary>
        /// Wait Handle which is signaled when job is finished.
        /// This is set when state of the job is set to Completed,
        /// Stopped or Failed.
        /// </summary>
        public WaitHandle Finished
        {
            get
            {
                lock (this.syncObject)
                {
                    if (_finished != null)
                    {
                        return _finished;
                    }
                    else
                    {
                        // Damage control mode:
                        // Somebody is trying to get Finished handle for an already disposed Job instance.
                        // Return an already triggered handle (disposed job is finished by definition).
                        // The newly created handle will not be disposed in a deterministic manner
                        // and in some circumstances can be mistaken for a handle leak.
                        return new ManualResetEvent(true);
                    }
                }
            }
        }

        /// <summary>
        /// unique identifier for this job
        /// </summary>
        public Guid InstanceId { get; } = Guid.NewGuid();

        /// <summary>
        /// Short identifier for this result which will be
        /// recycled and used within a process
        /// </summary>
        public Int32 Id { get; }

        /// <summary>
        /// Name for identifying this job object
        /// </summary>
        public String Name
        {
            get
            {
                return _name;
            }
            set
            {
                AssertNotDisposed();
                _name = value;
            }
        }

        /// <summary>
        /// List of child jobs contained within this job
        /// </summary>
        public IList<Job> ChildJobs
        {
            get
            {
                if (_childJobs == null)
                {
                    lock (syncObject)
                    {
                        if (_childJobs == null)
                        {
                            _childJobs = new List<Job>();
                        }
                    }
                }
                return _childJobs;
            }
        }

        ///<summary>
        /// Success status of the command execution.
        /// </summary>
        public abstract String StatusMessage { get; }

        /// <summary>
        /// Indicates that more data is available in this
        /// result object for reading.
        /// </summary>
        public abstract bool HasMoreData { get; }

        /// <summary>
        /// Time job was started.
        /// </summary>
        public DateTime? PSBeginTime { get; protected set; } = null;

        /// <summary>
        /// Time job stopped.
        /// </summary>
        public DateTime? PSEndTime { get; protected set; } = null;

        /// <summary>
        /// Job type name.
        /// </summary>
        public string PSJobTypeName
        {
            get { return _jobTypeName; }
            protected internal set
            {
                _jobTypeName = value ?? this.GetType().ToString();
            }
        }

        #region results

        /// <summary>
        /// Result objects from this job. If this object is not a
        /// leaf node (with no children), then this will
        /// aggregate the results from all child jobs
        /// </summary>
        internal PSDataCollection<PSStreamObject> Results
        {
            get
            {
                return _results;
            }
            set
            {
                if (value == null)
                {
                    throw PSTraceSource.NewArgumentNullException("Results");
                }
                lock (syncObject)
                {
                    AssertChangesAreAccepted();
                    _resultsOwner = false;
                    _results = value;
                }
            }
        }

        /// <summary>
        /// Indicates if a particular Job type uses the
        /// internal results collection
        /// </summary>
        internal bool UsesResultsCollection { get; set; }

        /// <summary>
        /// Suppresses forwarding of job output into a cmdlet (like Receive-Job).
        /// This flag modifies functionality of <see cref="WriteObject"/> method, so that it doesnt add output-processing to <see cref="Results"/> collection.
        /// </summary>
        internal bool SuppressOutputForwarding { get; set; }

        internal virtual void WriteObject(object outputObject)
        {
            PSObject pso = (outputObject == null) ? null : PSObject.AsPSObject(outputObject);
            this.Output.Add(pso);

            if (!SuppressOutputForwarding)
            {
                this.Results.Add(new PSStreamObject(PSStreamObjectType.Output, pso));
            }
        }

        /// <summary>
        /// Allows propagating of terminating exceptions from remote "throw" statement
        /// (normally / by default all remote errors are transformed into non-terminating errors
        /// </summary>
        internal bool PropagateThrows { get; set; }

        private void WriteError(Cmdlet cmdlet, ErrorRecord errorRecord)
        {
            if (this.PropagateThrows)
            {
                Exception e = GetExceptionFromErrorRecord(errorRecord);
                if (e != null)
                    throw e;
            }

            errorRecord.PreserveInvocationInfoOnce = true;
            cmdlet.WriteError(errorRecord);
        }

        private static Exception GetExceptionFromErrorRecord(ErrorRecord errorRecord)
        {
            RuntimeException runtimeException = errorRecord.Exception as RuntimeException;
            if (runtimeException == null)
                return null;

            RemoteException remoteException = runtimeException as RemoteException;
            if (remoteException == null)
                return null;

            PSPropertyInfo wasThrownFromThrow =
                remoteException.SerializedRemoteException.Properties["WasThrownFromThrowStatement"];
            if (wasThrownFromThrow == null || !((bool)wasThrownFromThrow.Value))
                return null;

            runtimeException.WasThrownFromThrowStatement = true;
            return runtimeException;
        }

        internal virtual void WriteError(ErrorRecord errorRecord)
        {
            Error.Add(errorRecord);
            if (PropagateThrows)
            {
                Exception exception = GetExceptionFromErrorRecord(errorRecord);
                if (exception != null)
                {
                    Results.Add(new PSStreamObject(PSStreamObjectType.Exception, exception));
                    return;
                }
            }
            Results.Add(new PSStreamObject(PSStreamObjectType.Error, errorRecord));
        }

        internal void WriteError(ErrorRecord errorRecord, out Exception exceptionThrownOnCmdletThread)
        {
            this.Error.Add(errorRecord);
            this.InvokeCmdletMethodAndWaitForResults<object>(
                    delegate (Cmdlet cmdlet)
                    {
                        this.WriteError(cmdlet, errorRecord);
                        return null;
                    },
                    out exceptionThrownOnCmdletThread);
        }

        internal virtual void WriteWarning(string message)
        {
            this.Warning.Add(new WarningRecord(message));
            this.Results.Add(new PSStreamObject(PSStreamObjectType.Warning, message));
        }

        internal virtual void WriteVerbose(string message)
        {
            this.Verbose.Add(new VerboseRecord(message));
            this.Results.Add(new PSStreamObject(PSStreamObjectType.Verbose, message));
        }

        internal virtual void WriteDebug(string message)
        {
            this.Debug.Add(new DebugRecord(message));
            this.Results.Add(new PSStreamObject(PSStreamObjectType.Debug, message));
        }

        internal virtual void WriteProgress(ProgressRecord progressRecord)
        {
            if ((progressRecord.ParentActivityId == (-1)) && (_parentActivityId != null))
            {
                progressRecord = new ProgressRecord(progressRecord) { ParentActivityId = _parentActivityId.Value };
            }

            Progress.Add(progressRecord);
            Results.Add(new PSStreamObject(PSStreamObjectType.Progress, progressRecord));
        }

        internal virtual void WriteInformation(InformationRecord informationRecord)
        {
            Information.Add(informationRecord);
            Results.Add(new PSStreamObject(PSStreamObjectType.Information, informationRecord));
        }

        private Lazy<int> _parentActivityId;
        internal void SetParentActivityIdGetter(Func<int> parentActivityIdGetter)
        {
            Dbg.Assert(parentActivityIdGetter != null, "Caller should verify parentActivityIdGetter != null");
            _parentActivityId = new Lazy<int>(parentActivityIdGetter);
        }

        internal bool ShouldContinue(string query, string caption)
        {
            Exception exceptionThrownOnCmdletThread;
            return this.ShouldContinue(query, caption, out exceptionThrownOnCmdletThread);
        }

        internal bool ShouldContinue(string query, string caption, out Exception exceptionThrownOnCmdletThread)
        {
            bool methodResult = InvokeCmdletMethodAndWaitForResults(
                cmdlet => cmdlet.ShouldContinue(query, caption),
                out exceptionThrownOnCmdletThread);
            return methodResult;
        }

        internal virtual void NonblockingShouldProcess(
            string verboseDescription,
            string verboseWarning,
            string caption)
        {
            InvokeCmdletMethodAndIgnoreResults(
                    delegate (Cmdlet cmdlet)
                    {
                        ShouldProcessReason throwAwayProcessReason;
                        cmdlet.ShouldProcess(
                            verboseDescription,
                            verboseWarning,
                            caption,
                            out throwAwayProcessReason);
                    });
        }

        internal virtual bool ShouldProcess(
            string verboseDescription,
            string verboseWarning,
            string caption,
            out ShouldProcessReason shouldProcessReason,
            out Exception exceptionThrownOnCmdletThread)
        {
            ShouldProcessReason closureSafeShouldProcessReason = ShouldProcessReason.None;

            bool methodResult = InvokeCmdletMethodAndWaitForResults(
                cmdlet => cmdlet.ShouldProcess(
                    verboseDescription,
                    verboseWarning,
                    caption,
                    out closureSafeShouldProcessReason),
                out exceptionThrownOnCmdletThread);

            shouldProcessReason = closureSafeShouldProcessReason;
            return methodResult;
        }

        private void InvokeCmdletMethodAndIgnoreResults(Action<Cmdlet> invokeCmdletMethod)
        {
            object resultsLock = new object();
            CmdletMethodInvoker<object> methodInvoker = new CmdletMethodInvoker<object>
            {
                Action = delegate (Cmdlet cmdlet) { invokeCmdletMethod(cmdlet); return null; },
                Finished = null,
                SyncObject = resultsLock
            };
            Results.Add(new PSStreamObject(PSStreamObjectType.BlockingError, methodInvoker));
        }

        private T InvokeCmdletMethodAndWaitForResults<T>(Func<Cmdlet, T> invokeCmdletMethodAndReturnResult, out Exception exceptionThrownOnCmdletThread)
        {
            Dbg.Assert(invokeCmdletMethodAndReturnResult != null, "Caller should verify invokeCmdletMethodAndReturnResult != null");

            T methodResult = default(T);
            Exception closureSafeExceptionThrownOnCmdletThread = null;
            object resultsLock = new object();
            using (var gotResultEvent = new ManualResetEventSlim(false))
            {
                EventHandler<JobStateEventArgs> stateChangedEventHandler =
                    delegate (object sender, JobStateEventArgs eventArgs)
                        {
                            if (IsFinishedState(eventArgs.JobStateInfo.State) || eventArgs.JobStateInfo.State == JobState.Stopping)
                            {
                                lock (resultsLock)
                                {
                                    closureSafeExceptionThrownOnCmdletThread = new OperationCanceledException();
                                }
                                gotResultEvent.Set();
                            }
                        };
                this.StateChanged += stateChangedEventHandler;
                Interlocked.MemoryBarrier();
                try
                {
                    stateChangedEventHandler(null, new JobStateEventArgs(this.JobStateInfo));

                    if (!gotResultEvent.IsSet)
                    {
                        this.SetJobState(JobState.Blocked);

                        // addition to results column happens here
                        CmdletMethodInvoker<T> methodInvoker = new CmdletMethodInvoker<T>
                        {
                            Action = invokeCmdletMethodAndReturnResult,
                            Finished = gotResultEvent,
                            SyncObject = resultsLock
                        };
                        PSStreamObjectType objectType = PSStreamObjectType.ShouldMethod;

                        if (typeof(T) == typeof(object))
                            objectType = PSStreamObjectType.BlockingError;

                        Results.Add(new PSStreamObject(objectType, methodInvoker));

                        gotResultEvent.Wait();
                        this.SetJobState(JobState.Running);

                        lock (resultsLock)
                        {
                            if (closureSafeExceptionThrownOnCmdletThread == null) // stateChangedEventHandler didn't set the results?  = ok to clobber results?
                            {
                                closureSafeExceptionThrownOnCmdletThread = methodInvoker.ExceptionThrownOnCmdletThread;
                                methodResult = methodInvoker.MethodResult;
                            }
                        }
                    }
                }
                finally
                {
                    this.StateChanged -= stateChangedEventHandler;
                }
            }

            lock (resultsLock)
            {
                exceptionThrownOnCmdletThread = closureSafeExceptionThrownOnCmdletThread;
                return methodResult;
            }
        }

        internal virtual void ForwardAvailableResultsToCmdlet(Cmdlet cmdlet)
        {
            foreach (PSStreamObject obj in Results.ReadAll())
            {
                obj.WriteStreamObject(cmdlet);
            }
        }

        internal virtual void ForwardAllResultsToCmdlet(Cmdlet cmdlet)
        {
            foreach (PSStreamObject obj in this.Results)
            {
                obj.WriteStreamObject(cmdlet);
            }
        }

        /// <summary>
        /// This method is introduce for delaying the loading of streams
        /// for a particular job.
        /// </summary>
        protected virtual void DoLoadJobStreams()
        {
        }

        /// <summary>
        /// Unloads job streams information. Enables jobs to
        /// clear stream information from memory
        /// </summary>
        protected virtual void DoUnloadJobStreams()
        {
        }

        /// <summary>
        /// Load the required job streams
        /// </summary>
        public void LoadJobStreams()