// Copyright (c) Microsoft Corporation.// Licensed under the MIT License.using System;using System.Collections.ObjectModel;using System.Management.Automation;using System.Management.Automation.Runspaces;using System.Runtime.InteropServices;using System.Threading;namespace System.Management.Automation.Internal{/// <summary>/// A PipelineReader for an ObjectStream./// </summary>/// <remarks>/// This class is not safe for multi-threaded operations./// </remarks>internal abstract class ObjectReaderBase<T> : PipelineReader<T>, IDisposable{/// <summary>/// Construct with an existing ObjectStream./// </summary>/// <param name="stream">The stream to read.</param>/// <exception cref="ArgumentNullException">Thrown if the specified stream is null.</exception>public ObjectReaderBase([In, Out] ObjectStreamBase stream){if (stream == null){throw new ArgumentNullException("stream", "stream may not be null");}_stream = stream;}#region Events/// <summary>/// Event fired when objects are added to the underlying stream./// </summary>public override event EventHandler DataReady{add{lock (_monitorObject){bool firstRegistrant = (InternalDataReady == null);InternalDataReady += value;if (firstRegistrant){_stream.DataReady += new EventHandler(this.OnDataReady);}}}remove{lock (_monitorObject){InternalDataReady -= value;if (InternalDataReady == null){_stream.DataReady -= new EventHandler(this.OnDataReady);}}}}public event EventHandler InternalDataReady = null;#endregion Events#region Public Properties/// <summary>/// Waitable handle for caller's to block until data is ready to read from the underlying stream./// </summary>public override WaitHandle WaitHandle{get{return _stream.ReadHandle;}}/// <summary>/// Check if the stream is closed and contains no data./// </summary>/// <value>True if the stream is closed and contains no data, otherwise; false.</value>/// <remarks>/// Attempting to read from the underlying stream if EndOfPipeline is true returns/// zero objects./// </remarks>public override bool EndOfPipeline{get{return _stream.EndOfPipeline;}}/// <summary>/// Check if the stream is open for further writes./// </summary>/// <value>true if the underlying stream is open, otherwise; false.</value>/// <remarks>/// The underlying stream may be readable after it is closed if data remains in the/// internal buffer. Check <see cref="EndOfPipeline"/> to determine if/// the underlying stream is closed and contains no data./// </remarks>public override bool IsOpen{get{return _stream.IsOpen;}}/// <summary>/// Returns the number of objects in the underlying stream./// </summary>public override int Count{get{return _stream.Count;}}/// <summary>/// Get the capacity of the stream./// </summary>/// <value>/// The capacity of the stream./// </value>/// <remarks>/// The capacity is the number of objects that stream may contain at one time. Once this/// limit is reached, attempts to write into the stream block until buffer space/// becomes available./// </remarks>public override int MaxCapacity{get{return _stream.MaxCapacity;}}#endregion Public Properties#region Public Methods/// <summary>/// Close the stream./// </summary>/// <remarks>/// Causes subsequent calls to IsOpen to return false and calls to/// a write operation to throw an ObjectDisposedException./// All calls to Close() after the first call are silently ignored./// </remarks>/// <exception cref="ObjectDisposedException">/// The stream is already disposed/// </exception>public override void Close(){// 2003年09月02日-JonN added call to close underlying stream_stream.Close();}#endregion Public Methods#region Private Methods/// <summary>/// Handle DataReady events from the underlying stream./// </summary>/// <param name="sender">The stream raising the event.</param>/// <param name="args">Standard event args.</param>private void OnDataReady(object sender, EventArgs args){// call any event handlers on this, replacing the// ObjectStream sender with 'this' since receivers// are expecting a PipelineReader<object>InternalDataReady.SafeInvoke(this, args);}#endregion Private Methods#region Private fields/// <summary>/// The underlying stream./// </summary>/// <remarks>Can never be null</remarks>protected ObjectStreamBase _stream;/// <summary>/// This object is used to acquire an exclusive lock/// on event handler registration./// </summary>/// <remarks>/// Note that we lock _monitorObject rather than "this" so that/// we are protected from outside code interfering in our/// critical section. Thanks to Wintellect for the hint./// </remarks>private object _monitorObject = new object();#endregion Private fields#region IDisposable/// <summary>/// Public method for dispose./// </summary>public void Dispose(){Dispose(true);GC.SuppressFinalize(this);}/// <summary>/// Release all resources./// </summary>/// <param name="disposing">If true, release all managed resources.</param>protected abstract void Dispose(bool disposing);#endregion IDisposable}/// <summary>/// A PipelineReader reading objects from an ObjectStream./// </summary>/// <remarks>/// This class is not safe for multi-threaded operations./// </remarks>internal class ObjectReader : ObjectReaderBase<object>{#region ctor/// <summary>/// Construct with an existing ObjectStream./// </summary>/// <param name="stream">The stream to read.</param>/// <exception cref="ArgumentNullException">Thrown if the specified stream is null.</exception>public ObjectReader([In, Out] ObjectStream stream): base(stream){ }#endregion ctor/// <summary>/// Read at most <paramref name="count"/> objects./// </summary>/// <param name="count">The maximum number of objects to read.</param>/// <returns>The objects read.</returns>/// <remarks>/// This method blocks if the number of objects in the stream is less than <paramref name="count"/>/// and the stream is not closed./// </remarks>public override Collection<object> Read(int count){return _stream.Read(count);}/// <summary>/// Read a single object from the stream./// </summary>/// <returns>The next object in the stream.</returns>/// <remarks>This method blocks if the stream is empty</remarks>public override object Read(){return _stream.Read();}/// <summary>/// Blocks until the pipeline closes and reads all objects./// </summary>/// <returns>A collection of zero or more objects.</returns>/// <remarks>/// If the stream is empty, an empty collection is returned./// </remarks>public override Collection<object> ReadToEnd(){return _stream.ReadToEnd();}/// <summary>/// Reads all objects currently in the stream, but does not block./// </summary>/// <returns>A collection of zero or more objects.</returns>/// <remarks>/// This method performs a read of all objects currently in the/// stream. The method will block until exclusive access to the/// stream is acquired. If there are no objects in the stream,/// an empty collection is returned./// </remarks>public override Collection<object> NonBlockingRead(){return _stream.NonBlockingRead(Int32.MaxValue);}/// <summary>/// Reads objects currently in the stream, but does not block./// </summary>/// <returns>A collection of zero or more objects.</returns>/// <remarks>/// This method performs a read of objects currently in the/// stream. The method will block until exclusive access to the/// stream is acquired. If there are no objects in the stream,/// an empty collection is returned./// </remarks>/// <param name="maxRequested">/// Return no more than maxRequested objects./// </param>public override Collection<object> NonBlockingRead(int maxRequested){return _stream.NonBlockingRead(maxRequested);}/// <summary>/// Peek the next object./// </summary>/// <returns>The next object in the stream or ObjectStream.EmptyObject if the stream is empty.</returns>public override object Peek(){return _stream.Peek();}/// <summary>/// Release all resources./// </summary>/// <param name="disposing">If true, release all managed resources.</param>protected override void Dispose(bool disposing){if (disposing){_stream.Close();}}}/// <summary>/// A PipelineReader reading PSObjects from an ObjectStream./// </summary>/// <remarks>/// This class is not safe for multi-threaded operations./// </remarks>internal class PSObjectReader : ObjectReaderBase<PSObject>{#region ctor/// <summary>/// Construct with an existing ObjectStream./// </summary>/// <param name="stream">The stream to read.</param>/// <exception cref="ArgumentNullException">Thrown if the specified stream is null.</exception>public PSObjectReader([In, Out] ObjectStream stream): base(stream){ }#endregion ctor/// <summary>/// Read at most <paramref name="count"/> objects./// </summary>/// <param name="count">The maximum number of objects to read.</param>/// <returns>The objects read.</returns>/// <remarks>/// This method blocks if the number of objects in the stream is less than <paramref name="count"/>/// and the stream is not closed./// </remarks>public override Collection<PSObject> Read(int count){return MakePSObjectCollection(_stream.Read(count));}/// <summary>/// Read a single PSObject from the stream./// </summary>/// <returns>The next PSObject in the stream.</returns>/// <remarks>This method blocks if the stream is empty</remarks>public override PSObject Read(){return MakePSObject(_stream.Read());}/// <summary>/// Blocks until the pipeline closes and reads all objects./// </summary>/// <returns>A collection of zero or more objects.</returns>/// <remarks>/// If the stream is empty, an empty collection is returned./// </remarks>public override Collection<PSObject> ReadToEnd(){return MakePSObjectCollection(_stream.ReadToEnd());}/// <summary>/// Reads all objects currently in the stream, but does not block./// </summary>/// <returns>A collection of zero or more objects.</returns>/// <remarks>/// This method performs a read of all objects currently in the/// stream. The method will block until exclusive access to the/// stream is acquired. If there are no objects in the stream,/// an empty collection is returned./// </remarks>public override Collection<PSObject> NonBlockingRead(){return MakePSObjectCollection(_stream.NonBlockingRead(Int32.MaxValue));}/// <summary>/// Reads objects currently in the stream, but does not block./// </summary>/// <returns>A collection of zero or more objects.</returns>/// <remarks>/// This method performs a read of objects currently in the/// stream. The method will block until exclusive access to the/// stream is acquired. If there are no objects in the stream,/// an empty collection is returned./// </remarks>/// <param name="maxRequested">/// Return no more than maxRequested objects./// </param>public override Collection<PSObject> NonBlockingRead(int maxRequested){return MakePSObjectCollection(_stream.NonBlockingRead(maxRequested));}/// <summary>/// Peek the next PSObject./// </summary>/// <returns>The next PSObject in the stream or ObjectStream.EmptyObject if the stream is empty.</returns>public override PSObject Peek(){return MakePSObject(_stream.Peek());}/// <summary>/// Release all resources./// </summary>/// <param name="disposing">If true, release all managed resources.</param>protected override void Dispose(bool disposing){if (disposing){_stream.Close();}}#region Privateprivate static PSObject MakePSObject(object o){if (o == null)return null;return PSObject.AsPSObject(o);}// It might ultimately be more efficient to// make ObjectStream generic and convert the objects to PSObject// before inserting them into the initial Collection, so that we// don't have to convert the collection later.private static Collection<PSObject> MakePSObjectCollection(Collection<object> coll){if (coll == null)return null;Collection<PSObject> retval = new Collection<PSObject>();foreach (object o in coll){retval.Add(MakePSObject(o));}return retval;}#endregion Private}/// <summary>/// A ObjectReader for a PSDataCollection ObjectStream./// </summary>/// <remarks>/// PSDataCollection is introduced after 1.0. PSDataCollection is/// used to store data which can be used with different/// commands concurrently./// Only Read() operation is supported currently./// </remarks>internal class PSDataCollectionReader<DataStoreType, ReturnType>: ObjectReaderBase<ReturnType>{#region Private Dataprivate PSDataCollectionEnumerator<DataStoreType> _enumerator;#endregion#region ctor/// <summary>/// Construct with an existing ObjectStream./// </summary>/// <param name="stream">The stream to read.</param>/// <exception cref="ArgumentNullException">Thrown if the specified stream is null.</exception>public PSDataCollectionReader(PSDataCollectionStream<DataStoreType> stream): base(stream){System.Management.Automation.Diagnostics.Assert(stream.ObjectStore != null,"Stream should have a valid data store");_enumerator = (PSDataCollectionEnumerator<DataStoreType>)stream.ObjectStore.GetEnumerator();}#endregion ctor/// <summary>/// This method is not supported./// </summary>/// <param name="count">The maximum number of objects to read.</param>/// <returns>The objects read.</returns>public override Collection<ReturnType> Read(int count){throw new NotSupportedException();}/// <summary>/// Read a single object from the stream./// </summary>/// <returns>/// The next object in the buffer or AutomationNull if buffer is closed/// and data is not available./// </returns>/// <remarks>/// This method blocks if the buffer is empty./// </remarks>public override ReturnType Read(){object result = AutomationNull.Value;if (_enumerator.MoveNext()){result = _enumerator.Current;}return ConvertToReturnType(result);}/// <summary>/// This method is not supported./// </summary>/// <returns></returns>/// <remarks></remarks>public override Collection<ReturnType> ReadToEnd(){throw new NotSupportedException();}/// <summary>/// This method is not supported./// </summary>/// <returns></returns>/// <remarks></remarks>public override Collection<ReturnType> NonBlockingRead(){return NonBlockingRead(Int32.MaxValue);}/// <summary>/// This method is not supported./// </summary>/// <returns></returns>/// <remarks></remarks>/// <param name="maxRequested">/// Return no more than maxRequested objects./// </param>public override Collection<ReturnType> NonBlockingRead(int maxRequested){if (maxRequested < 0){throw PSTraceSource.NewArgumentOutOfRangeException("maxRequested", maxRequested);}if (maxRequested == 0){return new Collection<ReturnType>();}Collection<ReturnType> results = new Collection<ReturnType>();int readCount = maxRequested;while (readCount > 0){if (_enumerator.MoveNext(false)){results.Add(ConvertToReturnType(_enumerator.Current));continue;}break;}return results;}/// <summary>/// This method is not supported./// </summary>/// <returns></returns>public override ReturnType Peek(){throw new NotSupportedException();}/// <summary>/// Release all resources./// </summary>/// <param name="disposing">If true, release all managed resources.</param>protected override void Dispose(bool disposing){if (disposing){_stream.Close();}}private ReturnType ConvertToReturnType(object inputObject){Type resultType = typeof(ReturnType);if (typeof(PSObject) == resultType || typeof(object) == resultType){ReturnType result;LanguagePrimitives.TryConvertTo(inputObject, out result);return result;}System.Management.Automation.Diagnostics.Assert(false,"ReturnType should be either object or PSObject only");throw PSTraceSource.NewNotSupportedException();}}/// <summary>/// A ObjectReader for a PSDataCollection ObjectStream./// </summary>/// <remarks>/// PSDataCollection is introduced after 1.0. PSDataCollection is/// used to store data which can be used with different/// commands concurrently./// Only Read() operation is supported currently./// </remarks>internal class PSDataCollectionPipelineReader<DataStoreType, ReturnType>: ObjectReaderBase<ReturnType>{#region Private Dataprivate PSDataCollection<DataStoreType> _datastore;#endregion Private Data#region ctor/// <summary>/// Construct with an existing ObjectStream./// </summary>/// <param name="stream">The stream to read.</param>/// <param name="computerName"></param>/// <param name="runspaceId"></param>internal PSDataCollectionPipelineReader(PSDataCollectionStream<DataStoreType> stream,string computerName, Guid runspaceId): base(stream){System.Management.Automation.Diagnostics.Assert(stream.ObjectStore != null,"Stream should have a valid data store");_datastore = stream.ObjectStore;ComputerName = computerName;RunspaceId = runspaceId;}#endregion ctor/// <summary>/// Computer name passed in by the pipeline which/// created this reader./// </summary>internal string ComputerName { get; }/// <summary>/// Runspace Id passed in by the pipeline which/// created this reader./// </summary>internal Guid RunspaceId { get; }/// <summary>/// This method is not supported./// </summary>/// <param name="count">The maximum number of objects to read.</param>/// <returns>The objects read.</returns>public override Collection<ReturnType> Read(int count){throw new NotSupportedException();}/// <summary>/// Read a single object from the stream./// </summary>/// <returns>/// The next object in the buffer or AutomationNull if buffer is closed/// and data is not available./// </returns>/// <remarks>/// This method blocks if the buffer is empty./// </remarks>public override ReturnType Read(){object result = AutomationNull.Value;if (_datastore.Count > 0){Collection<DataStoreType> resultCollection = _datastore.ReadAndRemove(1);// ReadAndRemove returns a Collection<DataStoreType> type but we// just want the single object contained in the collection.if (resultCollection.Count == 1){result = resultCollection[0];}}return ConvertToReturnType(result);}/// <summary>/// This method is not supported./// </summary>/// <returns></returns>/// <remarks></remarks>public override Collection<ReturnType> ReadToEnd(){throw new NotSupportedException();}/// <summary>/// This method is not supported./// </summary>/// <returns></returns>/// <remarks></remarks>public override Collection<ReturnType> NonBlockingRead(){return NonBlockingRead(Int32.MaxValue);}/// <summary>/// This method is not supported./// </summary>/// <returns></returns>/// <remarks></remarks>/// <param name="maxRequested">/// Return no more than maxRequested objects./// </param>public override Collection<ReturnType> NonBlockingRead(int maxRequested){if (maxRequested < 0){throw PSTraceSource.NewArgumentOutOfRangeException("maxRequested", maxRequested);}if (maxRequested == 0){return new Collection<ReturnType>();}Collection<ReturnType> results = new Collection<ReturnType>();int readCount = maxRequested;while (readCount > 0){if (_datastore.Count > 0){results.Add(ConvertToReturnType((_datastore.ReadAndRemove(1))[0]));readCount--;continue;}break;}return results;}/// <summary>/// This method is not supported./// </summary>/// <returns></returns>public override ReturnType Peek(){throw new NotSupportedException();}/// <summary>/// Converts to the return type based on language primitives./// </summary>/// <param name="inputObject">Input object to convert.</param>/// <returns>Input object converted to the specified return type.</returns>private ReturnType ConvertToReturnType(object inputObject){Type resultType = typeof(ReturnType);if (typeof(PSObject) == resultType || typeof(object) == resultType){ReturnType result;LanguagePrimitives.TryConvertTo(inputObject, out result);return result;}System.Management.Automation.Diagnostics.Assert(false,"ReturnType should be either object or PSObject only");throw PSTraceSource.NewNotSupportedException();}#region IDisposable/// <summary>/// Release all resources./// </summary>/// <param name="disposing">If true, release all managed resources.</param>protected override void Dispose(bool disposing){if (disposing){_datastore.Dispose();}}#endregion IDisposable}}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。