PSProxyJob.cs

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

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

    /// <summary>
    /// 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
    /// </summary>
    public sealed class PSJobProxy : Job2
    {
        #region Constructors

        /// <summary>
        /// Internal constructor
        /// </summary>
        /// <param name="command">the command to execute</param>
        internal PSJobProxy(string command)
            : base(command)
        {
            _tracer.WriteMessage(ClassNameTrace, "ctor", _remoteJobInstanceId, this,
                "Constructing proxy job", null);
            StateChanged += HandleMyStateChange;
        }

        #endregion Constructors

        #region Overrides of Job

        ///<summary>
        /// Success status of the command execution.
        /// </summary>
        public override string StatusMessage
        {
            get { return _remoteJobStatusMessage; }
        }

        /// <summary>
        /// Indicates that more data is available in this
        /// result object for reading.
        /// </summary>
        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;
            }
        }

        /// <summary>
        /// This is the location string from the remote job.
        /// </summary>
        public override string Location
        {
            get { return _remoteJobLocation; }
        }

        #endregion

        #region Overrides of Job2

        /// <summary>
        /// start a job. The job will be started with the parameters
        /// specified in StartParameters
        /// </summary>
        /// <exception cref="InvalidJobStateException">Thrown if the job
        ///  is already running, if there is no runspace or runspace pool
        ///  assigned.</exception>
        /// <exception cref="NotSupportedException">Thrown if the job is
        ///  otherwise started, finished, or suspended.</exception>
        /// <exception cref="ObjectDisposedException">Thrown if job is
        /// disposed.</exception>
        public override void StartJob()
        {
            StartJob(null, null, null);
        }

        /// <summary>
        /// Start a job asynchronously
        /// </summary>
        /// <remarks>When a job is started all the data in the
        /// job streams from a previous invocation will be cleared</remarks>
        public override void StartJobAsync()
        {
            StartJobAsync(null, null, null);
        }

        /// <summary>
        /// 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
        /// </summary>
        /// <exception cref="InvalidJobStateException">Thrown if job is blocked.</exception>
        /// <exception cref="ObjectDisposedException">Thrown if job is disposed.</exception>
        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)
            {
                _tracer.WriteMessage(ClassNameTrace, "StopJob", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                throw;
            }
        }

        /// <summary>
        /// Stop a job asynchronously
        /// </summary>
        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.
                _tracer.WriteMessage(ClassNameTrace, "StopJobAsync", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                OnStopJobCompleted(new AsyncCompletedEventArgs(error, false, null));
            }
#pragma warning restore 56500
        }

        /// <summary>
        /// StopJob
        /// </summary>
        /// <param name="force"></param>
        /// <param name="reason"></param>
        public override void StopJob(bool force, string reason)
        {
            throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyJobControlNotSupported);
        }

        /// <summary>
        /// StopJobAsync
        /// </summary>
        /// <param name="force"></param>
        /// <param name="reason"></param>
        public override void StopJobAsync(bool force, string reason)
        {
            throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyJobControlNotSupported);
        }

        /// <summary>
        /// Suspend a job
        /// </summary>
        /// <exception cref="InvalidJobStateException">Throws if the job is not in
        /// a running or suspended state.</exception>
        /// <exception cref="ObjectDisposedException">Thrown if job is
        /// disposed.</exception>
        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)
            {
                _tracer.WriteMessage(ClassNameTrace, "SuspendJob", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                throw;
            }
        }

        /// <summary>
        /// Asynchronously suspend a job
        /// </summary>
        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.
                _tracer.WriteMessage(ClassNameTrace, "SuspendJobAsync", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                OnSuspendJobCompleted(new AsyncCompletedEventArgs(error, false, null));
            }
#pragma warning restore 56500
        }

        /// <summary>
        /// SuspendJob
        /// </summary>
        /// <param name="force"></param>
        /// <param name="reason"></param>
        public override void SuspendJob(bool force, string reason)
        {
            throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyJobControlNotSupported);
        }

        /// <summary>
        /// SuspendJobAsync
        /// </summary>
        /// <param name="force"></param>
        /// <param name="reason"></param>
        public override void SuspendJobAsync(bool force, string reason)
        {
            throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyJobControlNotSupported);
        }

        /// <summary>
        /// Resume a suspended job
        /// </summary>
        /// <exception cref="InvalidJobStateException">Throws if the job
        /// is not in a suspended or running state.</exception>
        /// <exception cref="ObjectDisposedException">Thrown if job is
        /// disposed.</exception>
        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)
            {
                _tracer.WriteMessage(ClassNameTrace, "ResumeJob", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                throw;
            }
        }

        /// <summary>
        /// Resume a suspended job asynchronously.
        /// </summary>
        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.
                _tracer.WriteMessage(ClassNameTrace, "ResumeJobAsync", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                OnResumeJobCompleted(new AsyncCompletedEventArgs(error, false, null));
            }
#pragma warning restore 56500
        }

        /// <summary>
        /// Unblock a blocked job
        /// </summary>
        /// <exception cref="NotSupportedException">Unblock job is not supported on PSJobProxy.</exception>
        public override void UnblockJob()
        {
            throw PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyUnblockJobNotSupported);
        }

        /// <summary>
        /// Unblock a blocked job asynchronously
        /// </summary>
        public override void UnblockJobAsync()
        {
            OnUnblockJobCompleted(new AsyncCompletedEventArgs(PSTraceSource.NewNotSupportedException(PowerShellStrings.ProxyUnblockJobNotSupported), false, null));
        }

        #endregion

        #region Additional State Management Methods

        /// <summary>
        /// Start execution of the workflow with the
        /// specified input. This input will serve as
        /// input to the underlying pipeline
        /// </summary>
        /// <param name="input">collection of input
        /// objects</param>
        public void StartJobAsync(PSDataCollection<object> input)
        {
            StartJobAsync(null, null, input);
        }

        /// <summary>
        /// Start execution of the job with the
        /// specified input. This input will serve as
        /// input to the underlying pipeline
        /// </summary>
        /// <param name="input"></param>
        /// <remarks>Not sure if this method is needed. This has
        /// been added just to be in sync with the PowerShell
        /// APIs</remarks>
        public void StartJob(PSDataCollection<object> input)
        {
            StartJob(null, null, input);
        }

        /// <summary>
        /// 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.
        /// </summary>
        /// <param name="dataAdded">delegate used to subscribe to data added events on the child jobs</param>
        /// <param name="stateChanged">delegate used to subscribe to state changed events on the child jobs</param>
        /// <param name="input">collection of input
        /// objects</param>
        public void StartJob(EventHandler<JobDataAddedEventArgs> dataAdded, EventHandler<JobStateEventArgs> stateChanged, PSDataCollection<object> 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);
                _jobInitializedWaitHandle.WaitOne();
            }
            catch (Exception error)
            {
                _tracer.WriteMessage(ClassNameTrace, "StartJob", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                throw;
            }
        }

        /// <summary>
        /// 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.
        /// </summary>
        /// <param name="dataAdded">delegate used to subscribe to data added events on the child jobs</param>
        /// <param name="stateChanged">delegate used to subscribe to state changed events on the child jobs</param>
        /// <param name="input">collection of input
        /// objects</param>
        public void StartJobAsync(EventHandler<JobDataAddedEventArgs> dataAdded, EventHandler<JobStateEventArgs> stateChanged, PSDataCollection<object> input)
        {
#pragma warning disable 56500
            try
            {
                DoStartAsync(dataAdded, stateChanged, input);
            }
            catch (Exception error)
            {
                // Exception transferred using event arguments.
                _tracer.WriteMessage(ClassNameTrace, "StartJobAsync", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                OnStartJobCompleted(new AsyncCompletedEventArgs(error, false, null));
            }
#pragma warning restore 56500
        }

        /// <summary>
        /// Removes the job.
        /// If remoteJob is true, the job output that has been transferred to this
        /// client object will be preserved.
        /// </summary>
        /// <param name="removeRemoteJob">Indicates whether the remove operation should
        /// be applied to the remote or local job.</param>
        /// <param name="force">Force will stop the job on the server before
        /// attempting removal. Default value is false.</param>
        /// <exception cref="InvalidOperationException">Thrown if the job is not in
        /// a completed state.</exception>
        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)
            {
                _tracer.WriteMessage(ClassNameTrace, "RemoveJob", _remoteJobInstanceId, this, "Error", null);
                _tracer.TraceException(error);
                throw;
            }
        }

        /// <summary>
        /// Removes the job.
        /// If remoteJob is true, the job output that has been transferred to this
        /// client object will be preserved.
        /// </summary>
        /// <param name="removeRemoteJob">Indicates whether the remove operation should
        /// be applied to the remote or local job.</param>
        /// <exception cref="InvalidOperationException">Thrown if the job is not in
        /// a completed state.</exception>
        public void RemoveJob(bool removeRemoteJob)
        {
            RemoveJob(removeRemoteJob, false);
        }

        /// <summary>
        /// Removes the job on the remote server.
        /// The job output that has been transferred to this client object will be
        /// preserved.
        /// </summary>
        /// <param name="removeRemoteJob">Indicates whether the remove operation should
        /// be applied to the remote or local job.</param>
        /// <param name="force">Force will stop the job on the server before
        /// attempting removal.</param>
        public void RemoveJobAsync(bool removeRemoteJob, bool force)
        {
            if (!removeRemoteJob)
            {
                Dispose();
                OnRemoveJobCompleted(new AsyncCompletedEventArgs(null, false, null));
                return;
            }

            var asyncOp = AsyncOperationManager.CreateOperation(force);
            var workerDelegate = new JobActionWorkerDelegate(JobActionWorker);
            workerDelegate.BeginInvoke(asyncOp, ActionType.Remove, null, null);
        }

        /// <summary>
        /// Removes the job on the remote server.
        /// The job output that has been transferred to this client object will be
        /// preserved.
        /// </summary>
        /// <param name="removeRemoteJob">Indicates whether the remove operation should
        /// be applied to the remote or local job.</param>
        public void RemoveJobAsync(bool removeRemoteJob)
        {
            RemoveJobAsync(removeRemoteJob, false);
        }

        /// <summary>
        /// This event should be raised whenever the asynchronous removal of
        /// a server side job is completed.
        /// </summary>
        public event EventHandler<AsyncCompletedEventArgs> RemoveJobCompleted;

        /// <summary>
        /// Method to raise the event when removing a
        /// server side job is completed
        /// </summary>
        /// <param name="eventArgs">argument describing
        /// an exception that is associated with the event</param>
        private void OnRemoveJobCompleted(AsyncCompletedEventArgs eventArgs)
        {
            // Make a temporary copy of the event to avoid possibility of
            // a race condition if the last subscriber unsubscribes
            // immediately after the null check and before the event is raised.
            EventHandler<AsyncCompletedEventArgs> 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
                _tracer.TraceException(exception);
            }
#pragma warning restore 56500
        }

        #endregion Additional State Management Methods

        #region Additional Properties

        /// <summary>
        /// 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.
        /// </summary>
        public bool RemoveRemoteJobOnCompletion
        {
            get
            {
                return _removeRemoteJobOnCompletion;
            }
            set
            {
                AssertChangesCanBeAccepted();
                _removeRemoteJobOnCompletion = value;
            }
        }
        private bool _removeRemoteJobOnCompletion;

        /// <summary>
        /// The instance ID of the remote job that this proxy interacts with.
        /// </summary>
        public Guid RemoteJobInstanceId
        {
            get { return _remoteJobInstanceId; }
        }

        /// <summary>
        /// Runspace in which this job will be executed
        /// </summary>
        /// <remarks>At any point of time only a runspace or a
        /// runspacepool may be specified</remarks>
        public Runspace Runspace
        {
            get
            {
                return _runspace;
            }
            set
            {
                if (value == null)
                {
                    throw PSTraceSource.NewArgumentNullException("value");
                }

                lock (SyncRoot)
                {
                    AssertChangesCanBeAccepted();
                    _runspacePool = null;
                    _runspace = value;
                }
            }
        }

        /// <summary>
        /// RunspacePool in which this job will be executed
        /// </summary>
        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

        /// <summary>
        /// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
        /// </summary>
        /// <param name="runspace">Runspace containing the jobs to base the proxy on.</param>
        /// <param name="filter">Hashtable to use for the Get-Job -filter command.</param>
        /// <param name="dataAdded">Handler to subscribe to any child job data added events.</param>
        /// <param name="stateChanged">Handler to subscribe to any child job state changed events.</param>
        /// <returns>A collection of job proxies that represent the jobs collected based on the filter.</returns>
        public static ICollection<PSJobProxy> Create(Runspace runspace, Hashtable filter,
            EventHandler<JobDataAddedEventArgs> dataAdded, EventHandler<JobStateEventArgs> stateChanged)
        {
            return Create(runspace, filter, dataAdded, stateChanged, true);
        }

        /// <summary>
        /// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
        /// </summary>
        /// <param name="runspace">Runspace containing the jobs to base the proxy on.</param>
        /// <param name="filter">Hashtable to use for the Get-Job -filter command.</param>
        /// <param name="receiveImmediately">If true, the data streaming will start immediately. If false,
        /// the user must call "ReceiveJob()" to start data streaming.</param>
        /// <returns>A collection of job proxies that represent the jobs collected based on the filter.</returns>
        public static ICollection<PSJobProxy> Create(Runspace runspace, Hashtable filter, bool receiveImmediately)
        {
            return Create(runspace, filter, null, null, receiveImmediately);
        }

        /// <summary>
        /// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
        /// </summary>
        /// <param name="runspace">Runspace containing the jobs to base the proxy on.</param>
        /// <param name="filter">Hashtable to use for the Get-Job -filter command.</param>
        /// <returns>A collection of job proxies that represent the jobs collected based on the filter.</returns>
        public static ICollection<PSJobProxy> Create(Runspace runspace, Hashtable filter)
        {
            return Create(runspace, filter, null, null, true);
        }

        /// <summary>
        /// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
        /// </summary>
        /// <param name="runspace">Runspace containing the jobs to base the proxy on.</param>
        /// <returns>A collection of job proxies that represent the jobs collected based on the filter.</returns>
        public static ICollection<PSJobProxy> Create(Runspace runspace)
        {
            return Create(runspace, null, null, null, true);
        }

        /// <summary>
        /// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
        /// </summary>
        /// <param name="runspacePool">RunspacePool containing the jobs to base the proxy on.</param>
        /// <param name="filter">Hashtable to use for the Get-Job -filter command.</param>
        /// <param name="dataAdded">Handler to subscribe to any child job data added events.</param>
        /// <param name="stateChanged">Handler to subscribe to any child job state changed events.</param>
        /// <returns>A collection of job proxies that represent the jobs collected based on the filter.</returns>
        public static ICollection<PSJobProxy> Create(RunspacePool runspacePool, Hashtable filter,
            EventHandler<JobDataAddedEventArgs> dataAdded, EventHandler<JobStateEventArgs> stateChanged)
        {
            return Create(runspacePool, filter, dataAdded, stateChanged, true);
        }

        /// <summary>
        /// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
        /// </summary>
        /// <param name="runspacePool">RunspacePool containing the jobs to base the proxy on.</param>
        /// <param name="filter">Hashtable to use for the Get-Job -filter command.</param>
        /// <param name="receiveImmediately">If true, the data streaming will start immediately. If false,
        /// the user must call "ReceiveJob()" to start data streaming.</param>
        /// <returns>A collection of job proxies that represent the jobs collected based on the filter.</returns>
        public static ICollection<PSJobProxy> Create(RunspacePool runspacePool, Hashtable filter, bool receiveImmediately)
        {
            return Create(runspacePool, filter, null, null, receiveImmediately);
        }

        /// <summary>
        /// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
        /// </summary>
        /// <param name="runspacePool">RunspacePool containing the jobs to base the proxy on.</param>
        /// <param name="filter">Hashtable to use for the Get-Job -filter command.</param>
        /// <returns>A collection of job proxies that represent the jobs collected based on the filter.</returns>
        public static ICollection<PSJobProxy> Create(RunspacePool runspacePool, Hashtable filter)
        {
            return Create(runspacePool, filter, null, null, true);
        }

        /// <summary>
        /// Queries the runspace for jobs and constructs a collection of job proxies to interact with them.
        /// </summary>
        /// <param name="runspacePool">RunspacePool containing the jobs to base the proxy on.</param>
        /// <returns>A collection of job proxies that represent the jobs collected based on the filter.</returns>
        public static ICollection<PSJobProxy> Create(RunspacePool runspacePool)
        {
            return Create(runspacePool, null, null, null, true);
        }

        private static ICollection<PSJobProxy> Create(RunspacePool runspacePool, Hashtable filter,
            EventHandler<JobDataAddedEventArgs> dataAdded, EventHandler<JobStateEventArgs> stateChanged, bool connectImmediately)
        {
            if (runspacePool == null)
                throw new PSArgumentNullException("runspacePool");
            return Create(null, runspacePool, filter, dataAdded, stateChanged, connectImmediately);
        }

        private static ICollection<PSJobProxy> Create(Runspace runspace, Hashtable filter,
            EventHandler<JobDataAddedEventArgs> dataAdded, EventHandler<JobStateEventArgs> stateChanged, bool connectImmediately)
        {
            if (runspace == null)
                throw new PSArgumentNullException("runspace");
            return Create(runspace, null, filter, dataAdded, stateChanged, connectImmediately);
        }

        private static ICollection<PSJobProxy> Create(Runspace runspace, RunspacePool runspacePool, Hashtable filter,
            EventHandler<JobDataAddedEventArgs> dataAdded, EventHandler<JobStateEventArgs> stateChanged, bool connectImmediately)
        {
            // run command
            Collection<PSObject> jobResults;

            using (PowerShell ps = PowerShell.Create())
            {
                Dbg.Assert(runspacePool == null ^ runspace == 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<PSJobProxy> jobs = new Collection<PSJobProxy>();
            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

        /// <summary>
        /// Will begin streaming data for a job object created by the "create" method that is in a not started state.
        /// </summary>
        public void ReceiveJob()
        {
            ReceiveJob(null, null);
        }

        /// <summary>
        /// Will begin streaming data for a job object created by the "create" method that is in a not started state.
        /// </summary>
        /// <param name="dataAdded">delegate used to subscribe to data added events on the child jobs</param>
        /// <param name="stateChanged">delegate used to subscribe to state changed events on the child jobs</param>
        public void ReceiveJob(EventHandler<JobDataAddedEventArgs> dataAdded, EventHandler<JobStateEventArgs> 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<PSObject>)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<CommandParameterCollection> { 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<Hashtable> psParamCollection = new List<Hashtable>();
            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;
        }

        /// <summary>
        /// Helper to do error checking for getting a property of type T from a PSobject.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="o"></param>
        /// <param name="propertyName"></param>
        /// <param name="propertyValue"></param>
        /// <returns></returns>
        internal static bool TryGetJobPropertyValue<T>(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)