/********************************************************************++ Copyright (c) Microsoft Corporation. All rights reserved. --********************************************************************/ using System.Collections; using System.Collections.Generic; using System.Management.Automation.Runspaces; namespace System.Management.Automation.Internal { /// /// Corresponds to -OutputVariable, -ErrorVariable, -WarningVariable, and -InformationVariable. /// internal enum VariableStreamKind { Output, Error, Warning, Information }; /// /// Pipe provides a way to stitch two commands. /// /// /// The Pipe class is not thread-safe, so methods such as /// AddItems and Retrieve should not be called simultaneously. /// ExternalReader and ExternalWriter can provide thread-safe buffering. /// internal class Pipe { private ExecutionContext _context; // If a pipeline object has been added, then // write objects to it, stepping one at a time... internal PipelineProcessor PipelineProcessor { get; } /// /// This is the downstream cmdlet in the "streamlet model" /// which is invoked during each call to Add/AddItems. /// internal CommandProcessorBase DownstreamCmdlet { get { return _downstreamCmdlet; } set { Diagnostics.Assert(_resultList == null, "Tried to set downstream cmdlet when _resultList not null"); _downstreamCmdlet = value; } } private CommandProcessorBase _downstreamCmdlet; /// /// This is the upstream external object source. If this is set, /// Retrieve() will attempt to read objects from the upstream source /// before indicating that the pipe is empty. /// /// It is improper to change this once the pipeline has started /// executing, although the checks for this are in the /// PipelineProcessor class and not here. /// /// internal PipelineReader ExternalReader { get; set; } /// /// This is the downstream object recipient. If this is set, /// Add() and AddItems() write to this recipient instead of /// to the internal queue. This also disables the /// DownstreamCmdlet. /// /// It is improper to change this once the pipeline has started /// executing, although the checks for this are in the /// PipelineProcessor class and not here. /// /// internal PipelineWriter ExternalWriter { get { return _externalWriter; } set { Diagnostics.Assert(_resultList == null, "Tried to set Pipe ExternalWriter when resultList not null"); _externalWriter = value; } } private PipelineWriter _externalWriter; /// /// for diagnostic purposes /// /// public override string ToString() { if (_downstreamCmdlet != null) return _downstreamCmdlet.ToString(); return base.ToString(); } /// /// OutBufferCount configures the number of objects to buffer before calling the downstream Cmdlet /// internal int OutBufferCount { get; set; } = 0; /// /// If true, then all input added to this pipe will simply be discarded... /// internal bool NullPipe { get { return _nullPipe; } set { _isRedirected = true; _nullPipe = value; } } private bool _nullPipe; /// /// A queue that is shared between commands on either side of the pipe to transfer objects. /// internal Queue ObjectQueue { get; } /// /// True if there are items in this pipe that need processing... /// /// This does not take into account the presence of ExternalInput; /// it only indicates whether there is currently any data queued up /// or if there is data in the enumerator... /// /// internal bool Empty { get { if (_enumeratorToProcess != null) return _enumeratorToProcessIsEmpty; if (ObjectQueue != null) return ObjectQueue.Count == 0; return true; } } /// /// Is true if there is someone consuming this pipe already, either through /// a Pipe object that processes it's output or there is downstream cmdlet... /// internal bool IsRedirected { get { return _downstreamCmdlet != null || _isRedirected; } } private bool _isRedirected; /// /// If non-null, output written to the pipe are also added to this list. /// private List _outVariableList; /// /// If non-null, errors written to the pipe are also added to this list. /// private List _errorVariableList; /// /// If non-null, warnings written to the pipe are also added to this list. /// private List _warningVariableList; /// /// If non-null, information objects written to the pipe are also added to this list. /// private List _informationVariableList; /// /// If non-null, the current object being written to the pipe is stored in /// this variable. /// private PSVariable _pipelineVariableObject; private static void AddToVarList(List varList, object obj) { if (varList != null && varList.Count > 0) { for (int i = 0; i < varList.Count; i++) { varList[i].Add(obj); } } } internal void AppendVariableList(VariableStreamKind kind, object obj) { switch (kind) { case VariableStreamKind.Error: AddToVarList(_errorVariableList, obj); break; case VariableStreamKind.Warning: AddToVarList(_warningVariableList, obj); break; case VariableStreamKind.Output: AddToVarList(_outVariableList, obj); break; case VariableStreamKind.Information: AddToVarList(_informationVariableList, obj); break; } } internal void AddVariableList(VariableStreamKind kind, IList list) { switch (kind) { case VariableStreamKind.Error: if (_errorVariableList == null) { _errorVariableList = new List(); } _errorVariableList.Add(list); break; case VariableStreamKind.Warning: if (_warningVariableList == null) { _warningVariableList = new List(); } _warningVariableList.Add(list); break; case VariableStreamKind.Output: if (_outVariableList == null) { _outVariableList = new List(); } _outVariableList.Add(list); break; case VariableStreamKind.Information: if (_informationVariableList == null) { _informationVariableList = new List(); } _informationVariableList.Add(list); break; } } internal void SetPipelineVariable(PSVariable pipelineVariable) { _pipelineVariableObject = pipelineVariable; } internal void RemoveVariableList(VariableStreamKind kind, IList list) { switch (kind) { case VariableStreamKind.Error: _errorVariableList.Remove(list); break; case VariableStreamKind.Warning: _warningVariableList.Remove(list); break; case VariableStreamKind.Output: _outVariableList.Remove(list); break; case VariableStreamKind.Information: _informationVariableList.Remove(list); break; } } internal void RemovePipelineVariable() { if (_pipelineVariableObject != null) { _pipelineVariableObject.Value = null; _pipelineVariableObject = null; } } /// /// When a temporary pipe is used in the middle of execution, then we need to pass along /// the error and warning variable list to hold the errors and warnings get written out /// while the temporary pipe is being used. /// /// We don't need to pass along the out variable list because we don't care about the output /// generated in the middle of execution. /// internal void SetVariableListForTemporaryPipe(Pipe tempPipe) { CopyVariableToTempPipe(VariableStreamKind.Error, _errorVariableList, tempPipe); CopyVariableToTempPipe(VariableStreamKind.Warning, _warningVariableList, tempPipe); CopyVariableToTempPipe(VariableStreamKind.Information, _informationVariableList, tempPipe); } private void CopyVariableToTempPipe(VariableStreamKind streamKind, List variableList, Pipe tempPipe) { if (variableList != null && variableList.Count > 0) { for (int i = 0; i < variableList.Count; i++) { tempPipe.AddVariableList(streamKind, variableList[i]); } } } #region ctor /// /// Default constructor - Creates the object queue /// /// /// The initial Queue capacity is 1, but it will grow automatically. /// internal Pipe() { ObjectQueue = new Queue(); } /// /// This overload causes output to be written into a List /// /// internal Pipe(List resultList) { Diagnostics.Assert(resultList != null, "resultList cannot be null"); _isRedirected = true; _resultList = resultList; } private readonly List _resultList; /// /// This overload causes output to be /// written onto an Collection[PSObject] which is more useful /// in many circumstances than arraylist /// /// The collection to write into internal Pipe(System.Collections.ObjectModel.Collection resultCollection) { Diagnostics.Assert(resultCollection != null, "resultCollection cannot be null"); _isRedirected = true; _resultCollection = resultCollection; } private System.Collections.ObjectModel.Collection _resultCollection; /// /// This pipe writes into another pipeline processor allowing /// pipelines to be chained together... /// /// The execution context object for this engine instance /// The pipeline to write into... internal Pipe(ExecutionContext context, PipelineProcessor outputPipeline) { Diagnostics.Assert(outputPipeline != null, "outputPipeline cannot be null"); Diagnostics.Assert(outputPipeline != null, "context cannot be null"); _isRedirected = true; _context = context; PipelineProcessor = outputPipeline; } /// /// Read from an enumerator instead of a pipeline reader... /// /// The enumerator to process... internal Pipe(IEnumerator enumeratorToProcess) { Diagnostics.Assert(enumeratorToProcess != null, "enumeratorToProcess cannot be null"); _enumeratorToProcess = enumeratorToProcess; // since there is an enumerator specified, we // assume that there is some stuff to read _enumeratorToProcessIsEmpty = false; } private IEnumerator _enumeratorToProcess; private bool _enumeratorToProcessIsEmpty; #endregion ctor /// /// Writes an object to the pipe. This could recursively call to the /// downstream cmdlet, or write the object to the external output. /// /// The object to add to the pipe /// /// AutomationNull.Value is ignored /// /// /// a terminating error occurred, or the pipeline was otherwise stopped /// /// /// The ExternalWriter stream is closed /// internal void Add(object obj) { if (obj == AutomationNull.Value) return; // OutVariable is appended for null pipes so that the following works: // foo -OutVariable bar > $null AddToVarList(_outVariableList, obj); if (_nullPipe) return; // Store the current pipeline variable if (_pipelineVariableObject != null) { _pipelineVariableObject.Value = obj; } AddToPipe(obj); } internal void AddWithoutAppendingOutVarList(object obj) { if (obj == AutomationNull.Value || _nullPipe) return; AddToPipe(obj); } private void AddToPipe(object obj) { if (PipelineProcessor != null) { // Put the pipeline on the notification stack for stop. _context.PushPipelineProcessor(PipelineProcessor); PipelineProcessor.Step(obj); _context.PopPipelineProcessor(false); } else if (_resultCollection != null) { _resultCollection.Add(obj != null ? PSObject.AsPSObject(obj) : null); } else if (_resultList != null) { _resultList.Add(obj); } else if (null != _externalWriter) { _externalWriter.Write(obj); } else if (ObjectQueue != null) { ObjectQueue.Enqueue(obj); // This is the "streamlet" recursive call if (null != _downstreamCmdlet && ObjectQueue.Count > OutBufferCount) { _downstreamCmdlet.DoExecute(); } } } /// /// Writes a set of objects to the pipe. This could recursively /// call to the downstream cmdlet, or write the objects to the /// external output. /// /// /// Each of the objects are added to the pipe /// /// /// The pipeline has already been stopped, /// or a terminating error occurred in a downstream cmdlet. /// /// /// The ExternalWriter stream is closed /// internal void AddItems(object objects) { // Use the extended type system to try and get an enumerator for the object being added. // If we get an enumerator, then add the individual elements. If the object isn't // enumerable (i.e. the call returned null) then add the object to the pipe // as a single element. IEnumerator ie = LanguagePrimitives.GetEnumerator(objects); try { if (ie == null) { Add(objects); } else { while (ParserOps.MoveNext(_context, null, ie)) { object o = ParserOps.Current(null, ie); // Slip over any instance of AutomationNull.Value in the pipeline... if (o == AutomationNull.Value) { continue; } Add(o); } } } finally { // If our object came from GetEnumerator (and hence is not IEnumerator), then we need to dispose // Otherwise, we don't own the object, so don't dispose. var disposable = ie as IDisposable; if (disposable != null && !(objects is IEnumerator)) { disposable.Dispose(); } } if (null != _externalWriter) return; // If there are objects waiting for the downstream command // call it now if (_downstreamCmdlet != null && ObjectQueue != null && ObjectQueue.Count > OutBufferCount) { _downstreamCmdlet.DoExecute(); } } // internal void AddItems(object objects) /// /// Returns an object from the pipe. If pipe is empty returns null. /// This will try the ExternalReader if there are no queued objects. /// /// /// object that is retrieved, or AutomationNull.Value if none /// internal object Retrieve() { if (ObjectQueue != null && ObjectQueue.Count != 0) { return ObjectQueue.Dequeue(); } else if (_enumeratorToProcess != null) { if (_enumeratorToProcessIsEmpty) return AutomationNull.Value; if (!ParserOps.MoveNext(_context, null, _enumeratorToProcess)) { _enumeratorToProcessIsEmpty = true; return AutomationNull.Value; } return ParserOps.Current(null, _enumeratorToProcess); } else if (null != ExternalReader) { try { object o = ExternalReader.Read(); if (AutomationNull.Value == o) { // NOTICE-2004/06/08-JonN 963367 // The fix to this bug involves making one last // attempt to read from the pipeline in DoComplete. // We should be sure to not hit the ExternalReader // again if it already reported completion. ExternalReader = null; } return o; } catch (PipelineClosedException) { return AutomationNull.Value; } catch (ObjectDisposedException) { return AutomationNull.Value; } } else return AutomationNull.Value; } /// /// Removes all the objects from the Pipe. /// internal void Clear() { if (ObjectQueue != null) ObjectQueue.Clear(); } /// /// Returns the currently queued items in the pipe. Note that this will /// not block on ExternalInput, and it does not modify the contents of /// the pipe. /// /// possibly empty array of objects, but not null internal object[] ToArray() { if (ObjectQueue == null || ObjectQueue.Count == 0) return MshCommandRuntime.StaticEmptyArray; return ObjectQueue.ToArray(); } } } // namespace System.Management.Automation.Internal