OutOfProcTransportManager.cs

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

/*
 * Common file that contains implementation for both server and client transport
 * managers for Out-Of-Process and Named Pipe (on the local machine) remoting implementation.
 * These interfaces are used by *-Job cmdlets to support background jobs and
 * attach-to-process feature without depending on WinRM (WinRM has complex requirements like 
 * elevation to support local machine remoting).
 */

using System.Collections.Generic;
using System.Diagnostics;
using System.Globalization;
using System.IO;
using System.Management.Automation.Internal;
using System.Management.Automation.Runspaces;
using System.Management.Automation.Tracing;
using System.Net;
using System.Threading;
using System.Xml;

using PSRemotingCryptoHelper = System.Management.Automation.Internal.PSRemotingCryptoHelper;
using PSRemotingCryptoHelperServer = System.Management.Automation.Internal.PSRemotingCryptoHelperServer;
using RunspaceConnectionInfo = System.Management.Automation.Runspaces.RunspaceConnectionInfo;
using ClientRemotePowerShell = System.Management.Automation.Runspaces.Internal.ClientRemotePowerShell;
using NewProcessConnectionInfo = System.Management.Automation.Runspaces.NewProcessConnectionInfo;
using PSTask = System.Management.Automation.Internal.PSTask;
using PSOpcode = System.Management.Automation.Internal.PSOpcode;
using PSEventId = System.Management.Automation.Internal.PSEventId;
using TypeTable = System.Management.Automation.Runspaces.TypeTable;
using Dbg = System.Management.Automation.Diagnostics;

namespace System.Management.Automation.Remoting
{
    internal static class OutOfProcessUtils
    {
        #region Helper Fields

        internal const string PS_OUT_OF_PROC_DATA_TAG = "Data";
        internal const string PS_OUT_OF_PROC_DATA_ACK_TAG = "DataAck";
        internal const string PS_OUT_OF_PROC_STREAM_ATTRIBUTE = "Stream";
        internal const string PS_OUT_OF_PROC_PSGUID_ATTRIBUTE = "PSGuid";
        internal const string PS_OUT_OF_PROC_CLOSE_TAG = "Close";
        internal const string PS_OUT_OF_PROC_CLOSE_ACK_TAG = "CloseAck";
        internal const string PS_OUT_OF_PROC_COMMAND_TAG = "Command";
        internal const string PS_OUT_OF_PROC_COMMAND_ACK_TAG = "CommandAck";
        internal const string PS_OUT_OF_PROC_SIGNAL_TAG = "Signal";
        internal const string PS_OUT_OF_PROC_SIGNAL_ACK_TAG = "SignalAck";
        internal const int EXITCODE_UNHANDLED_EXCEPTION = 0x0FA0;

        internal static XmlReaderSettings XmlReaderSettings;

        #endregion

        #region Static Constructor

        static OutOfProcessUtils()
        {
            // data coming from inputs stream is in Xml format. create appropriate
            // xml reader settings only once and reuse the same settings for all
            // the reads from StdIn stream.
            XmlReaderSettings = new XmlReaderSettings();
            XmlReaderSettings.CheckCharacters = false;
            XmlReaderSettings.IgnoreComments = true;
            XmlReaderSettings.IgnoreProcessingInstructions = true;
#if !CORECLR  // No XmlReaderSettings.XmlResolver in CoreCLR
            XmlReaderSettings.XmlResolver = null;
#endif
            XmlReaderSettings.ConformanceLevel = ConformanceLevel.Fragment;
        }

        #endregion

        #region Packet Creation Helper Methods

        internal static string CreateDataPacket(byte[] data, DataPriorityType streamType, Guid psGuid)
        {
            string result = string.Format(CultureInfo.InvariantCulture,
                "<{0} {1}='{2}' {3}='{4}'>{5}</{0}>",
                PS_OUT_OF_PROC_DATA_TAG,
                PS_OUT_OF_PROC_STREAM_ATTRIBUTE,
                streamType.ToString(),
                PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                psGuid.ToString(),
                Convert.ToBase64String(data));

            return result;
        }

        internal static string CreateDataAckPacket(Guid psGuid)
        {
            return CreatePSGuidPacket(PS_OUT_OF_PROC_DATA_ACK_TAG, psGuid);
        }

        internal static string CreateCommandPacket(Guid psGuid)
        {
            return CreatePSGuidPacket(PS_OUT_OF_PROC_COMMAND_TAG, psGuid);
        }

        internal static string CreateCommandAckPacket(Guid psGuid)
        {
            return CreatePSGuidPacket(PS_OUT_OF_PROC_COMMAND_ACK_TAG, psGuid);
        }

        internal static string CreateClosePacket(Guid psGuid)
        {
            return CreatePSGuidPacket(PS_OUT_OF_PROC_CLOSE_TAG, psGuid);
        }

        internal static string CreateCloseAckPacket(Guid psGuid)
        {
            return CreatePSGuidPacket(PS_OUT_OF_PROC_CLOSE_ACK_TAG, psGuid);
        }

        internal static string CreateSignalPacket(Guid psGuid)
        {
            return CreatePSGuidPacket(PS_OUT_OF_PROC_SIGNAL_TAG, psGuid);
        }

        internal static string CreateSignalAckPacket(Guid psGuid)
        {
            return CreatePSGuidPacket(PS_OUT_OF_PROC_SIGNAL_ACK_TAG, psGuid);
        }

        /// <summary>
        /// Common method to create a packet that contains only a PS Guid
        /// with element name changing
        /// </summary>
        /// <param name="element"></param>
        /// <param name="psGuid"></param>
        /// <returns></returns>
        private static string CreatePSGuidPacket(string element, Guid psGuid)
        {
            string result = string.Format(CultureInfo.InvariantCulture,
                "<{0} {1}='{2}' />",
                element,
                PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                psGuid.ToString());

            return result;
        }
        #endregion

        #region Packet Processing Helper Methods / Delegates

        internal delegate void DataPacketReceived(byte[] rawData, string stream, Guid psGuid);
        internal delegate void DataAckPacketReceived(Guid psGuid);
        internal delegate void CommandCreationPacketReceived(Guid psGuid);
        internal delegate void CommandCreationAckReceived(Guid psGuid);
        internal delegate void ClosePacketReceived(Guid psGuid);
        internal delegate void CloseAckPacketReceived(Guid psGuid);
        internal delegate void SignalPacketReceived(Guid psGuid);
        internal delegate void SignalAckPacketReceived(Guid psGuid);

        internal struct DataProcessingDelegates
        {
            internal DataPacketReceived DataPacketReceived;
            internal DataAckPacketReceived DataAckPacketReceived;
            internal CommandCreationPacketReceived CommandCreationPacketReceived;
            internal CommandCreationAckReceived CommandCreationAckReceived;
            internal SignalPacketReceived SignalPacketReceived;
            internal SignalAckPacketReceived SignalAckPacketReceived;
            internal ClosePacketReceived ClosePacketReceived;
            internal CloseAckPacketReceived CloseAckPacketReceived;
        }

        /// <summary>
        /// Process's data. Data should be a valid XML.
        /// </summary>
        /// <param name="data"></param>
        /// <param name="callbacks"></param>
        /// <exception cref="PSRemotingTransportException">
        /// 1. Expected only two attributes with names "{0}" and "{1}" in "{2}" element.
        /// 2. Not enough data available to process "{0}" element.
        /// 3. Unknown node "{0}" in "{1}" element. Only "{2}" is expected in "{1}" element.
        /// </exception>
        internal static void ProcessData(string data, DataProcessingDelegates callbacks)
        {
            if (string.IsNullOrEmpty(data))
            {
                return;
            }

            XmlReader reader = XmlReader.Create(new StringReader(data), XmlReaderSettings);
            while (reader.Read())
            {
                switch (reader.NodeType)
                {
                    case XmlNodeType.Element:
                        ProcessElement(reader, callbacks);
                        break;
                    case XmlNodeType.EndElement:
                        break;
                    default:
                        throw new PSRemotingTransportException(PSRemotingErrorId.IPCUnknownNodeType, RemotingErrorIdStrings.IPCUnknownNodeType,
                            reader.NodeType.ToString(),
                            XmlNodeType.Element.ToString(),
                            XmlNodeType.EndElement.ToString());
                }
            }
        }

        /// <summary>
        /// Process an XmlElement. The element name must be one of the following:
        ///         "Data"
        /// </summary>
        /// <param name="xmlReader"></param>
        /// <param name="callbacks"></param>
        /// <exception cref="PSRemotingTransportException">
        /// 1. Expected only two attributes with names "{0}" and "{1}" in "{2}" element.
        /// 2. Not enough data available to process "{0}" element.
        /// 3. Unknown node "{0}" in "{1}" element. Only "{2}" is expected in "{1}" element.
        /// </exception>
        private static void ProcessElement(XmlReader xmlReader, DataProcessingDelegates callbacks)
        {
            Dbg.Assert(null != xmlReader, "xmlReader cannot be null.");
            Dbg.Assert(xmlReader.NodeType == XmlNodeType.Element, "xmlReader's NodeType should be of type Element");

            PowerShellTraceSource tracer = PowerShellTraceSourceFactory.GetTraceSource();

            switch (xmlReader.LocalName)
            {
                case OutOfProcessUtils.PS_OUT_OF_PROC_DATA_TAG:
                    {
                        // A <Data> should have 1 attribute identifying the stream
                        if (xmlReader.AttributeCount != 2)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCWrongAttributeCountForDataElement,
                                RemotingErrorIdStrings.IPCWrongAttributeCountForDataElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_STREAM_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_DATA_TAG);
                        }
                        string stream = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_STREAM_ATTRIBUTE);
                        string psGuidString = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE);
                        Guid psGuid = new Guid(psGuidString);

                        // Now move the reader to the data portion
                        if (!xmlReader.Read())
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCInsufficientDataforElement,
                                RemotingErrorIdStrings.IPCInsufficientDataforElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_DATA_TAG);
                        }

                        if (xmlReader.NodeType != XmlNodeType.Text)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCOnlyTextExpectedInDataElement,
                                RemotingErrorIdStrings.IPCOnlyTextExpectedInDataElement,
                                xmlReader.NodeType, OutOfProcessUtils.PS_OUT_OF_PROC_DATA_TAG, XmlNodeType.Text);
                        }

                        string data = xmlReader.Value;
                        tracer.WriteMessage("OutOfProcessUtils.ProcessElement : PS_OUT_OF_PROC_DATA received, psGuid : " + psGuid.ToString());
                        byte[] rawData = Convert.FromBase64String(data);
                        callbacks.DataPacketReceived(rawData, stream, psGuid);
                    }

                    break;
                case OutOfProcessUtils.PS_OUT_OF_PROC_DATA_ACK_TAG:
                    {
                        if (xmlReader.AttributeCount != 1)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCWrongAttributeCountForElement,
                                RemotingErrorIdStrings.IPCWrongAttributeCountForElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_DATA_ACK_TAG);
                        }
                        string psGuidString = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE);
                        Guid psGuid = new Guid(psGuidString);

                        tracer.WriteMessage("OutOfProcessUtils.ProcessElement : PS_OUT_OF_PROC_DATA_ACK received, psGuid : " + psGuid.ToString());
                        callbacks.DataAckPacketReceived(psGuid);
                    }
                    break;
                case OutOfProcessUtils.PS_OUT_OF_PROC_COMMAND_TAG:
                    {
                        if (xmlReader.AttributeCount != 1)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCWrongAttributeCountForElement,
                                RemotingErrorIdStrings.IPCWrongAttributeCountForElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_COMMAND_TAG);
                        }
                        string psGuidString = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE);
                        Guid psGuid = new Guid(psGuidString);

                        tracer.WriteMessage("OutOfProcessUtils.ProcessElement : PS_OUT_OF_PROC_COMMAND received, psGuid : " + psGuid.ToString());
                        callbacks.CommandCreationPacketReceived(psGuid);
                    }
                    break;
                case OutOfProcessUtils.PS_OUT_OF_PROC_COMMAND_ACK_TAG:
                    {
                        if (xmlReader.AttributeCount != 1)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCWrongAttributeCountForElement,
                                RemotingErrorIdStrings.IPCWrongAttributeCountForElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_COMMAND_ACK_TAG);
                        }
                        string psGuidString = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE);
                        Guid psGuid = new Guid(psGuidString);
                        tracer.WriteMessage("OutOfProcessUtils.ProcessElement : PS_OUT_OF_PROC_COMMAND_ACK received, psGuid : " + psGuid.ToString());
                        callbacks.CommandCreationAckReceived(psGuid);
                    }
                    break;
                case OutOfProcessUtils.PS_OUT_OF_PROC_CLOSE_TAG:
                    {
                        if (xmlReader.AttributeCount != 1)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCWrongAttributeCountForElement,
                                RemotingErrorIdStrings.IPCWrongAttributeCountForElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_CLOSE_TAG);
                        }
                        string psGuidString = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE);
                        Guid psGuid = new Guid(psGuidString);

                        tracer.WriteMessage("OutOfProcessUtils.ProcessElement : PS_OUT_OF_PROC_CLOSE received, psGuid : " + psGuid.ToString());
                        callbacks.ClosePacketReceived(psGuid);
                    }
                    break;
                case OutOfProcessUtils.PS_OUT_OF_PROC_CLOSE_ACK_TAG:
                    {
                        if (xmlReader.AttributeCount != 1)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCWrongAttributeCountForElement,
                            RemotingErrorIdStrings.IPCWrongAttributeCountForElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_CLOSE_ACK_TAG);
                        }
                        string psGuidString = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE);
                        Guid psGuid = new Guid(psGuidString);
                        tracer.WriteMessage("OutOfProcessUtils.ProcessElement : PS_OUT_OF_PROC_CLOSE_ACK received, psGuid : " + psGuid.ToString());
                        callbacks.CloseAckPacketReceived(psGuid);
                    }
                    break;
                case OutOfProcessUtils.PS_OUT_OF_PROC_SIGNAL_TAG:
                    {
                        if (xmlReader.AttributeCount != 1)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCWrongAttributeCountForElement,
                            RemotingErrorIdStrings.IPCWrongAttributeCountForElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_SIGNAL_TAG);
                        }
                        string psGuidString = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE);
                        Guid psGuid = new Guid(psGuidString);

                        tracer.WriteMessage("OutOfProcessUtils.ProcessElement : PS_OUT_OF_PROC_SIGNAL received, psGuid : " + psGuid.ToString());
                        callbacks.SignalPacketReceived(psGuid);
                    }
                    break;
                case OutOfProcessUtils.PS_OUT_OF_PROC_SIGNAL_ACK_TAG:
                    {
                        if (xmlReader.AttributeCount != 1)
                        {
                            throw new PSRemotingTransportException(PSRemotingErrorId.IPCWrongAttributeCountForElement,
                            RemotingErrorIdStrings.IPCWrongAttributeCountForElement,
                                OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE,
                                OutOfProcessUtils.PS_OUT_OF_PROC_SIGNAL_ACK_TAG);
                        }
                        string psGuidString = xmlReader.GetAttribute(OutOfProcessUtils.PS_OUT_OF_PROC_PSGUID_ATTRIBUTE);
                        Guid psGuid = new Guid(psGuidString);
                        tracer.WriteMessage("OutOfProcessUtils.ProcessElement : PS_OUT_OF_PROC_SIGNAL_ACK received, psGuid : " + psGuid.ToString());
                        callbacks.SignalAckPacketReceived(psGuid);
                    }
                    break;
                default:
                    throw new PSRemotingTransportException(PSRemotingErrorId.IPCUnknownElementReceived,
                    RemotingErrorIdStrings.IPCUnknownElementReceived,
                        xmlReader.LocalName);
            }
        }

        #endregion
    }


    /// <summary>
    /// A wrapper around TextWriter to allow for synchronized writing to a stream.
    /// Synchronization is required to avoid collision when multiple TransportManager's
    /// write data at the same time to the same writer
    /// </summary>
    internal class OutOfProcessTextWriter
    {
        #region Private Data

        private TextWriter _writer;
        private bool _isStopped;
        private object _syncObject = new object();

        #endregion

        #region Constructors

        /// <summary>
        /// Constructs the wrapper
        /// </summary>
        /// <param name="writerToWrap"></param>
        internal OutOfProcessTextWriter(TextWriter writerToWrap)
        {
            Dbg.Assert(null != writerToWrap, "Cannot wrap a null writer.");
            _writer = writerToWrap;
        }

        #endregion

        #region Methods

        /// <summary>
        /// Calls writer.WriteLine() with data.
        /// </summary>
        /// <param name="data"></param>
        internal virtual void WriteLine(string data)
        {
            if (_isStopped)
            {
                return;
            }

            lock (_syncObject)
            {
                _writer.WriteLine(data);
                _writer.Flush();
            }
        }

        /// <summary>
        /// Stops the writer from writing anything to the stream.
        /// This is used by session transport manager when the server
        /// process is exited but the session data structure handlers
        /// are not notified yet. So while the data structure handler
        /// is disposing we should not let anyone use the stream.
        /// </summary>
        internal void StopWriting()
        {
            _isStopped = true;
        }

        #endregion
    }
}

namespace System.Management.Automation.Remoting.Client
{
    internal abstract class OutOfProcessClientSessionTransportManagerBase : BaseClientSessionTransportManager
    {
        #region Data

        private PrioritySendDataCollection.OnDataAvailableCallback _onDataAvailableToSendCallback;
        private OutOfProcessUtils.DataProcessingDelegates _dataProcessingCallbacks;
        private Dictionary<Guid, OutOfProcessClientCommandTransportManager> _cmdTransportManagers;
        private Timer _closeTimeOutTimer;

        protected OutOfProcessTextWriter stdInWriter;
        protected PowerShellTraceSource _tracer;

        #endregion

        #region Constructor

        internal OutOfProcessClientSessionTransportManagerBase(
            Guid runspaceId,
            PSRemotingCryptoHelper cryptoHelper)
            : base(runspaceId, cryptoHelper)
        {
            _onDataAvailableToSendCallback =
                new PrioritySendDataCollection.OnDataAvailableCallback(OnDataAvailableCallback);

            _cmdTransportManagers = new Dictionary<Guid, OutOfProcessClientCommandTransportManager>();

            _dataProcessingCallbacks = new OutOfProcessUtils.DataProcessingDelegates();
            _dataProcessingCallbacks.DataPacketReceived += new OutOfProcessUtils.DataPacketReceived(OnDataPacketReceived);
            _dataProcessingCallbacks.DataAckPacketReceived += new OutOfProcessUtils.DataAckPacketReceived(OnDataAckPacketReceived);
            _dataProcessingCallbacks.CommandCreationPacketReceived += new OutOfProcessUtils.CommandCreationPacketReceived(OnCommandCreationPacketReceived);
            _dataProcessingCallbacks.CommandCreationAckReceived += new OutOfProcessUtils.CommandCreationAckReceived(OnCommandCreationAckReceived);
            _dataProcessingCallbacks.SignalPacketReceived += new OutOfProcessUtils.SignalPacketReceived(OnSignalPacketReceived);
            _dataProcessingCallbacks.SignalAckPacketReceived += new OutOfProcessUtils.SignalAckPacketReceived(OnSiganlAckPacketReceived);
            _dataProcessingCallbacks.ClosePacketReceived += new OutOfProcessUtils.ClosePacketReceived(OnClosePacketReceived);
            _dataProcessingCallbacks.CloseAckPacketReceived += new OutOfProcessUtils.CloseAckPacketReceived(OnCloseAckReceived);

            dataToBeSent.Fragmentor = base.Fragmentor;
            // session transport manager can recieve unlimited data..however each object is limited
            // by maxRecvdObjectSize. this is to allow clients to use a session for an unlimited time..
            // also the messages that can be sent to a session are limited and very controlled.
            // However a command transport manager can be restricted to receive only a fixed amount of data
            // controlled by maxRecvdDataSizeCommand..This is because commands can accept any number of input
            // objects.
            ReceivedDataCollection.MaximumReceivedDataSize = null;
            ReceivedDataCollection.MaximumReceivedObjectSize = BaseTransportManager.MaximumReceivedObjectSize;
            // timers initialization
            _closeTimeOutTimer = new Timer(OnCloseTimeOutTimerElapsed, null, Timeout.Infinite, Timeout.Infinite);

            _tracer = PowerShellTraceSourceFactory.GetTraceSource();
        }

        #endregion

        #region Overrides

        internal override void ConnectAsync()
        {
            throw new NotImplementedException(RemotingErrorIdStrings.IPCTransportConnectError);
        }

        /// <summary>
        /// Closes the server process.
        /// </summary>
        internal override void CloseAsync()
        {
            bool shouldRaiseCloseCompleted = false;
            lock (syncObject)
            {
                if (isClosed == true)
                {
                    return;
                }

                // first change the state..so other threads
                // will know that we are closing.
                isClosed = true;

                if (null == stdInWriter)
                {
                    // this will happen if CloseAsync() is called
                    // before ConnectAsync()..in which case we 
                    // just need to raise close completed event.
                    shouldRaiseCloseCompleted = true;
                }
            }

            base.CloseAsync();

            if (shouldRaiseCloseCompleted)
            {
                RaiseCloseCompleted();
                return;
            }

            PSEtwLog.LogAnalyticInformational(PSEventId.WSManCloseShell,
                PSOpcode.Disconnect, PSTask.None, PSKeyword.Transport | PSKeyword.UseAlwaysAnalytic,
                RunspacePoolInstanceId.ToString());

            _tracer.WriteMessage("OutOfProcessClientSessionTransportManager.CloseAsync, when sending close session packet, progress command count should be zero, current cmd count: " + _cmdTransportManagers.Count + ", RunSpacePool Id : " + this.RunspacePoolInstanceId);

            try
            {
                // send Close signal to the server and let it die gracefully.
                stdInWriter.WriteLine(OutOfProcessUtils.CreateClosePacket(Guid.Empty));

                // start the timer..so client can fail deterministically
                _closeTimeOutTimer.Change(60 * 1000, Timeout.Infinite);
            }
            catch (IOException)
            {
                // Cannot communicate with server.  Allow client to complete close operation.
                shouldRaiseCloseCompleted = true;
            }

            if (shouldRaiseCloseCompleted)
            {
                RaiseCloseCompleted();
            }
        }

        /// <summary>
        /// Create a transport manager for command
        /// </summary>
        /// <param name="connectionInfo"></param>
        /// <param name="cmd"></param>
        /// <param name="noInput"></param>
        /// <returns></returns>
        internal override BaseClientCommandTransportManager CreateClientCommandTransportManager(
            RunspaceConnectionInfo connectionInfo,
            ClientRemotePowerShell cmd,
            bool noInput)
        {
            Dbg.Assert(null != cmd, "Cmd cannot be null");

            OutOfProcessClientCommandTransportManager result = new
                OutOfProcessClientCommandTransportManager(cmd, noInput, this, stdInWriter);
            AddCommandTransportManager(cmd.InstanceId, result);

            return result;
        }

        /// <summary>
        /// Kills the server process and disposes other resources
        /// </summary>
        /// <param name="isDisposing"></param>
        internal override void Dispose(bool isDisposing)
        {
            base.Dispose(isDisposing);
            if (isDisposing)
            {
                _cmdTransportManagers.Clear();
                _closeTimeOutTimer.Dispose();
            }
        }

        #endregion

        #region Helper Methods

        private void AddCommandTransportManager(Guid key, OutOfProcessClientCommandTransportManager cmdTM)
        {
            lock (syncObject)
            {
                if (isClosed)
                {
                    // It is possible for this add command to occur after/during session close via
                    // asynchronous stop pipeline or Stop-Job.  In this case ignore the command.
                    _tracer.WriteMessage("OutOfProcessClientSessionTransportManager.AddCommandTransportManager, Adding command transport on closed session, RunSpacePool Id : " + this.RunspacePoolInstanceId);
                    return;
                }

                Dbg.Assert(!_cmdTransportManagers.ContainsKey(key), "key already exists");
                _cmdTransportManagers.Add(key, cmdTM);
            }
        }

        internal override void RemoveCommandTransportManager(Guid key)
        {
            lock (syncObject)
            {
                // We always need to remove commands from collection, even if isClosed is true.
                // If we don't then we hang because CloseAsync() will not complete until all 
                // commands are closed.
                if (!_cmdTransportManagers.Remove(key))
                {
                    _tracer.WriteMessage("key does not exist to remove from cmdTransportManagers");
                }
            }
        }

        private OutOfProcessClientCommandTransportManager GetCommandTransportManager(Guid key)
        {
            lock (syncObject)
            {
                OutOfProcessClientCommandTransportManager result = null;
                _cmdTransportManagers.TryGetValue(key, out result);
                return result;
            }
        }

        private void OnCloseSessionCompleted()
        {
            //stop timer
            _closeTimeOutTimer.Change(Timeout.Infinite, Timeout.Infinite);
            RaiseCloseCompleted();
            CleanupConnection();
        }

        protected abstract void CleanupConnection();

        #endregion

        #region Event Handlers

        protected void HandleOutputDataReceived(string data)
        {
            try
            {
                OutOfProcessUtils.ProcessData(data, _dataProcessingCallbacks);
            }
            catch (Exception exception)
            {
                CommandProcessorBase.CheckForSevereException(exception);

                PSRemotingTransportException psrte =
                    new PSRemotingTransportException(PSRemotingErrorId.IPCErrorProcessingServerData,
                        RemotingErrorIdStrings.IPCErrorProcessingServerData,
                        exception.Message);
                RaiseErrorHandler(new TransportErrorOccuredEventArgs(psrte, TransportMethodEnum.ReceiveShellOutputEx));
            }
        }

        protected void HandleErrorDataReceived(string data)
        {
            lock (syncObject)
            {
                if (isClosed)
                {
                    return;
                }
            }

            PSRemotingTransportException psrte = new PSRemotingTransportException(PSRemotingErrorId.IPCServerProcessReportedError,
                RemotingErrorIdStrings.IPCServerProcessReportedError,
                data);
            RaiseErrorHandler(new TransportErrorOccuredEventArgs(psrte, TransportMethodEnum.Unknown));
        }

        protected void OnExited(object sender, EventArgs e)
        {
            TransportMethodEnum transportMethod = TransportMethodEnum.Unknown;
            lock (syncObject)
            {
                // There is no need to return when IsClosed==true here as in a legitimate case process exits
                // after Close is called..In that legitimate case, Exit handler is removed before
                // calling Exit..So, this Exit must have been called abnormally.
                if (isClosed)
                {
                    transportMethod = TransportMethodEnum.CloseShellOperationEx;
                }

                // dont let the writer write new data as the process is exited.
                // Not asssigning null to stdInWriter to fix the race condition between OnExited() and CloseAsync() methods.
                // 
                stdInWriter.StopWriting();
            }
            PSRemotingTransportException psrte = new PSRemotingTransportException(PSRemotingErrorId.IPCServerProcessExited,
                                RemotingErrorIdStrings.IPCServerProcessExited);
            RaiseErrorHandler(new TransportErrorOccuredEventArgs(psrte, transportMethod));
        }

        #endregion

        #region Sending Data related Methods

        protected void SendOneItem()
        {
            DataPriorityType priorityType;
            // This will either return data or register callback but doesn't do both.
            byte[] data = dataToBeSent.ReadOrRegisterCallback(_onDataAvailableToSendCallback,
                out priorityType);
            if (null != data)
            {
                SendData(data, priorityType);
            }
        }

        private void OnDataAvailableCallback(byte[] data, DataPriorityType priorityType)
        {
            Dbg.Assert(null != data, "data cannot be null in the data available callback");

            tracer.WriteLine("Received data to be sent from the callback.");
            SendData(data, priorityType);
        }

        private void SendData(byte[] data, DataPriorityType priorityType)
        {
            PSEtwLog.LogAnalyticInformational(
                       PSEventId.WSManSendShellInputEx, PSOpcode.Send, PSTask.None,
                       PSKeyword.Transport | PSKeyword.UseAlwaysAnalytic,
                       RunspacePoolInstanceId.ToString(),
                       Guid.Empty.ToString(),
                       data.Length.ToString(CultureInfo.InvariantCulture));

            lock (syncObject)
            {
                if (isClosed)
                {
                    return;
                }

                stdInWriter.WriteLine(OutOfProcessUtils.CreateDataPacket(data,
                    priorityType,
                    Guid.Empty));
            }
        }

        private void OnRemoteSessionSendCompleted()
        {
            PSEtwLog.LogAnalyticInformational(PSEventId.WSManSendShellInputExCallbackReceived,
                PSOpcode.Connect, PSTask.None, PSKeyword.Transport | PSKeyword.UseAlwaysAnalytic,
                RunspacePoolInstanceId.ToString(), Guid.Empty.ToString());

            SendOneItem();
        }

        #endregion

        #region Data Processing handlers

        private void OnDataPacketReceived(byte[] rawData, string stream, Guid psGuid)
        {
            string streamTemp = System.Management.Automation.Remoting.Client.WSManNativeApi.WSMAN_STREAM_ID_STDOUT;
            if (stream.Equals(DataPriorityType.PromptResponse.ToString(), StringComparison.OrdinalIgnoreCase))
            {
                streamTemp = System.Management.Automation.Remoting.Client.WSManNativeApi.WSMAN_STREAM_ID_PROMPTRESPONSE;
            }

            if (psGuid == Guid.Empty)
            {
                PSEtwLog.LogAnalyticInformational(
                    PSEventId.WSManReceiveShellOutputExCallbackReceived, PSOpcode.Receive, PSTask.None,
                    PSKeyword.Transport | PSKeyword.UseAlwaysAnalytic,
                    RunspacePoolInstanceId.ToString(),
                    Guid.Empty.ToString(),
                    rawData.Length.ToString(CultureInfo.InvariantCulture));

                // this data is meant for session.
                base.ProcessRawData(rawData, streamTemp);
            }
            else
            {
                // this is for a command
                OutOfProcessClientCommandTransportManager cmdTM = GetCommandTransportManager(psGuid);
                if (null != cmdTM)
                {
                    // not throwing the exception in null case as the command might have already
                    // closed. The RS data structure handler does not wait for the close ack before
                    // it clears the command transport manager..so this might happen.
                    cmdTM.OnRemoteCmdDataReceived(rawData, streamTemp);
                }
            }
        }

        private void OnDataAckPacketReceived(Guid psGuid)
        {
            if (psGuid == Guid.Empty)
            {
                // this data is meant for session.
                OnRemoteSessionSendCompleted();
            }
            else
            {
                // this is for a command
                OutOfProcessClientCommandTransportManager cmdTM = GetCommandTransportManager(psGuid);
                if (null != cmdTM)
                {
                    // not throwing the exception in null case as the command might have already
                    // closed. The RS data structure handler does not wait for the close ack before
                    // it clears the command transport manager..so this might happen.
                    cmdTM.OnRemoteCmdSendCompleted();
                }
            }
        }

        private void OnCommandCreationPacketReceived(Guid psGuid)
        {
            throw new PSRemotingTransportException(PSRemotingErrorId.IPCUnknownElementReceived,
                RemotingErrorIdStrings.IPCUnknownElementReceived,
                   OutOfProcessUtils.PS_OUT_OF_PROC_COMMAND_TAG);
        }

        private void OnCommandCreationAckReceived(Guid psGuid)
        {
            OutOfProcessClientCommandTransportManager cmdTM = GetCommandTransportManager(psGuid);
            if (null == cmdTM)
            {
                throw new PSRemotingTransportException(PSRemotingErrorId.IPCUnknownCommandGuid,
                    RemotingErrorIdStrings.IPCUnknownCommandGuid,
                    psGuid.ToString(), OutOfProcessUtils.PS_OUT_OF_PROC_COMMAND_ACK_TAG);
            }

            cmdTM.OnCreateCmdCompleted();

            _tracer.WriteMessage("OutOfProcessClientSessionTransportManager.OnCommandCreationAckReceived, in progress command count after cmd creation ACK : " + _cmdTransportManagers.Count + ", psGuid : " + psGuid.ToString());
        }

        private void OnSignalPacketReceived(Guid psGuid)
        {
            throw new PSRemotingTransportException(PSRemotingErrorId.IPCUnknownElementReceived,
                RemotingErrorIdStrings.IPCUnknownElementReceived,
                   OutOfProcessUtils.PS_OUT_OF_PROC_SIGNAL_TAG);
        }

        private void OnSiganlAckPacketReceived(Guid psGuid)
        {
            if (psGuid == Guid.Empty)
            {
                throw new PSRemotingTransportException(PSRemotingErrorId.IPCNoSignalForSession,
                    RemotingErrorIdStrings.IPCNoSignalForSession,
                    OutOfProcessUtils.PS_OUT_OF_PROC_SIGNAL_ACK_TAG);
            }
            else
            {
                OutOfProcessClientCommandTransportManager cmdTM = GetCommandTransportManager(psGuid);
                if (null != cmdTM)
                {
                    cmdTM.OnRemoteCmdSignalCompleted();
                }
            }
        }

        private void OnClosePacketReceived(Guid psGuid)
        {
            throw new PSRemotingTransportException(PSRemotingErrorId.IPCUnknownElementReceived,
                RemotingErrorIdStrings.IPCUnknownElementReceived,
                   OutOfProcessUtils.PS_OUT_OF_PROC_CLOSE_TAG);
        }

        private void OnCloseAckReceived(Guid psGuid)
        {
            int commandCount;
            lock (syncObject)
            {
                commandCount = _cmdTransportManagers.Count;
            }

            if (psGuid == Guid.Empty)
            {
                _tracer.WriteMessage("OutOfProcessClientSessionTransportManager.OnCloseAckReceived, progress command count after CLOSE ACK should be zero = " + commandCount + " psGuid : " + psGuid.ToString());

                this.OnCloseSessionCompleted();
            }
            else
            {
                _tracer.WriteMessage("OutOfProcessClientSessionTransportManager.OnCloseAckReceived, in progress command count should be greater than zero: " + commandCount + ", RunSpacePool Id : " + this.RunspacePoolInstanceId + ", psGuid : " + psGuid.ToString());

                OutOfProcessClientCommandTransportManager cmdTM = GetCommandTransportManager(psGuid);
                if (null != cmdTM)
                {
                    // this might legitimately happen if cmd is already closed before we get an
                    // ACK back from server.
                    cmdTM.OnCloseCmdCompleted();
                }
            }
        }

        #endregion

        #region Private Timeout handlers

        internal void OnCloseTimeOutTimerElapsed(object source)
        {
            PSRemotingTransportException psrte = new PSRemotingTransportException(PSRemotingErrorId.IPCCloseTimedOut, RemotingErrorIdStrings.IPCCloseTimedOut);
            RaiseErrorHandler(new TransportErrorOccuredEventArgs(psrte, TransportMethodEnum.CloseShellOperationEx));
        }

        #endregion
    }

    internal class OutOfProcessClientSessionTransportManager : OutOfProcessClientSessionTransportManagerBase
    {
        #region Private Data

        private Process _serverProcess;
        private NewProcessConnectionInfo _connectionInfo;
        private bool _processCreated = true;
        private PowerShellProcessInstance _processInstance;

        #endregion

        #region Constructor

        internal OutOfProcessClientSessionTransportManager(Guid runspaceId,
            NewProcessConnectionInfo connectionInfo,
            PSRemotingCryptoHelper cryptoHelper)
            : base(runspaceId, cryptoHelper)
        {
            _connectionInfo = connectionInfo;
        }

        #endregion

        #region Overrides

        /// <summary>
        /// Launch a new Process (PowerShell.exe -s) to perform remoting. This is used by *-Job cmdlets
        /// to support background jobs without depending on WinRM (WinRM has complex requirements like
        /// elevation to support local machine remoting)
        /// </summary>
        /// <exception cref="System.InvalidOperationException">
        /// </exception>
        /// <exception cref="System.ComponentModel.Win32Exception">
        /// 1. There was an error in opening the associated file. 
        /// </exception>
        internal override void CreateAsync()
        {
            if (null != _connectionInfo)
            {
                _processInstance = _connectionInfo.Process ?? new PowerShellProcessInstance(_connectionInfo.PSVersion,
                                                                                           _connectionInfo.Credential,
                                                                                           _connectionInfo.InitializationScript,
                                                                                           _connectionInfo.RunAs32);
                if (_connectionInfo.Process != null)
                {
                    _processCreated = false;
                }
                // _processInstance.Start();
            }

            PSEtwLog.LogAnalyticInformational(PSEventId.WSManCreateShell, PSOpcode.Connect,
                            PSTask.CreateRunspace, PSKeyword.Transport | PSKeyword.UseAlwaysAnalytic,
                            RunspacePoolInstanceId.ToString());

            try
            {
                lock (syncObject)
                {
                    if (isClosed)
                    {
                        return;
                    }

                    // Attach handlers and start the process
                    _serverProcess = _processInstance.Process;

                    if (_processInstance.RunspacePool != null)
                    {
                        _processInstance.RunspacePool.Close();
                        _processInstance.RunspacePool.Dispose();
                    }