{"action":"created","comment":{"url":"https://api.github.com/repos/PowerShell/PowerShell/pulls/comments/88130158","pull_request_review_id":8713096,"id":88130158,"diff_hunk":"@@ -1344,676 +1431,434 @@ private bool IsMiniShell()\n         internal static bool IsServerSide { get; set; }\n     }\n \n-    /// <summary>\n-    /// Helper class to handle writing input to a process.\n-    /// </summary>\n-    internal class ProcessInputWriter\n+    internal class ProcessOutputHandler\n     {\n-        #region constructor \n-\n-        private InternalCommand _command;\n-        /// <summary>\n-        /// Creates an instance of ProcessInputWriter \n-        /// </summary>\n-        internal ProcessInputWriter(InternalCommand command)\n-        {\n-            Dbg.Assert(command != null, \"Caller should validate the parameter\");\n-            _command = command;\n-        }\n+        internal const string XmlCliTag = \"#< CLIXML\";\n \n-        #endregion constructor\n+        private int _refCount;\n+        private BlockingCollection<ProcessOutputObject> _queue;\n+        private bool _isFirstOutput;\n+        private bool _isFirstError;\n+        private bool _isXmlCliOutput;\n+        private bool _isXmlCliError;\n+        private string _processFileName;\n \n-        /// <summary>\n-        /// Input is collected in this list\n-        /// </summary>\n-        private ArrayList _inputList = new ArrayList();\n-        /// <summary>\n-        /// Add an object to write to process\n-        /// </summary>\n-        /// <param name=\"input\"></param>\n-        internal void Add(object input)\n+        public ProcessOutputHandler(Process process, BlockingCollection<ProcessOutputObject> queue)\n         {\n-            _inputList.Add(input);\n-        }\n+            Debug.Assert(process.StartInfo.RedirectStandardOutput || process.StartInfo.RedirectStandardError, \"Caller should redirect at least one stream\");\n+            _refCount = 0;\n+            _processFileName = process.StartInfo.FileName;\n+            _queue = queue;\n \n-        /// <summary>\n-        /// Count of object in inputlist\n-        /// </summary>\n-        internal int Count\n-        {\n-            get\n+            // we incrementing refCount on the same thread and before running any processing\n+            // so it's safe to do it without Interlocked.\n+            if (process.StartInfo.RedirectStandardOutput) { _refCount++; }\n+            if (process.StartInfo.RedirectStandardError) { _refCount++; }\n+\n+            // once we have _refCount, we can start processing\n+            if (process.StartInfo.RedirectStandardOutput)\n             {\n-                return _inputList.Count;\n+                _isFirstOutput = true;\n+                _isXmlCliOutput = false;\n+                process.OutputDataReceived += new DataReceivedEventHandler(OutputHandler);\n+                process.BeginOutputReadLine();\n             }\n-        }\n-\n-        /// <summary>\n-        /// Stream to which input is written\n-        /// </summary>\n-        private StreamWriter _streamWriter;\n-\n-        /// <summary>\n-        /// Format of input.\n-        /// </summary>\n-        private NativeCommandIOFormat _inputFormat;\n-\n-        /// <summary>\n-        /// Thread which writes the input\n-        /// </summary>\n-        private Thread _inputThread;\n-\n-        /// <summary>\n-        /// Start writing input to process\n-        /// </summary>\n-        /// <param name=\"process\">\n-        /// process to which input is written\n-        /// </param>\n-        /// <param name=\"inputFormat\">\n-        /// </param>\n-        internal void Start(Process process, NativeCommandIOFormat inputFormat)\n-        {\n-            Dbg.Assert(process != null, \"caller should validate the parameter\");\n-\n-            //Get the encoding for writing to native command. Note we get the Encoding \n-            //from the current scope so a script or function can use a different encoding\n-            //than global value.\n-            Encoding pipeEncoding = _command.Context.GetVariableValue(SpecialVariables.OutputEncodingVarPath) as System.Text.Encoding ??\n-                                    Encoding.ASCII;\n-\n-            _streamWriter = new StreamWriter(process.StandardInput.BaseStream,\n-                                            pipeEncoding);\n-            _inputFormat = inputFormat;\n \n-            if (inputFormat == NativeCommandIOFormat.Text)\n+            if (process.StartInfo.RedirectStandardError)\n             {\n-                ConvertToString();\n+                _isFirstError = true;\n+                _isXmlCliError = false;\n+                process.ErrorDataReceived += new DataReceivedEventHandler(ErrorHandler);\n+                process.BeginErrorReadLine();\n             }\n-            _inputThread = new Thread(new ThreadStart(this.WriterThreadProc));\n-            _inputThread.Start();\n-        }\n-\n-        private bool _stopping = false;\n-        /// <summary>\n-        /// Stop writing input to process\n-        /// </summary>\n-        internal void Stop()\n-        {\n-            _stopping = true;\n         }\n \n-        /// <summary>\n-        /// This method wait for writer thread to finish.\n-        /// </summary>\n-        internal void Done()\n+        private void decrementRefCount()\n         {\n-            if (_inputThread != null)\n+            Debug.Assert(_refCount > 0, \"RefCount should always be positive, when we are trying to decrement it\");\n+            if (Interlocked.Decrement(ref _refCount) == 0)\n             {\n-                _inputThread.Join();\n+                _queue.CompleteAdding();\n             }\n         }\n \n-        /// <summary>\n-        /// Thread procedure for writing data to the child process...\n-        /// </summary>\n-        private void WriterThreadProc()\n+        private void OutputHandler(object sender, DataReceivedEventArgs outputReceived)\n         {\n-            try\n+            if (outputReceived.Data != null)\n             {\n-                if (_inputFormat == NativeCommandIOFormat.Text)\n+                if (_isFirstOutput)\n                 {\n-                    WriteTextInput();\n+                    _isFirstOutput = false;\n+                    if (string.Equals(outputReceived.Data, XmlCliTag, StringComparison.Ordinal))\n+                    {\n+                        _isXmlCliOutput = true;\n+                        return;\n+                    }\n                 }\n-                else\n+\n+                if (_isXmlCliOutput)\n                 {\n-                    WriteXmlInput();\n-                }\n+                    foreach (var record in DeserializeCliXmlObject(outputReceived.Data, true))\n+                    {\n+                        _queue.Add(record);\n+                    }\n+                }\n+                else\n+                {\n+                    _queue.Add(new ProcessOutputObject(outputReceived.Data, MinishellStream.Output));\n+                }\n             }\n-            catch (System.IO.IOException)\n+            else\n             {\n+                decrementRefCount();\n             }\n         }\n \n-        private void WriteTextInput()\n+        private void ErrorHandler(object sender, DataReceivedEventArgs errorReceived)\n         {\n-            try\n+            if (errorReceived.Data != null)\n             {\n-                foreach (object o in _inputList)\n+                if (string.Equals(errorReceived.Data, XmlCliTag, StringComparison.Ordinal))\n                 {\n-                    if (_stopping) return;\n+                    _isXmlCliError = true;\n+                    return;\n+                }\n \n-                    string line = PSObject.ToStringParser(_command.Context, o);\n-                    _streamWriter.Write(line);\n+                if (_isXmlCliError)\n+                {\n+                    foreach (var record in DeserializeCliXmlObject(errorReceived.Data, false))\n+                    {\n+                        _queue.Add(record);\n+                    }\n+                }\n+                else\n+                {\n+                    ErrorRecord errorRecord;\n+                    if (_isFirstError)\n+                    {\n+                        _isFirstError = false;\n+                        // Produce a regular error record for the first line of the output\n+                        errorRecord = new ErrorRecord(new RemoteException(errorReceived.Data), \"NativeCommandError\", ErrorCategory.NotSpecified, errorReceived.Data);\n+                    }\n+                    else\n+                    {\n+                        // Wrap the rest of the output in ErrorRecords with the \"NativeCommandErrorMessage\" error ID\n+                        errorRecord = new ErrorRecord(new RemoteException(errorReceived.Data), \"NativeCommandErrorMessage\", ErrorCategory.NotSpecified, null);\n+                    }\n+\n+                    _queue.Add(new ProcessOutputObject(errorRecord, MinishellStream.Error));\n                 }\n             }\n-            finally\n+            else\n             {\n-                _streamWriter.Dispose();\n+                decrementRefCount();\n             }\n         }\n \n-        private void WriteXmlInput()\n+        private List<ProcessOutputObject> DeserializeCliXmlObject(string xml, bool isOutput)\n         {\n+            var result = new List<ProcessOutputObject>();\n             try\n             {\n-                //Write header\n-                _streamWriter.WriteLine(\"#< CLIXML\");\n-\n-                // When (if) switching to XmlTextWriter.Create remember the OmitXmlDeclaration difference\n-                XmlWriter writer = XmlWriter.Create(_streamWriter);\n-                Serializer ser = new Serializer(writer);\n-                foreach (object o in _inputList)\n+                using (var streamReader = new MemoryStream(Encoding.UTF8.GetBytes(xml)))\n                 {\n-                    if (_stopping) return;\n-                    ser.Serialize(o);\n+                    XmlReader xmlReader = XmlReader.Create(streamReader, InternalDeserializer.XmlReaderSettingsForCliXml);\n+                    Deserializer des = new Deserializer(xmlReader);\n+                    while (!des.Done())\n+                    {\n+                        string streamName;\n+                        object obj = des.Deserialize(out streamName);\n+\n+                        //Decide the stream to which data belongs\n+                        MinishellStream stream = MinishellStream.Unknown;\n+                        if (streamName != null)\n+                        {\n+                            stream = StringToMinishellStreamConverter.ToMinishellStream(streamName);\n+                        }\n+                        if (stream == MinishellStream.Unknown)\n+                        {\n+                            stream = isOutput ? MinishellStream.Output : MinishellStream.Error;\n+                        }\n+\n+                        //Null is allowed only in output stream\n+                        if (stream != MinishellStream.Output && obj == null)\n+                        {\n+                            continue;\n+                        }\n+\n+                        if (stream == MinishellStream.Error)\n+                        {\n+                            if (obj is PSObject)\n+                            {\n+                                obj = ErrorRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj));\n+                            }\n+                            else\n+                            {\n+                                string errorMessage = null;\n+                                try\n+                                {\n+                                    errorMessage = (string)LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture);\n+                                }\n+                                catch (PSInvalidCastException)\n+                                {\n+                                    continue;\n+                                }\n+                                obj = new ErrorRecord(new RemoteException(errorMessage),\n+                                                    \"NativeCommandError\", ErrorCategory.NotSpecified, errorMessage);\n+                            }\n+                        }\n+                        else if (stream == MinishellStream.Information)\n+                        {\n+                            if (obj is PSObject)\n+                            {\n+                                obj = InformationRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj));\n+                            }\n+                            else\n+                            {\n+                                string messageData = null;\n+                                try\n+                                {\n+                                    messageData = (string)LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture);\n+                                }\n+                                catch (PSInvalidCastException)\n+                                {\n+                                    continue;\n+                                }\n+\n+                                obj = new InformationRecord(messageData, null);\n+                            }\n+                        }\n+                        else if (stream == MinishellStream.Debug ||\n+                                 stream == MinishellStream.Verbose ||\n+                                 stream == MinishellStream.Warning)\n+                        {\n+                            //Convert to string\n+                            try\n+                            {\n+                                obj = LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture);\n+                            }\n+                            catch (PSInvalidCastException)\n+                            {\n+                                continue;\n+                            }\n+                        }\n+                        result.Add(new ProcessOutputObject(obj, stream));\n+                    }\n                 }\n-                ser.Done();\n             }\n-            finally\n+            catch (XmlException originalException)\n             {\n-                _streamWriter.Dispose();\n-            }\n-        }\n+                string template = NativeCP.CliXmlError;\n+                string message = string.Format(\n+                    null,\n+                    template,\n+                    isOutput ? MinishellStream.Output : MinishellStream.Error,\n+                    _processFileName,\n+                    originalException.Message);\n+                XmlException newException = new XmlException(\n+                    message,\n+                    originalException);\n \n-        /// <summary>\n-        /// Formats the input objects using out-string. Output of out-string\n-        /// is given as input to native command processor.\n-        /// This method is to be called from the pipeline thread and not from the\n-        /// thread which writes input in to process.\n-        /// </summary>\n-        private void ConvertToString()\n-        {\n-            Dbg.Assert(_inputFormat == NativeCommandIOFormat.Text, \"InputFormat should be Text\");\n+                ErrorRecord error = new ErrorRecord(\n+                    newException,\n+                    \"ProcessStreamReader_CliXmlError\",\n+                    ErrorCategory.SyntaxError,\n+                    _processFileName);\n+                result.Add(new ProcessOutputObject(error, MinishellStream.Error));\n+            }\n \n-            PipelineProcessor p = new PipelineProcessor();\n-            p.Add(_command.Context.CreateCommand(\"out-string\", false));\n-            object[] result = (object[])p.SynchronousExecuteEnumerate(_inputList.ToArray());\n-            _inputList = new ArrayList(result);\n+            return result;\n         }\n     }\n \n     /// <summary>\n-    /// This helper class reads the output from error and output streams of\n-    /// process.\n+    /// Helper class to handle writing input to a process.\n     /// </summary>\n-    internal class ProcessOutputReader\n+    internal class ProcessInputWriter\n     {\n-        #region constructor\n-\n-        /// <summary>\n-        /// Process whose output is to be read.\n-        /// </summary>\n-        private Process _process;\n-\n-        /// <summary>\n-        /// Path of the process application\n-        /// </summary>\n-        private string _processPath;\n-\n-        private bool _redirectOutput;\n-\n-        private bool _redirectError;\n+        #region constructor \n \n+        private InternalCommand _command;\n         /// <summary>\n-        /// Process whose output is read\n+        /// Creates an instance of ProcessInputWriter \n         /// </summary>\n-        internal ProcessOutputReader(Process process, string processPath, bool redirectOutput, bool redirectError)\n+        internal ProcessInputWriter(InternalCommand command)\n         {\n-            Dbg.Assert(process != null, \"caller should validate the parameter\");\n-            Dbg.Assert(processPath != null, \"caller should validate the parameter\");\n-            Dbg.Assert(redirectOutput || redirectError, \"Either redirectOutput or redirectError must be true\");\n-            _process = process;\n-            _processPath = processPath;\n-            _redirectOutput = redirectOutput;\n-            _redirectError = redirectError;\n+            Dbg.Assert(command != null, \"Caller should validate the parameter\");\n+            _command = command;\n         }\n \n         #endregion constructor\n \n-        /// <summary>\n-        /// Reader for output stream of process\n-        /// </summary>\n-        private ProcessStreamReader _outputReader;\n-        /// <summary>\n-        /// Reader for error stream of process\n-        /// </summary>\n-        private ProcessStreamReader _errorReader;\n-        /// <summary>\n-        /// Synchronized object queue in which object read form output and \n-        /// error streams are deposited\n-        /// </summary>\n-        private ObjectStream _processOutput;\n+        private SteppablePipeline _pipeline;\n+        private Serializer _xmlSerializer;\n \n         /// <summary>\n-        /// Start reading the output/error. Note all the work is done asynchronously.\n+        /// Add an object to write to process\n         /// </summary>\n-        internal void Start()\n+        /// <param name=\"input\"></param>\n+        internal void Add(object input)\n         {\n-            _processOutput = new ObjectStream(128);\n+            if (_stopping || _streamWriter == null)\n+            {\n+                // if _streamWriter is already null, then we already called Done()\n+                // so we should just discard the input.\n+                return;\n+            }\n \n-            // Start async reading of error and output\n-            // readercount variable is used by multiple threads to close \"processOutput\" ObjectStream.\n-            // Without properly initializing the readercount, the ObjectStream might get\n-            // closed early. readerCount is protected here by using the lock.\n-            lock (_readerLock)\n+            if (_inputFormat == NativeCommandIOFormat.Text)\n             {\n-                if (_redirectOutput)\n-                {\n-                    _readerCount++;\n-                    _outputReader = new ProcessStreamReader(_process.StandardOutput, _processPath, true, _processOutput.ObjectWriter, this);\n-                    _outputReader.Start();\n-                }\n-                if (_redirectError)\n-                {\n-                    _readerCount++;\n-                    _errorReader = new ProcessStreamReader(_process.StandardError, _processPath, false, _processOutput.ObjectWriter, this);\n-                    _errorReader.Start();\n-                }\n+                AddTextInput(input);\n+            }\n+            else // Xml\n+            {\n+                AddXmlInput(input);\n             }\n         }\n \n-        /// <summary>\n-        /// Stops reading from streams. This is called from NativeCommandProcessor's StopProcessing\n-        /// method. Note return of this method doesn't mean reading has stopped and all threads are\n-        /// done. \n-        /// Use Done to ensure that all reading threads are finished.\n-        /// </summary>\n-        internal void Stop()\n+        private void AddTextInput(object input)\n         {\n-            if (_processOutput != null)\n-            {\n-                try\n-                {\n-                    //Close the reader for the stream.\n-                    _processOutput.ObjectReader.Close();\n-                }\n-                catch (Exception e) // ignore non-severe exceptions\n-                {\n-                    CommandProcessorBase.CheckForSevereException(e);\n-                }\n+            AddTextInputFromFormattedArray(_pipeline.Process(input));\n+        }\n \n+        private void AddTextInputFromFormattedArray(Array formattedObjects)\n+        {\n+            foreach (var item in formattedObjects)\n+            {\n+                string line = PSObject.ToStringParser(_command.Context, item);\n+                // if process is already finished and we are trying to write something to it,\n+                // we will get IOException\n                 try\n                 {\n-                    _processOutput.Close();\n+                    _streamWriter.WriteLine(line);\n                 }\n-                catch (Exception e) // ignore non-severe exceptions\n+                catch (IOException)\n                 {\n-                    CommandProcessorBase.CheckForSevereException(e);\n+                    // we are assuming that process is already finished\n+                    // we should just stop processing at this point\n+                    this.Done();\n+                    // stop foreach execution\n+                    break;\n                 }\n             }\n         }\n \n-        /// <summary>\n-        /// This method returns when all output reader threads have returned\n-        /// </summary>\n-        internal void Done()\n-        {\n-            if (_outputReader != null)\n-            {\n-                _outputReader.Done();\n-            }\n-            if (_errorReader != null)\n-            {\n-                _errorReader.Done();\n-            }\n-        }\n-\n-        /// <summary>\n-        /// Return one object which was read from the process.\n-        /// </summary>\n-        /// <returns>\n-        /// AutomationNull.Value if no more objects.\n-        /// object of type ProcessOutputObject otherwise\n-        /// </returns>\n-        internal object Read()\n+        private void AddXmlInput(object input)\n         {\n-            return _processOutput.ObjectReader.Read();\n-        }\n-\n-        /// <summary>\n-        /// object used for synchronizing ReaderDone call between two readers\n-        /// </summary>\n-        private object _readerLock = new object();\n-\n-        /// <summary>\n-        /// Count of readers - this is set by Start. If both output and error\n-        /// are redirected, it will be 2. If only one is redirected, it'll be 1.\n-        /// </summary>\n-        private int _readerCount;\n-\n-        /// <summary>\n-        /// This method is called by output or error reader when they are\n-        /// done reading. When it is called two times, we close the writer.\n-        /// </summary>\n-        /// <param name=\"isOutput\"></param>\n-        internal void ReaderDone(bool isOutput)\n-        {\n-            int temp;\n-            lock (_readerLock)\n+            try\n             {\n-                temp = --_readerCount;\n+                _xmlSerializer.Serialize(input);\n             }\n-            if (temp == 0)\n+            catch (IOException)\n             {\n-                _processOutput.ObjectWriter.Close();\n+                // we are assuming that process is already finished\n+                // we should just stop processing at this point\n+                this.Done();\n             }\n         }\n-    }\n-\n-    /// <summary>\n-    /// This class reads the string from output or error streams of process\n-    /// and processes them appropriately.\n-    /// </summary>\n-    /// <remarks>\n-    /// This class is not thread safe. It is assumed that NativeCommandProcessor\n-    /// class will synchronize access to this class between different threads.\n-    /// </remarks>\n-    internal class ProcessStreamReader\n-    {\n-        #region constructor \n \n         /// <summary>\n-        /// Stream from which data is read.\n-        /// </summary>\n-        private StreamReader _streamReader;\n-\n-        /// <summary>\n-        /// Flag which tells if streamReader is for stdout or stderr stream of process\n-        /// </summary>\n-        private bool _isOutput;\n-\n-        /// <summary>\n-        /// Writer to which data read from stream are written\n-        /// </summary>\n-        private PipelineWriter _writer;\n-\n-        /// <summary>\n-        /// Path to the process. This is used for setting the name of the thread.\n+        /// Stream to which input is written\n         /// </summary>\n-        private string _processPath;\n+        private StreamWriter _streamWriter;\n \n         /// <summary>\n-        /// ProcessReader which owns this stream reader\n+        /// Format of input.\n         /// </summary>\n-        private ProcessOutputReader _processOutputReader;\n+        private NativeCommandIOFormat _inputFormat;\n \n         /// <summary>\n-        /// Creates an instance of ProcessStreamReader\n+        /// Start writing input to process\n         /// </summary>\n-        /// <param name=\"streamReader\">\n-        /// Stream from which data is read\n-        /// </param>\n-        /// <param name=\"processPath\">\n-        /// Path to the process. This is used for setting the name of the thread.\n-        /// </param>\n-        /// <param name=\"isOutput\">\n-        /// if true stream is output stream of process \n-        /// else stream is error stream. \n-        /// </param>\n-        /// <param name=\"writer\">\n-        /// Processed data is written to it\n+        /// <param name=\"process\">\n+        /// process to which input is written\n         /// </param>\n-        /// <param name=\"processOutputReader\">\n-        /// ProcessOutputReader which owns this stream reader\n+        /// <param name=\"inputFormat\">\n         /// </param>\n-        internal ProcessStreamReader(StreamReader streamReader, string processPath, bool isOutput,\n-            PipelineWriter writer, ProcessOutputReader processOutputReader)\n+        internal void Start(Process process, NativeCommandIOFormat inputFormat)\n         {\n-            Dbg.Assert(streamReader != null, \"Caller should validate the parameter\");\n-            Dbg.Assert(processPath != null, \"Caller should validate the parameter\");\n-            Dbg.Assert(writer != null, \"Caller should validate the parameter\");\n-            Dbg.Assert(processOutputReader != null, \"Caller should validate the parameter\");\n-\n-            _streamReader = streamReader;\n-            _processPath = processPath;\n-            _isOutput = isOutput;\n-            _writer = writer;\n-            _processOutputReader = processOutputReader;\n-        }\n+            Dbg.Assert(process != null, \"caller should validate the paramter\");\n \n-        #endregion constructor\n+            //Get the encoding for writing to native command. Note we get the Encoding \n+            //from the current scope so a script or function can use a different encoding\n+            //than global value.\n+            Encoding pipeEncoding = _command.Context.GetVariableValue(SpecialVariables.OutputEncodingVarPath) as System.Text.Encoding ??\n+                                    Encoding.ASCII;\n \n-        /// <summary>\n-        /// Thread on which reading happens\n-        /// </summary>\n-        private Thread _thread = null;\n+            _streamWriter = new StreamWriter(process.StandardInput.BaseStream, pipeEncoding);\n+            _streamWriter.AutoFlush = true;\n \n-        /// <summary>\n-        /// Launches a new thread to start reading.\n-        /// </summary>\n-        internal void Start()\n-        {\n-            _thread = new Thread(new ThreadStart(ReaderStartProc));\n-            if (_isOutput)\n+            _inputFormat = inputFormat;\n+\n+            if (_inputFormat == NativeCommandIOFormat.Xml)\n             {\n-                _thread.Name = string.Format(CultureInfo.InvariantCulture, \"{0} :Output Reader\", _processPath);\n+                _streamWriter.WriteLine(ProcessOutputHandler.XmlCliTag);\n+                _xmlSerializer = new Serializer(XmlWriter.Create(_streamWriter));\n             }\n-            else\n+            else // Text\n             {\n-                _thread.Name = string.Format(CultureInfo.InvariantCulture, \"{0} :Error Reader\", _processPath);\n+                _pipeline = ScriptBlock.Create(\"Out-String -Stream\").GetSteppablePipeline();\n+                _pipeline.Begin(true);\n             }\n-            _thread.Start();\n         }\n \n+        bool _stopping = false;\n+\n         /// <summary>\n-        /// This method returns when reader thread has returned.\n+        /// Stop writing input to process\n         /// </summary>\n-        internal void Done()\n+        internal void Stop()\n         {\n-            if (_thread != null)\n-            {\n-                _thread.Join();\n-            }\n+            _stopping = true;\n         }\n \n         /// <summary>\n-        /// Thread proc for reading\n+        /// This method wait for writer thread to finish.\n         /// </summary>\n-        private void ReaderStartProc()\n+        internal void Done()\n         {\n-            try\n-            {\n-                ReaderStartProcHelper();\n-            }\n-            catch (Exception ex)\n-            {\n-                CommandProcessorBase.CheckForSevereException(ex);\n-            }\n-            finally\n+            // we allow call Done() multiply times.\n+            // For example one time from ProcessRecord() code path,\n+            // when we detect that process already finished\n+            // and once from Complete() code path.\n+            // Even though Done() could be called multiple times,\n+            // the calls are on the same thread, so there is no race condition.\n+            if (_pipeline != null)\n             {\n-                _processOutputReader.ReaderDone(_isOutput);\n+                var finalResults = _pipeline.End();\n+                _pipeline.Dispose();\n+                _pipeline = null;\n+                // AddTextInputFromFormattedArray can recursively call Done(),\n+                // if the downstream process already exited.\n+                // to Prevent it, we first finalize the pipeline and set it to null,\n+                // then calling the result processing for the last time\n+                AddTextInputFromFormattedArray(finalResults);","path":"src/System.Management.Automation/engine/NativeCommandProcessor.cs","position":1334,"original_position":1334,"commit_id":"9aa165b9796cbe82f6f8cdf462c456dce58660b9","original_commit_id":"9aa165b9796cbe82f6f8cdf462c456dce58660b9","user":{"login":"vors","id":816680,"avatar_url":"https://avatars.githubusercontent.com/u/816680?v=3","gravatar_id":"","url":"https://api.github.com/users/vors","html_url":"https://github.com/vors","followers_url":"https://api.github.com/users/vors/followers","following_url":"https://api.github.com/users/vors/following{/other_user}","gists_url":"https://api.github.com/users/vors/gists{/gist_id}","starred_url":"https://api.github.com/users/vors/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vors/subscriptions","organizations_url":"https://api.github.com/users/vors/orgs","repos_url":"https://api.github.com/users/vors/repos","events_url":"https://api.github.com/users/vors/events{/privacy}","received_events_url":"https://api.github.com/users/vors/received_events","type":"User","site_admin":false},"body":"Good suggestions. 👍 👍 \r\nI also don't like how cumbersome this code became.\r\nLet me re-think it a little bit and I'll probably do something similar to your suggestion.","created_at":"2016-11-15T22:36:53Z","updated_at":"2016-11-15T22:36:53Z","html_url":"https://github.com/PowerShell/PowerShell/pull/2450#discussion_r88130158","pull_request_url":"https://api.github.com/repos/PowerShell/PowerShell/pulls/2450","_links":{"self":{"href":"https://api.github.com/repos/PowerShell/PowerShell/pulls/comments/88130158"},"html":{"href":"https://github.com/PowerShell/PowerShell/pull/2450#discussion_r88130158"},"pull_request":{"href":"https://api.github.com/repos/PowerShell/PowerShell/pulls/2450"}}},"pull_request":{"url":"https://api.github.com/repos/PowerShell/PowerShell/pulls/2450","id":88760047,"html_url":"https://github.com/PowerShell/PowerShell/pull/2450","diff_url":"https://github.com/PowerShell/PowerShell/pull/2450.diff","patch_url":"https://github.com/PowerShell/PowerShell/pull/2450.patch","issue_url":"https://api.github.com/repos/PowerShell/PowerShell/issues/2450","number":2450,"state":"open","locked":false,"title":"Native pipe","user":{"login":"vors","id":816680,"avatar_url":"https://avatars.githubusercontent.com/u/816680?v=3","gravatar_id":"","url":"https://api.github.com/users/vors","html_url":"https://github.com/vors","followers_url":"https://api.github.com/users/vors/followers","following_url":"https://api.github.com/users/vors/following{/other_user}","gists_url":"https://api.github.com/users/vors/gists{/gist_id}","starred_url":"https://api.github.com/users/vors/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vors/subscriptions","organizations_url":"https://api.github.com/users/vors/orgs","repos_url":"https://api.github.com/users/vors/repos","events_url":"https://api.github.com/users/vors/events{/privacy}","received_events_url":"https://api.github.com/users/vors/received_events","type":"User","site_admin":false},"body":"Fix #559 \r\nImprove pipeline for native commands\r\n- Start native process in Prepare() instead of Complete()\r\n  in NativeCommandProcessor.\r\n- Remove unneeded input, output, error threads. Replaced by\r\n  simpler primitives.\r\n\r\nMotivation example:\r\n\r\n`ping | %{powershell-code} | grep` now works in real-time.\r\nSee the [test](https://github.com/PowerShell/PowerShell/pull/2450/commits/e34a81a30fda95685084b09d4a0d5f60006e4d49#diff-e2e61256a6481fbfcebf5757b220793aR35)\r\n\r\nTODO before merging:\r\n\r\n- [x] replace spin-lock by something like `notify/wait`\r\n- [x] resurrect and test minishell with custom host serialization\r\n- [x] Changelog entry\r\n- [x] Fix feature tests for https://github.com/PowerShell/PowerShell/blob/ae7b857c235203d1fd3708fe05c076e5a376859a/test/powershell/Host/ConsoleHost.Tests.ps1#L57\r\n- [x] Investigate, is this test flaky or not https://ci.appveyor.com/project/PowerShell/powershell/build/6.0.0-alpha.12-1287/tests","created_at":"2016-10-11T02:06:00Z","updated_at":"2016-11-15T22:36:53Z","closed_at":null,"merged_at":null,"merge_commit_sha":"ad3374db3066ba08b735ae9da20de6400f6c174f","assignee":{"login":"daxian-dbw","id":127450,"avatar_url":"https://avatars.githubusercontent.com/u/127450?v=3","gravatar_id":"","url":"https://api.github.com/users/daxian-dbw","html_url":"https://github.com/daxian-dbw","followers_url":"https://api.github.com/users/daxian-dbw/followers","following_url":"https://api.github.com/users/daxian-dbw/following{/other_user}","gists_url":"https://api.github.com/users/daxian-dbw/gists{/gist_id}","starred_url":"https://api.github.com/users/daxian-dbw/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/daxian-dbw/subscriptions","organizations_url":"https://api.github.com/users/daxian-dbw/orgs","repos_url":"https://api.github.com/users/daxian-dbw/repos","events_url":"https://api.github.com/users/daxian-dbw/events{/privacy}","received_events_url":"https://api.github.com/users/daxian-dbw/received_events","type":"User","site_admin":false},"assignees":[{"login":"daxian-dbw","id":127450,"avatar_url":"https://avatars.githubusercontent.com/u/127450?v=3","gravatar_id":"","url":"https://api.github.com/users/daxian-dbw","html_url":"https://github.com/daxian-dbw","followers_url":"https://api.github.com/users/daxian-dbw/followers","following_url":"https://api.github.com/users/daxian-dbw/following{/other_user}","gists_url":"https://api.github.com/users/daxian-dbw/gists{/gist_id}","starred_url":"https://api.github.com/users/daxian-dbw/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/daxian-dbw/subscriptions","organizations_url":"https://api.github.com/users/daxian-dbw/orgs","repos_url":"https://api.github.com/users/daxian-dbw/repos","events_url":"https://api.github.com/users/daxian-dbw/events{/privacy}","received_events_url":"https://api.github.com/users/daxian-dbw/received_events","type":"User","site_admin":false}],"milestone":null,"commits_url":"https://api.github.com/repos/PowerShell/PowerShell/pulls/2450/commits","review_comments_url":"https://api.github.com/repos/PowerShell/PowerShell/pulls/2450/comments","review_comment_url":"https://api.github.com/repos/PowerShell/PowerShell/pulls/comments{/number}","comments_url":"https://api.github.com/repos/PowerShell/PowerShell/issues/2450/comments","statuses_url":"https://api.github.com/repos/PowerShell/PowerShell/statuses/9aa165b9796cbe82f6f8cdf462c456dce58660b9","head":{"label":"vors:native-pipe","ref":"native-pipe","sha":"9aa165b9796cbe82f6f8cdf462c456dce58660b9","user":{"login":"vors","id":816680,"avatar_url":"https://avatars.githubusercontent.com/u/816680?v=3","gravatar_id":"","url":"https://api.github.com/users/vors","html_url":"https://github.com/vors","followers_url":"https://api.github.com/users/vors/followers","following_url":"https://api.github.com/users/vors/following{/other_user}","gists_url":"https://api.github.com/users/vors/gists{/gist_id}","starred_url":"https://api.github.com/users/vors/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vors/subscriptions","organizations_url":"https://api.github.com/users/vors/orgs","repos_url":"https://api.github.com/users/vors/repos","events_url":"https://api.github.com/users/vors/events{/privacy}","received_events_url":"https://api.github.com/users/vors/received_events","type":"User","site_admin":false},"repo":{"id":66054642,"name":"PowerShell","full_name":"vors/PowerShell","owner":{"login":"vors","id":816680,"avatar_url":"https://avatars.githubusercontent.com/u/816680?v=3","gravatar_id":"","url":"https://api.github.com/users/vors","html_url":"https://github.com/vors","followers_url":"https://api.github.com/users/vors/followers","following_url":"https://api.github.com/users/vors/following{/other_user}","gists_url":"https://api.github.com/users/vors/gists{/gist_id}","starred_url":"https://api.github.com/users/vors/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vors/subscriptions","organizations_url":"https://api.github.com/users/vors/orgs","repos_url":"https://api.github.com/users/vors/repos","events_url":"https://api.github.com/users/vors/events{/privacy}","received_events_url":"https://api.github.com/users/vors/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/vors/PowerShell","description":"PowerShell for every system!","fork":true,"url":"https://api.github.com/repos/vors/PowerShell","forks_url":"https://api.github.com/repos/vors/PowerShell/forks","keys_url":"https://api.github.com/repos/vors/PowerShell/keys{/key_id}","collaborators_url":"https://api.github.com/repos/vors/PowerShell/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/vors/PowerShell/teams","hooks_url":"https://api.github.com/repos/vors/PowerShell/hooks","issue_events_url":"https://api.github.com/repos/vors/PowerShell/issues/events{/number}","events_url":"https://api.github.com/repos/vors/PowerShell/events","assignees_url":"https://api.github.com/repos/vors/PowerShell/assignees{/user}","branches_url":"https://api.github.com/repos/vors/PowerShell/branches{/branch}","tags_url":"https://api.github.com/repos/vors/PowerShell/tags","blobs_url":"https://api.github.com/repos/vors/PowerShell/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/vors/PowerShell/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/vors/PowerShell/git/refs{/sha}","trees_url":"https://api.github.com/repos/vors/PowerShell/git/trees{/sha}","statuses_url":"https://api.github.com/repos/vors/PowerShell/statuses/{sha}","languages_url":"https://api.github.com/repos/vors/PowerShell/languages","stargazers_url":"https://api.github.com/repos/vors/PowerShell/stargazers","contributors_url":"https://api.github.com/repos/vors/PowerShell/contributors","subscribers_url":"https://api.github.com/repos/vors/PowerShell/subscribers","subscription_url":"https://api.github.com/repos/vors/PowerShell/subscription","commits_url":"https://api.github.com/repos/vors/PowerShell/commits{/sha}","git_commits_url":"https://api.github.com/repos/vors/PowerShell/git/commits{/sha}","comments_url":"https://api.github.com/repos/vors/PowerShell/comments{/number}","issue_comment_url":"https://api.github.com/repos/vors/PowerShell/issues/comments{/number}","contents_url":"https://api.github.com/repos/vors/PowerShell/contents/{+path}","compare_url":"https://api.github.com/repos/vors/PowerShell/compare/{base}...{head}","merges_url":"https://api.github.com/repos/vors/PowerShell/merges","archive_url":"https://api.github.com/repos/vors/PowerShell/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/vors/PowerShell/downloads","issues_url":"https://api.github.com/repos/vors/PowerShell/issues{/number}","pulls_url":"https://api.github.com/repos/vors/PowerShell/pulls{/number}","milestones_url":"https://api.github.com/repos/vors/PowerShell/milestones{/number}","notifications_url":"https://api.github.com/repos/vors/PowerShell/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/vors/PowerShell/labels{/name}","releases_url":"https://api.github.com/repos/vors/PowerShell/releases{/id}","deployments_url":"https://api.github.com/repos/vors/PowerShell/deployments","created_at":"2016-08-19T05:14:51Z","updated_at":"2016-08-19T05:14:59Z","pushed_at":"2016-11-14T19:27:40Z","git_url":"git://github.com/vors/PowerShell.git","ssh_url":"git@github.com:vors/PowerShell.git","clone_url":"https://github.com/vors/PowerShell.git","svn_url":"https://github.com/vors/PowerShell","homepage":"https://microsoft.com/powershell","size":20768,"stargazers_count":0,"watchers_count":0,"language":"C#","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"PowerShell:master","ref":"master","sha":"bd9e49d8d979f33470b6e431207296fd46e13238","user":{"login":"PowerShell","id":11524380,"avatar_url":"https://avatars.githubusercontent.com/u/11524380?v=3","gravatar_id":"","url":"https://api.github.com/users/PowerShell","html_url":"https://github.com/PowerShell","followers_url":"https://api.github.com/users/PowerShell/followers","following_url":"https://api.github.com/users/PowerShell/following{/other_user}","gists_url":"https://api.github.com/users/PowerShell/gists{/gist_id}","starred_url":"https://api.github.com/users/PowerShell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/PowerShell/subscriptions","organizations_url":"https://api.github.com/users/PowerShell/orgs","repos_url":"https://api.github.com/users/PowerShell/repos","events_url":"https://api.github.com/users/PowerShell/events{/privacy}","received_events_url":"https://api.github.com/users/PowerShell/received_events","type":"Organization","site_admin":false},"repo":{"id":49609581,"name":"PowerShell","full_name":"PowerShell/PowerShell","owner":{"login":"PowerShell","id":11524380,"avatar_url":"https://avatars.githubusercontent.com/u/11524380?v=3","gravatar_id":"","url":"https://api.github.com/users/PowerShell","html_url":"https://github.com/PowerShell","followers_url":"https://api.github.com/users/PowerShell/followers","following_url":"https://api.github.com/users/PowerShell/following{/other_user}","gists_url":"https://api.github.com/users/PowerShell/gists{/gist_id}","starred_url":"https://api.github.com/users/PowerShell/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/PowerShell/subscriptions","organizations_url":"https://api.github.com/users/PowerShell/orgs","repos_url":"https://api.github.com/users/PowerShell/repos","events_url":"https://api.github.com/users/PowerShell/events{/privacy}","received_events_url":"https://api.github.com/users/PowerShell/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/PowerShell/PowerShell","description":"PowerShell for every system!","fork":false,"url":"https://api.github.com/repos/PowerShell/PowerShell","forks_url":"https://api.github.com/repos/PowerShell/PowerShell/forks","keys_url":"https://api.github.com/repos/PowerShell/PowerShell/keys{/key_id}","collaborators_url":"https://api.github.com/repos/PowerShell/PowerShell/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/PowerShell/PowerShell/teams","hooks_url":"https://api.github.com/repos/PowerShell/PowerShell/hooks","issue_events_url":"https://api.github.com/repos/PowerShell/PowerShell/issues/events{/number}","events_url":"https://api.github.com/repos/PowerShell/PowerShell/events","assignees_url":"https://api.github.com/repos/PowerShell/PowerShell/assignees{/user}","branches_url":"https://api.github.com/repos/PowerShell/PowerShell/branches{/branch}","tags_url":"https://api.github.com/repos/PowerShell/PowerShell/tags","blobs_url":"https://api.github.com/repos/PowerShell/PowerShell/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/PowerShell/PowerShell/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/PowerShell/PowerShell/git/refs{/sha}","trees_url":"https://api.github.com/repos/PowerShell/PowerShell/git/trees{/sha}","statuses_url":"https://api.github.com/repos/PowerShell/PowerShell/statuses/{sha}","languages_url":"https://api.github.com/repos/PowerShell/PowerShell/languages","stargazers_url":"https://api.github.com/repos/PowerShell/PowerShell/stargazers","contributors_url":"https://api.github.com/repos/PowerShell/PowerShell/contributors","subscribers_url":"https://api.github.com/repos/PowerShell/PowerShell/subscribers","subscription_url":"https://api.github.com/repos/PowerShell/PowerShell/subscription","commits_url":"https://api.github.com/repos/PowerShell/PowerShell/commits{/sha}","git_commits_url":"https://api.github.com/repos/PowerShell/PowerShell/git/commits{/sha}","comments_url":"https://api.github.com/repos/PowerShell/PowerShell/comments{/number}","issue_comment_url":"https://api.github.com/repos/PowerShell/PowerShell/issues/comments{/number}","contents_url":"https://api.github.com/repos/PowerShell/PowerShell/contents/{+path}","compare_url":"https://api.github.com/repos/PowerShell/PowerShell/compare/{base}...{head}","merges_url":"https://api.github.com/repos/PowerShell/PowerShell/merges","archive_url":"https://api.github.com/repos/PowerShell/PowerShell/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/PowerShell/PowerShell/downloads","issues_url":"https://api.github.com/repos/PowerShell/PowerShell/issues{/number}","pulls_url":"https://api.github.com/repos/PowerShell/PowerShell/pulls{/number}","milestones_url":"https://api.github.com/repos/PowerShell/PowerShell/milestones{/number}","notifications_url":"https://api.github.com/repos/PowerShell/PowerShell/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/PowerShell/PowerShell/labels{/name}","releases_url":"https://api.github.com/repos/PowerShell/PowerShell/releases{/id}","deployments_url":"https://api.github.com/repos/PowerShell/PowerShell/deployments","created_at":"2016-01-13T23:41:35Z","updated_at":"2016-11-15T19:44:43Z","pushed_at":"2016-11-15T19:54:18Z","git_url":"git://github.com/PowerShell/PowerShell.git","ssh_url":"git@github.com:PowerShell/PowerShell.git","clone_url":"https://github.com/PowerShell/PowerShell.git","svn_url":"https://github.com/PowerShell/PowerShell","homepage":"https://microsoft.com/powershell","size":27382,"stargazers_count":5543,"watchers_count":5543,"language":"C#","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":646,"mirror_url":null,"open_issues_count":471,"forks":646,"open_issues":471,"watchers":5543,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/PowerShell/PowerShell/pulls/2450"},"html":{"href":"https://github.com/PowerShell/PowerShell/pull/2450"},"issue":{"href":"https://api.github.com/repos/PowerShell/PowerShell/issues/2450"},"comments":{"href":"https://api.github.com/repos/PowerShell/PowerShell/issues/2450/comments"},"review_comments":{"href":"https://api.github.com/repos/PowerShell/PowerShell/pulls/2450/comments"},"review_comment":{"href":"https://api.github.com/repos/PowerShell/PowerShell/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/PowerShell/PowerShell/pulls/2450/commits"},"statuses":{"href":"https://api.github.com/repos/PowerShell/PowerShell/statuses/9aa165b9796cbe82f6f8cdf462c456dce58660b9"}}}}
{"actor":{"display_login":"vors"}}