ReceiveJob.cs

/********************************************************************++
Copyright (c) Microsoft Corporation.  All rights reserved.
--********************************************************************/

using System;
using System.Collections;
using System.Collections.ObjectModel;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Management.Automation;
using System.Management.Automation.Remoting;
using System.Management.Automation.Remoting.Internal;
using System.Management.Automation.Runspaces;
using System.Management.Automation.Tracing;
using System.Threading;
using Dbg = System.Management.Automation.Diagnostics;
using System.Management.Automation.Internal;

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

namespace Microsoft.PowerShell.Commands
{
    /// <summary>
    /// Cmdlet used for receiveing results from job object. 
    /// This cmdlet is intended to have a slightly different behavior
    /// in the following two cases:
    ///          1. The job object to receive results from is a PSRemotingJob
    ///               In this case, the cmdlet can use two additional 
    ///               parameters to filter results - ComputerName and Runspace
    ///               The parameters help filter out results for a specified
    ///               computer or runspace from the job object
    /// 
    ///               $job = Start-PSJob -Command 'get-process' -ComputerName server1, server2    
    ///               Receive-PSJob -Job $job -ComputerName server1
    /// 
    ///               $job = Start-PSJob -Command 'get-process' -Session $r1, $r2
    ///               Receive-PSJob -Job $job -Session $r1
    /// 
    ///         2. The job object to receive results is a PSJob (or derivative 
    ///            other than PSRemotingJob)
    ///              In this case, the user cannot will use the location parameter
    ///              to do any filtering and will not have ComputerName and Runspace
    ///              parameters
    /// 
    ///              $job = Get-WMIObject '....' -AsJob
    ///              Receive-PSJob -Job $job -Location "Server2"
    /// 
    ///              The following will result in an error:
    /// 
    ///              $job = Get-WMIObject '....' -AsJob
    ///              Receive-PSJob -Job $job -ComputerName "Server2"
    ///              The parameter ComputerName cannot be used with jobs which are
    ///              not PSRemotingJob
    /// 
    /// </summary>
    [Cmdlet("Receive", "Job", DefaultParameterSetName = ReceiveJobCommand.LocationParameterSet,
        HelpUri = "http://go.microsoft.com/fwlink/?LinkID=113372", RemotingCapability = RemotingCapability.SupportedByCommand)]
    public class ReceiveJobCommand : JobCmdletBase, IDisposable
    {
        #region Properties

        /// <summary>
        /// Job object from which specific results need to 
        /// be extracted
        /// </summary>
        [Parameter(Position = 0,
                   Mandatory = true,
                   ValueFromPipeline = true,
                   ValueFromPipelineByPropertyName = true,
                   ParameterSetName = ReceiveJobCommand.ComputerNameParameterSet)]
        [Parameter(Position = 0,
                   Mandatory = true,
                   ValueFromPipeline = true,
                   ValueFromPipelineByPropertyName = true,
                   ParameterSetName = ReceiveJobCommand.SessionParameterSet)]
        [Parameter(Position = 0,
                   Mandatory = true,
                   ValueFromPipeline = true,
                   ValueFromPipelineByPropertyName = true,
                   ParameterSetName = ReceiveJobCommand.LocationParameterSet)]
        [SuppressMessage("Microsoft.Performance", "CA1819:PropertiesShouldNotReturnArrays")]
        public Job[] Job
        {
            get
            {
                return _jobs;
            }
            set
            {
                _jobs = value;
            }
        }
        private Job[] _jobs;

        /// <summary>
        /// Name of the computer for which the results needs to be
        /// returned
        /// </summary>
        [Parameter(ValueFromPipelineByPropertyName = true,
                   ParameterSetName = ReceiveJobCommand.ComputerNameParameterSet,
                   Position = 1)]
        [Alias("Cn")]
        [SuppressMessage("Microsoft.Performance", "CA1819:PropertiesShouldNotReturnArrays")]
        [ValidateNotNullOrEmpty]
        public String[] ComputerName
        {
            get
            {
                return _computerNames;
            }
            set
            {
                _computerNames = value;
            }
        }
        private String[] _computerNames;

        /// <summary>
        /// Locations for which the results needs to be returned.
        /// This will cater to all kinds of jobs and not only
        /// remoting jobs
        /// </summary>
        [Parameter(ParameterSetName = ReceiveJobCommand.LocationParameterSet,
                   Position = 1)]
        [ValidateNotNullOrEmpty]
        [SuppressMessage("Microsoft.Performance", "CA1819:PropertiesShouldNotReturnArrays")]
        public String[] Location
        {
            get
            {
                return _locations;
            }
            set
            {
                _locations = value;
            }
        }
        private String[] _locations;

        /// <summary>
        /// Runspaces for which the results needs to be
        /// returned
        /// </summary>
        [Parameter(ValueFromPipelineByPropertyName = true,
                   ParameterSetName = ReceiveJobCommand.SessionParameterSet,
                   Position = 1)]
        [ValidateNotNull]
        [SuppressMessage("Microsoft.Performance", "CA1819:PropertiesShouldNotReturnArrays")]
        public PSSession[] Session
        {
            get
            {
                return _remoteRunspaceInfos;
            }
            set
            {
                _remoteRunspaceInfos = value;
            }
        }
        private PSSession[] _remoteRunspaceInfos;

        /// <summary>
        /// If the results need to be not removed from the store
        /// after being written. Default is results are removed.
        /// </summary>
        [Parameter()]
        public SwitchParameter Keep
        {
            get
            {
                return !_flush;
            }
            set
            {
                _flush = !value;
                ValidateWait();
            }
        }
        private bool _flush = true;

        /// <summary>
        /// 
        /// </summary>
        [Parameter()]
        public SwitchParameter NoRecurse
        {
            get
            {
                return !_recurse;
            }
            set
            {
                _recurse = !value;
            }
        }
        private bool _recurse = true;

        /// <summary>
        /// 
        /// </summary>
        [Parameter()]
        public SwitchParameter Force
        { get; set; }

        /// <summary>
        /// 
        /// </summary>
        public override JobState State
        {
            get
            {
                return JobState.NotStarted;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        public override Hashtable Filter
        {
            get { return null; }
        }

        /// <summary>
        /// 
        /// </summary>
        public override string[] Command
        {
            get
            {
                return null;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        protected const string LocationParameterSet = "Location";

        /// <summary>
        /// 
        /// </summary>
        [Parameter()]
        public SwitchParameter Wait
        {
            get
            {
                return _wait;
            }
            set
            {
                _wait = value;
                ValidateWait();
            }
        }

        /// <summary>
        /// 
        /// </summary>
        [Parameter()]
        public SwitchParameter AutoRemoveJob
        {
            get
            {
                return _autoRemoveJob;
            }
            set
            {
                _autoRemoveJob = value;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        [Parameter()]
        public SwitchParameter WriteEvents
        {
            get { return _writeStateChangedEvents; }
            set
            {
                _writeStateChangedEvents = value;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        [Parameter()]
        public SwitchParameter WriteJobInResults
        {
            get { return _outputJobFirst; }
            set
            {
                _outputJobFirst = value;
            }
        }

        private bool _autoRemoveJob;
        private bool _writeStateChangedEvents;
        private bool _wait;
        private bool _isStopping;
        private bool _isDisposed;
        private readonly ReaderWriterLockSlim _resultsReaderWriterLock = new ReaderWriterLockSlim();
        private readonly PowerShellTraceSource _tracer = PowerShellTraceSourceFactory.GetTraceSource();
        private readonly ManualResetEvent _writeExistingData = new ManualResetEvent(true);
        private readonly PSDataCollection<PSStreamObject> _results = new PSDataCollection<PSStreamObject>();
        private bool _holdingResultsRef;
        private readonly List<Job> _jobsBeingAggregated = new List<Job>();
        private readonly List<Guid> _jobsSpecifiedInParameters = new List<Guid>();
        private readonly object _syncObject = new object();
        private bool _outputJobFirst;
        private OutputProcessingState _outputProcessingNotification;
        private bool _processingOutput;

        private const string ClassNameTrace = "ReceiveJobCommand";

        #endregion Properties

        #region Overrides

        /// <summary>
        /// 
        /// </summary>
        protected override void BeginProcessing()
        {
            ValidateAutoRemove();
            ValidateWriteJobInResults();
            ValidateWriteEvents();
            ValidateForce();
        }

        /// <summary>
        /// Retrieve the results for the specified computers or
        /// runspaces 
        /// </summary>
        protected override void ProcessRecord()
        {
            bool checkForRecurse = false;
            List<Job> jobsToWrite = new List<Job>();

            switch (ParameterSetName)
            {
                case SessionParameterSet:
                    {
                        foreach (Job job in _jobs)
                        {
                            PSRemotingJob remoteJob =
                                        job as PSRemotingJob;

                            if (remoteJob == null)
                            {
                                String message = GetMessage(RemotingErrorIdStrings.RunspaceParamNotSupported);

                                WriteError(new ErrorRecord(new ArgumentException(message),
                                    "RunspaceParameterNotSupported", ErrorCategory.InvalidArgument,
                                        job));

                                continue;
                            }

                            //Runspace parameter is supported only on PSRemotingJob objects
                            foreach (PSSession remoteRunspaceInfo in _remoteRunspaceInfos)
                            {
                                // get the required child jobs
                                List<Job> childJobs = remoteJob.GetJobsForRunspace(remoteRunspaceInfo);
                                jobsToWrite.AddRange(childJobs);
                                //WriteResultsForJobsInCollection(childJobs, false);

                            } // foreach(RemoteRunspaceInfo...
                        } // foreach ...                        
                    }
                    break;

                case ComputerNameParameterSet:
                    {
                        foreach (Job job in _jobs)
                        {
                            // the job can either be a remoting job or another one
                            PSRemotingJob remoteJob =
                                        job as PSRemotingJob;

                            // ComputerName parameter can only be used with remoting jobs
                            if (remoteJob == null)
                            {
                                String message = GetMessage(RemotingErrorIdStrings.ComputerNameParamNotSupported);

                                WriteError(new ErrorRecord(new ArgumentException(message),
                                    "ComputerNameParameterNotSupported", ErrorCategory.InvalidArgument,
                                        job));

                                continue;
                            }

                            String[] resolvedComputernames = null;
                            ResolveComputerNames(_computerNames, out resolvedComputernames);

                            foreach (String resolvedComputerName in resolvedComputernames)
                            {
                                // get the required child Job objects
                                List<Job> childJobs = remoteJob.GetJobsForComputer(resolvedComputerName);
                                jobsToWrite.AddRange(childJobs);
                                //WriteResultsForJobsInCollection(childJobs, false);

                            } // foreach (String...
                        } // foreach ...                            
                    }
                    break;

                case "Location":
                    {
                        if (_locations == null)
                        {
                            //WriteAll();
                            jobsToWrite.AddRange(_jobs);
                            checkForRecurse = true;
                        }
                        else
                        {
                            foreach (Job job in _jobs)
                            {
                                foreach (String location in _locations)
                                {
                                    // get the required child Job objects
                                    List<Job> childJobs = job.GetJobsForLocation(location);
                                    jobsToWrite.AddRange(childJobs);
                                    //WriteResultsForJobsInCollection(childJobs, false);
                                } // foreach (String...
                            } // foreach ...
                        }
                    }
                    break;

                case ReceiveJobCommand.InstanceIdParameterSet:
                    {
                        List<Job> jobs = FindJobsMatchingByInstanceId(true, false, true, false);

                        jobsToWrite.AddRange(jobs);
                        checkForRecurse = true;
                        //WriteResultsForJobsInCollection(jobs, true);
                    }
                    break;

                case ReceiveJobCommand.SessionIdParameterSet:
                    {
                        List<Job> jobs = FindJobsMatchingBySessionId(true, false, true, false);
                        jobsToWrite.AddRange(jobs);
                        checkForRecurse = true;
                        //WriteResultsForJobsInCollection(jobs, true);
                    }
                    break;

                case ReceiveJobCommand.NameParameterSet:
                    {
                        List<Job> jobs = FindJobsMatchingByName(true, false, true, false);
                        jobsToWrite.AddRange(jobs);
                        checkForRecurse = true;
                        //WriteResultsForJobsInCollection(jobs, true);
                    }
                    break;
            }

            // if block has been specified and the cmdlet has not been
            // stopped, we continue to write recursively, until there
            // is no more data to write
            if (_wait)
            {
                _writeExistingData.Reset();

                // if writejobresults is specified we will write only the top level jobs
                // this is because that is what the proxy requires. Anything else being
                // written is useless and will only add weight to the serialization

                WriteJobsIfRequired(jobsToWrite);

                // Make a note of the jobs specified by the user (does not include child jobs)
                // for the purpose of removal. Only the parent jobs should have remove called.
                foreach (var job in jobsToWrite)
                {
                    _jobsSpecifiedInParameters.Add(job.InstanceId);
                }

                lock (_syncObject)
                {
                    if (_isDisposed || _isStopping) return;

                    // Check to see that we only AddRef once. ProcessRecord is called
                    // once per job on the pipeline.
                    if (!_holdingResultsRef)
                    {
                        _tracer.WriteMessage(ClassNameTrace, "ProcessRecord", Guid.Empty, (Job)null, "Adding Ref to results collection",
                         null);
                        _results.AddRef();
                        _holdingResultsRef = true;
                    }
                }

                _tracer.WriteMessage(ClassNameTrace, "ProcessRecord", Guid.Empty, (Job)null, "BEGIN Register for jobs");
                WriteResultsForJobsInCollection(jobsToWrite, checkForRecurse, true);
                _tracer.WriteMessage(ClassNameTrace, "ProcessRecord", Guid.Empty, (Job)null, "END Register for jobs");

                lock (_syncObject)
                {
                    if (_jobsBeingAggregated.Count == 0 && _holdingResultsRef)
                    {
                        _tracer.WriteMessage(ClassNameTrace, "ProcessRecord", Guid.Empty, (Job)null,
                                             "Removing Ref to results collection", null);
                        _results.DecrementRef();
                        _holdingResultsRef = false;
                    }
                }

                _tracer.WriteMessage(ClassNameTrace, "ProcessRecord", Guid.Empty, (Job)null, "BEGIN Write existing job data");
                WriteResultsForJobsInCollection(jobsToWrite, checkForRecurse, false);
                _tracer.WriteMessage(ClassNameTrace, "ProcessRecord", Guid.Empty, (Job)null, "END Write existing job data");
                _writeExistingData.Set();
            }
            else
            {
                WriteResultsForJobsInCollection(jobsToWrite, checkForRecurse, false);
            }
        } // ProcessRecord

        /// <summary>
        /// StopProcessing - when the command is stopped,
        /// unregister all the event handlers from the jobs
        /// and decrement reference for results
        /// </summary>
        protected override void StopProcessing()
        {
            _tracer.WriteMessage(ClassNameTrace, "StopProcessing", Guid.Empty, (Job)null, "Entered Stop Processing",
                     null);
            lock (_syncObject)
            {
                _isStopping = true;
            }

            _writeExistingData.Set();
            Job[] aggregatedJobs = new Job[_jobsBeingAggregated.Count];

            for (int i = 0; i < _jobsBeingAggregated.Count; i++)
            {
                aggregatedJobs[i] = _jobsBeingAggregated[i];
            }

            foreach (Job job in aggregatedJobs)
            {
                StopAggregateResultsFromJob(job);
            }

            _resultsReaderWriterLock.EnterWriteLock();
            try
            {
                _results.Complete();
                SetOutputProcessingState(false);
            }
            finally
            {
                _resultsReaderWriterLock.ExitWriteLock();
            }

            base.StopProcessing();
            _tracer.WriteMessage(ClassNameTrace, "StopProcessing", Guid.Empty, (Job)null, "Exiting Stop Processing",
                     null);
        }

        /// <summary>
        /// if we are not stopping, continue writing output
        /// as and when they are available
        /// </summary>
        protected override void EndProcessing()
        {
            try
            {
                if (_wait)
                {
                    int totalCount = 0;
                    foreach (PSStreamObject result in _results)
                    {
                        if (_isStopping) break;

                        SetOutputProcessingState(true);
                        result.WriteStreamObject(this, true, true);
                        if (++totalCount == _results.Count)
                        {
                            SetOutputProcessingState(false);
                        }
                    }

                    _eventArgsWritten.Clear();
                }
                else
                {
                    int totalCount = 0;
                    foreach (PSStreamObject result in _results)
                    {
                        if (_isStopping) break;

                        SetOutputProcessingState(true);
                        result.WriteStreamObject(this, false, true);
                        if (++totalCount == _results.Count)
                        {
                            SetOutputProcessingState(false);
                        }
                    }
                }
            }
            finally
            {
                SetOutputProcessingState(false);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="disposing"></param>
        protected void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (_isDisposed) return;
                lock (_syncObject)
                {
                    if (_isDisposed) return;
                    _isDisposed = true;
                }

                SetOutputProcessingState(false);

                if (_jobsBeingAggregated != null)
                {
                    foreach (var job in _jobsBeingAggregated)
                    {
                        if (job.MonitorOutputProcessing)
                        {
                            job.RemoveMonitorOutputProcessing(_outputProcessingNotification);
                        }

                        if (job.UsesResultsCollection)
                        {
                            job.Results.DataAdded -= ResultsAdded;
                        }
                        else
                        {
                            job.Output.DataAdded -= Output_DataAdded;
                            job.Error.DataAdded -= Error_DataAdded;
                            job.Progress.DataAdded -= Progress_DataAdded;
                            job.Verbose.DataAdded -= Verbose_DataAdded;
                            job.Warning.DataAdded -= Warning_DataAdded;
                            job.Debug.DataAdded -= Debug_DataAdded;
                            job.Information.DataAdded -= Information_DataAdded;
                        }
                        job.StateChanged -= HandleJobStateChanged;
                    }
                }

                _resultsReaderWriterLock.EnterWriteLock();
                try
                {
                    _results.Complete();
                }
                finally
                {
                    _resultsReaderWriterLock.ExitWriteLock();
                }

                _resultsReaderWriterLock.Dispose();
                _results.Clear();
                _results.Dispose();
                _writeExistingData.Set();
                _writeExistingData.Dispose();
            }
        }
        #endregion Overrides

        #region Private Methods

        private static void DoUnblockJob(Job job)
        {
            // we should not do anything for a parent job
            // the assumption is parent job states are
            // computed and so unblocking the child state
            // should be able to handle this
            if (job.ChildJobs.Count != 0) return;

            // we have a better way of handling blocked state logic
            // for remoting jobs, so use that if job is a remoting
            // job
            PSRemotingChildJob remotingChildJob = job as PSRemotingChildJob;
            if (remotingChildJob != null)
            {
                remotingChildJob.UnblockJob();
            }
            else
            {
                // for all other job types, simply set the job state
                // to running, the handling of the parent jobs state
                // should be taken care of by the job implementation
                job.SetJobState(JobState.Running, null);
            }
        }

        /// <summary>
        /// Write the results from this Job object. This does not write from the
        /// child jobs of this job object
        /// </summary>
        /// <param name="job">Job object from which to write the results from
        /// </param>
        /// 
        private void WriteJobResults(Job job)
        {
            if (job == null) return;

            // Q: Why do we need to unblock the job, before getting
            // the results
            // A: The job can get into a terminal state and we do
            // not want to set it to running at that point. Also, if
            // we do not explicitly signal that the job is unblocked
            // then the parent job cannot be unblocked. This is because
            // the parent job does not maintain a list of jobs which
            // are blocked but just simply a count (to keep things
            // light weight)

            // check if the state of the job is blocked, if so unblock it

            // Skip disconnected jobs that were in Blocked state before
            // the disconnect, since we cannot process host data until the 
            // job is re-connected.
            if (job.JobStateInfo.State == JobState.Disconnected)
            {
                PSRemotingChildJob remotingChildJob = job as PSRemotingChildJob;
                if (remotingChildJob != null && remotingChildJob.DisconnectedAndBlocked)
                {
                    return;
                }
            }

            // TODO: Fix Unblock() handling by Job2
            if (job.JobStateInfo.State == JobState.Blocked)
            {
                DoUnblockJob(job);
            }

            // for the jobs that PowerShell writes, there is a 
            // results collection internally used. This collection
            // can be used to write results. For all other jobs
            // results need to be written from the other collections
            // available. 
            // There is a bug in V2 that only remoting jobs work
            // with Receive-Job. This is being fixed

            if (!(job is Job2) && job.UsesResultsCollection)
            {
                // extract results and handle them
                Collection<PSStreamObject> results = ReadAll<PSStreamObject>(job.Results);

                if (_wait)
                {
                    foreach (var psStreamObject in results)
                    {
                        psStreamObject.WriteStreamObject(this, job.Results.SourceId);
                    }
                }
                else
                {
                    foreach (var psStreamObject in results)
                    {
                        psStreamObject.WriteStreamObject(this);
                    }
                }
            }
            else
            {
                Collection<PSObject> output = ReadAll<PSObject>(job.Output);

                foreach (PSObject o in output)
                {
                    if (o == null) continue;
                    WriteObject(o);
                }

                Collection<ErrorRecord> errorRecords = ReadAll<ErrorRecord>(job.Error);

                foreach (ErrorRecord e in errorRecords)
                {
                    if (e == null) continue;
                    MshCommandRuntime mshCommandRuntime = CommandRuntime as MshCommandRuntime;
                    if (mshCommandRuntime != null)
                    {
                        e.PreserveInvocationInfoOnce = true;
                        mshCommandRuntime.WriteError(e, true);
                    }
                }

                Collection<VerboseRecord> verboseRecords = ReadAll(job.Verbose);

                foreach (VerboseRecord v in verboseRecords)
                {
                    if (v == null) continue;
                    MshCommandRuntime mshCommandRuntime = CommandRuntime as MshCommandRuntime;
                    if (mshCommandRuntime != null)
                    {
                        mshCommandRuntime.WriteVerbose(v, true);
                    }
                }

                Collection<DebugRecord> debugRecords = ReadAll(job.Debug);

                foreach (DebugRecord d in debugRecords)
                {
                    if (d == null) continue;
                    MshCommandRuntime mshCommandRuntime = CommandRuntime as MshCommandRuntime;
                    if (mshCommandRuntime != null)
                    {
                        mshCommandRuntime.WriteDebug(d, true);
                    }
                }

                Collection<WarningRecord> warningRecords = ReadAll(job.Warning);

                foreach (WarningRecord w in warningRecords)
                {
                    if (w == null) continue;
                    MshCommandRuntime mshCommandRuntime = CommandRuntime as MshCommandRuntime;
                    if (mshCommandRuntime != null)
                    {
                        mshCommandRuntime.WriteWarning(w, true);
                    }
                }

                Collection<ProgressRecord> progressRecords = ReadAll(job.Progress);

                foreach (ProgressRecord p in progressRecords)
                {
                    if (p == null) continue;
                    MshCommandRuntime mshCommandRuntime = CommandRuntime as MshCommandRuntime;
                    if (mshCommandRuntime != null)
                    {
                        mshCommandRuntime.WriteProgress(p, true);
                    }
                }

                Collection<InformationRecord> informationRecords = ReadAll(job.Information);

                foreach (InformationRecord p in informationRecords)
                {
                    if (p == null) continue;
                    MshCommandRuntime mshCommandRuntime = CommandRuntime as MshCommandRuntime;
                    if (mshCommandRuntime != null)
                    {
                        mshCommandRuntime.WriteInformation(p, true);
                    }
                }
            }

            if (job.JobStateInfo.State != JobState.Failed) return;

            WriteReasonError(job);
        }

        private void WriteReasonError(Job job)
        {
            //Write better error for the remoting case and generic error for the other case
            PSRemotingChildJob child = job as PSRemotingChildJob;
            if (child != null && child.FailureErrorRecord != null)
            {
                _results.Add(new PSStreamObject(PSStreamObjectType.Error, child.FailureErrorRecord, child.InstanceId));
            }
            else if (job.JobStateInfo.Reason != null)
            {
                Exception baseReason = job.JobStateInfo.Reason;
                Exception resultReason = baseReason;

                // If it was generated by a job that gave location information, unpack the
                // base exception.
                JobFailedException exceptionWithLocation = baseReason as JobFailedException;
                if (exceptionWithLocation != null)
                {
                    resultReason = exceptionWithLocation.Reason;
                }

                ErrorRecord errorRecord = new ErrorRecord(resultReason, "JobStateFailed", ErrorCategory.InvalidResult, null);

                // If it was generated by a job that gave location information, set the
                // location information.
                if ((exceptionWithLocation != null) && (exceptionWithLocation.DisplayScriptPosition != null))
                {
                    if (errorRecord.InvocationInfo == null)
                    {
                        errorRecord.SetInvocationInfo(new InvocationInfo(null, null));
                    }

                    errorRecord.InvocationInfo.DisplayScriptPosition = exceptionWithLocation.DisplayScriptPosition;
                }

                _results.Add(new PSStreamObject(PSStreamObjectType.Error, errorRecord, job.InstanceId));
            }
        }

        /// <summary>
        /// Returns all the results from supplied PSDataCollection.
        /// </summary>
        /// <param name="psDataCollection">data collection to read from</param>
        /// <returns>collection with copy of data</returns>
        private Collection<T> ReadAll<T>(PSDataCollection<T> psDataCollection)
        {
            if (_flush)
            {
                return psDataCollection.ReadAll();
            }

            T[] array = new T[psDataCollection.Count];
            psDataCollection.CopyTo(array, 0);
            Collection<T> collection = new Collection<T>();
            foreach (T t in array)
            {
                collection.Add(t);
            }
            return collection;
        }

        /// <summary>
        /// Write the results from this Job object. It also writes the 
        /// results from its child objects recursively.
        /// </summary>
        /// 
        /// <param name="duplicate">Hashtable used for duplicate detection</param>
        /// <param name="job">Job whose results are written</param>
        /// <param name="registerInsteadOfWrite"></param>
        private void WriteJobResultsRecursivelyHelper(Hashtable duplicate, Job job, bool registerInsteadOfWrite)
        {
            //Check if this object is already visited. If not, add it to the cache
            if (duplicate.ContainsKey(job))
            {
                return;
            }
            duplicate.Add(job, job);

            //Write the results of child jobs
            IList<Job> childJobs = job.ChildJobs;

            foreach (Job childjob in childJobs)
            {
                WriteJobResultsRecursivelyHelper(duplicate, childjob, registerInsteadOfWrite);
            }

            if (registerInsteadOfWrite)
            {
                // at any point there will be only one thread which will have
                // access to an entry corresponding to a job
                // this is because of the way the synchronization happens
                // with the pipeline thread and event handler thread using
                // _writeExistingData
                _eventArgsWritten[job.InstanceId] = false;
                // register the job for future updates
                AggregateResultsFromJob(job);
            }
            else
            {
                //Write the results of this job
                WriteJobResults(job);
                WriteJobStateInformationIfRequired(job);
            }
        } // WriteAllEntities

        /// <summary>
        /// Writes the job objects if required by the cmdlet
        /// </summary>
        /// <param name="jobsToWrite">collection of jobs to write</param>
        /// <remarks>this method is intended to be called only from 
        /// ProcessRecord. When any changes are made ensure that this
        /// contract is not broken</remarks>
        private void WriteJobsIfRequired(IEnumerable<Job> jobsToWrite)
        {
            if (!_outputJobFirst) return;

            foreach (var job in jobsToWrite)
            {
                _tracer.WriteMessage("ReceiveJobCommand", "WriteJobsIfRequired", Guid.Empty, job, "Writing job object as output", null);
                WriteObject(job);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="job"></param>
        /// <remarks>this method should always be called before 
        /// writeExistingData is set in ProcessRecord</remarks>