Code:
/ 4.0 / 4.0 / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / Services / Monitoring / system / Diagnosticts / AsyncStreamReader.cs / 1305376 / AsyncStreamReader.cs
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
/*============================================================
**
** Class: AsyncStreamReader
**
** Purpose: For reading text from streams using a particular
** encoding in an asychronous manner used by the process class
**
**
===========================================================*/
namespace System.Diagnostics {
using System;
using System.IO;
using System.Text;
using System.Runtime.InteropServices;
using System.Threading;
using System.Collections;
internal delegate void UserCallBack(String data);
internal class AsyncStreamReader : IDisposable
{
internal const int DefaultBufferSize = 1024; // Byte buffer size
private const int MinBufferSize = 128;
private Stream stream;
private Encoding encoding;
private Decoder decoder;
private byte[] byteBuffer;
private char[] charBuffer;
// Record the number of valid bytes in the byteBuffer, for a few checks.
// This is the maximum number of chars we can get from one call to
// ReadBuffer. Used so ReadBuffer can tell when to copy data into
// a user's char[] directly, instead of our internal char[].
private int _maxCharsPerBuffer;
// Store a backpointer to the process class, to check for user callbacks
private Process process;
// Delegate to call user function.
private UserCallBack userCallBack;
// Internal Cancel operation
private bool cancelOperation;
private ManualResetEvent eofEvent;
private Queue messageQueue;
private StringBuilder sb;
private bool bLastCarriageReturn;
internal AsyncStreamReader(Process process, Stream stream, UserCallBack callback, Encoding encoding)
: this(process, stream, callback, encoding, DefaultBufferSize) {
}
// Creates a new AsyncStreamReader for the given stream. The
// character encoding is set by encoding and the buffer size,
// in number of 16-bit characters, is set by bufferSize.
//
internal AsyncStreamReader(Process process, Stream stream, UserCallBack callback, Encoding encoding, int bufferSize)
{
Debug.Assert (process != null && stream !=null && encoding !=null && callback != null, "Invalid arguments!");
Debug.Assert(stream.CanRead, "Stream must be readable!");
Debug.Assert(bufferSize > 0, "Invalid buffer size!");
Init(process, stream, callback, encoding, bufferSize);
messageQueue = new Queue();
}
private void Init(Process process, Stream stream, UserCallBack callback, Encoding encoding, int bufferSize) {
this.process = process;
this.stream = stream;
this.encoding = encoding;
this.userCallBack = callback;
decoder = encoding.GetDecoder();
if (bufferSize < MinBufferSize) bufferSize = MinBufferSize;
byteBuffer = new byte[bufferSize];
_maxCharsPerBuffer = encoding.GetMaxCharCount(bufferSize);
charBuffer = new char[_maxCharsPerBuffer];
cancelOperation = false;
eofEvent = new ManualResetEvent(false);
sb = null;
this.bLastCarriageReturn = false;
}
public virtual void Close()
{
Dispose(true);
}
void IDisposable.Dispose() {
Dispose(true);
GC.SuppressFinalize(this);
}
protected virtual void Dispose(bool disposing)
{
if (disposing) {
if (stream != null)
stream.Close();
}
if (stream != null) {
stream = null;
encoding = null;
decoder = null;
byteBuffer = null;
charBuffer = null;
}
if( eofEvent != null) {
eofEvent.Close();
eofEvent = null;
}
}
public virtual Encoding CurrentEncoding {
get { return encoding; }
}
public virtual Stream BaseStream {
get { return stream; }
}
// User calls BeginRead to start the asynchronous read
internal void BeginReadLine() {
if( cancelOperation) {
cancelOperation = false;
}
if( sb == null ) {
sb = new StringBuilder(DefaultBufferSize);
stream.BeginRead(byteBuffer, 0 , byteBuffer.Length, new AsyncCallback(ReadBuffer), null);
}
else {
FlushMessageQueue();
}
}
internal void CancelOperation() {
cancelOperation = true;
}
// This is the async callback function. Only one thread could/should call this.
private void ReadBuffer(IAsyncResult ar) {
int byteLen;
try {
byteLen = stream.EndRead(ar);
}
catch (IOException ) {
// We should ideally consume errors from operations getting cancelled
// so that we don't crash the unsuspecting parent with an unhandled exc.
// This seems to come in 2 forms of exceptions (depending on platform and scenario),
// namely OperationCanceledException and IOException (for errorcode that we don't
// map explicitly).
byteLen = 0; // Treat this as EOF
}
catch (OperationCanceledException ) {
// We should consume any OperationCanceledException from child read here
// so that we don't crash the parent with an unhandled exc
byteLen = 0; // Treat this as EOF
}
if (byteLen == 0) {
// We're at EOF, we won't call this function again from here on.
lock(messageQueue) {
if( sb.Length != 0) {
messageQueue.Enqueue(sb.ToString());
sb.Length = 0;
}
messageQueue.Enqueue(null);
}
try {
// UserCallback could throw, we should still set the eofEvent
FlushMessageQueue();
}
finally {
eofEvent.Set();
}
} else {
int charLen = decoder.GetChars(byteBuffer, 0, byteLen, charBuffer, 0);
sb.Append(charBuffer, 0, charLen);
GetLinesFromStringBuilder();
stream.BeginRead(byteBuffer, 0 , byteBuffer.Length, new AsyncCallback(ReadBuffer), null);
}
}
// Read lines stored in StringBuilder and the buffer we just read into.
// A line is defined as a sequence of characters followed by
// a carriage return ('\r'), a line feed ('\n'), or a carriage return
// immediately followed by a line feed. The resulting string does not
// contain the terminating carriage return and/or line feed. The returned
// value is null if the end of the input stream has been reached.
//
private void GetLinesFromStringBuilder() {
int i = 0;
int lineStart = 0;
int len = sb.Length;
// skip a beginning '\n' character of new block if last block ended
// with '\r'
if (bLastCarriageReturn && (len > 0) && sb[0] == '\n')
{
i = 1;
lineStart = 1;
bLastCarriageReturn = false;
}
while(i < len) {
char ch = sb[i];
// Note the following common line feed chars:
// \n - UNIX \r\n - DOS \r - Mac
if (ch == '\r' || ch == '\n') {
string s = sb.ToString(lineStart, i - lineStart);
lineStart = i + 1;
// skip the "\n" character following "\r" character
if ((ch == '\r') && (lineStart < len) && (sb[lineStart] == '\n'))
{
lineStart++;
i++;
}
lock(messageQueue) {
messageQueue.Enqueue(s);
}
}
i++;
}
if (sb[len - 1] == '\r') {
bLastCarriageReturn = true;
}
// Keep the rest characaters which can't form a new line in string builder.
if( lineStart < len) {
sb.Remove(0, lineStart);
}
else {
sb.Length = 0;
}
FlushMessageQueue();
}
private void FlushMessageQueue() {
while(true) {
// When we call BeginReadLine, we also need to flush the queue
// So there could be a ---- between the ReadBuffer and BeginReadLine
// We need to take lock before DeQueue.
if( messageQueue.Count > 0) {
lock(messageQueue) {
if( messageQueue.Count > 0) {
string s = (string)messageQueue.Dequeue();
// skip if the read is the read is cancelled
// this might happen inside UserCallBack
// However, continue to drain the queue
if (!cancelOperation)
{
userCallBack(s);
}
}
}
}
else {
break;
}
}
}
// Wait until we hit EOF. This is called from Process.WaitForExit
// We will lose some information if we don't do this.
internal void WaitUtilEOF() {
if( eofEvent != null) {
eofEvent.WaitOne();
eofEvent.Close();
eofEvent = null;
}
}
}
}
// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
/*============================================================
**
** Class: AsyncStreamReader
**
** Purpose: For reading text from streams using a particular
** encoding in an asychronous manner used by the process class
**
**
===========================================================*/
namespace System.Diagnostics {
using System;
using System.IO;
using System.Text;
using System.Runtime.InteropServices;
using System.Threading;
using System.Collections;
internal delegate void UserCallBack(String data);
internal class AsyncStreamReader : IDisposable
{
internal const int DefaultBufferSize = 1024; // Byte buffer size
private const int MinBufferSize = 128;
private Stream stream;
private Encoding encoding;
private Decoder decoder;
private byte[] byteBuffer;
private char[] charBuffer;
// Record the number of valid bytes in the byteBuffer, for a few checks.
// This is the maximum number of chars we can get from one call to
// ReadBuffer. Used so ReadBuffer can tell when to copy data into
// a user's char[] directly, instead of our internal char[].
private int _maxCharsPerBuffer;
// Store a backpointer to the process class, to check for user callbacks
private Process process;
// Delegate to call user function.
private UserCallBack userCallBack;
// Internal Cancel operation
private bool cancelOperation;
private ManualResetEvent eofEvent;
private Queue messageQueue;
private StringBuilder sb;
private bool bLastCarriageReturn;
internal AsyncStreamReader(Process process, Stream stream, UserCallBack callback, Encoding encoding)
: this(process, stream, callback, encoding, DefaultBufferSize) {
}
// Creates a new AsyncStreamReader for the given stream. The
// character encoding is set by encoding and the buffer size,
// in number of 16-bit characters, is set by bufferSize.
//
internal AsyncStreamReader(Process process, Stream stream, UserCallBack callback, Encoding encoding, int bufferSize)
{
Debug.Assert (process != null && stream !=null && encoding !=null && callback != null, "Invalid arguments!");
Debug.Assert(stream.CanRead, "Stream must be readable!");
Debug.Assert(bufferSize > 0, "Invalid buffer size!");
Init(process, stream, callback, encoding, bufferSize);
messageQueue = new Queue();
}
private void Init(Process process, Stream stream, UserCallBack callback, Encoding encoding, int bufferSize) {
this.process = process;
this.stream = stream;
this.encoding = encoding;
this.userCallBack = callback;
decoder = encoding.GetDecoder();
if (bufferSize < MinBufferSize) bufferSize = MinBufferSize;
byteBuffer = new byte[bufferSize];
_maxCharsPerBuffer = encoding.GetMaxCharCount(bufferSize);
charBuffer = new char[_maxCharsPerBuffer];
cancelOperation = false;
eofEvent = new ManualResetEvent(false);
sb = null;
this.bLastCarriageReturn = false;
}
public virtual void Close()
{
Dispose(true);
}
void IDisposable.Dispose() {
Dispose(true);
GC.SuppressFinalize(this);
}
protected virtual void Dispose(bool disposing)
{
if (disposing) {
if (stream != null)
stream.Close();
}
if (stream != null) {
stream = null;
encoding = null;
decoder = null;
byteBuffer = null;
charBuffer = null;
}
if( eofEvent != null) {
eofEvent.Close();
eofEvent = null;
}
}
public virtual Encoding CurrentEncoding {
get { return encoding; }
}
public virtual Stream BaseStream {
get { return stream; }
}
// User calls BeginRead to start the asynchronous read
internal void BeginReadLine() {
if( cancelOperation) {
cancelOperation = false;
}
if( sb == null ) {
sb = new StringBuilder(DefaultBufferSize);
stream.BeginRead(byteBuffer, 0 , byteBuffer.Length, new AsyncCallback(ReadBuffer), null);
}
else {
FlushMessageQueue();
}
}
internal void CancelOperation() {
cancelOperation = true;
}
// This is the async callback function. Only one thread could/should call this.
private void ReadBuffer(IAsyncResult ar) {
int byteLen;
try {
byteLen = stream.EndRead(ar);
}
catch (IOException ) {
// We should ideally consume errors from operations getting cancelled
// so that we don't crash the unsuspecting parent with an unhandled exc.
// This seems to come in 2 forms of exceptions (depending on platform and scenario),
// namely OperationCanceledException and IOException (for errorcode that we don't
// map explicitly).
byteLen = 0; // Treat this as EOF
}
catch (OperationCanceledException ) {
// We should consume any OperationCanceledException from child read here
// so that we don't crash the parent with an unhandled exc
byteLen = 0; // Treat this as EOF
}
if (byteLen == 0) {
// We're at EOF, we won't call this function again from here on.
lock(messageQueue) {
if( sb.Length != 0) {
messageQueue.Enqueue(sb.ToString());
sb.Length = 0;
}
messageQueue.Enqueue(null);
}
try {
// UserCallback could throw, we should still set the eofEvent
FlushMessageQueue();
}
finally {
eofEvent.Set();
}
} else {
int charLen = decoder.GetChars(byteBuffer, 0, byteLen, charBuffer, 0);
sb.Append(charBuffer, 0, charLen);
GetLinesFromStringBuilder();
stream.BeginRead(byteBuffer, 0 , byteBuffer.Length, new AsyncCallback(ReadBuffer), null);
}
}
// Read lines stored in StringBuilder and the buffer we just read into.
// A line is defined as a sequence of characters followed by
// a carriage return ('\r'), a line feed ('\n'), or a carriage return
// immediately followed by a line feed. The resulting string does not
// contain the terminating carriage return and/or line feed. The returned
// value is null if the end of the input stream has been reached.
//
private void GetLinesFromStringBuilder() {
int i = 0;
int lineStart = 0;
int len = sb.Length;
// skip a beginning '\n' character of new block if last block ended
// with '\r'
if (bLastCarriageReturn && (len > 0) && sb[0] == '\n')
{
i = 1;
lineStart = 1;
bLastCarriageReturn = false;
}
while(i < len) {
char ch = sb[i];
// Note the following common line feed chars:
// \n - UNIX \r\n - DOS \r - Mac
if (ch == '\r' || ch == '\n') {
string s = sb.ToString(lineStart, i - lineStart);
lineStart = i + 1;
// skip the "\n" character following "\r" character
if ((ch == '\r') && (lineStart < len) && (sb[lineStart] == '\n'))
{
lineStart++;
i++;
}
lock(messageQueue) {
messageQueue.Enqueue(s);
}
}
i++;
}
if (sb[len - 1] == '\r') {
bLastCarriageReturn = true;
}
// Keep the rest characaters which can't form a new line in string builder.
if( lineStart < len) {
sb.Remove(0, lineStart);
}
else {
sb.Length = 0;
}
FlushMessageQueue();
}
private void FlushMessageQueue() {
while(true) {
// When we call BeginReadLine, we also need to flush the queue
// So there could be a ---- between the ReadBuffer and BeginReadLine
// We need to take lock before DeQueue.
if( messageQueue.Count > 0) {
lock(messageQueue) {
if( messageQueue.Count > 0) {
string s = (string)messageQueue.Dequeue();
// skip if the read is the read is cancelled
// this might happen inside UserCallBack
// However, continue to drain the queue
if (!cancelOperation)
{
userCallBack(s);
}
}
}
}
else {
break;
}
}
}
// Wait until we hit EOF. This is called from Process.WaitForExit
// We will lose some information if we don't do this.
internal void WaitUtilEOF() {
if( eofEvent != null) {
eofEvent.WaitOne();
eofEvent.Close();
eofEvent = null;
}
}
}
}
// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
Link Menu
This book is available now!
Buy at Amazon US or
Buy at Amazon UK
- Schema.cs
- RC2.cs
- ThreadAbortException.cs
- DocumentReference.cs
- GuidTagList.cs
- ThaiBuddhistCalendar.cs
- HtmlGenericControl.cs
- SerialReceived.cs
- HasActivatableWorkflowEvent.cs
- UnhandledExceptionEventArgs.cs
- Events.cs
- ServiceHttpModule.cs
- BamlRecords.cs
- HtmlInputButton.cs
- AutomationAttributeInfo.cs
- webproxy.cs
- BooleanExpr.cs
- ToolStripDropDownClosingEventArgs.cs
- ConfigDefinitionUpdates.cs
- SessionChannels.cs
- Enlistment.cs
- ClientOptions.cs
- DependencyPropertyKey.cs
- DesignTimeVisibleAttribute.cs
- ErrorEventArgs.cs
- FieldToken.cs
- LinkLabelLinkClickedEvent.cs
- Guid.cs
- ZipIOZip64EndOfCentralDirectoryBlock.cs
- ObjectDataSourceEventArgs.cs
- XmlDocumentType.cs
- ViewBase.cs
- PrintPreviewControl.cs
- FontUnitConverter.cs
- AVElementHelper.cs
- XmlCustomFormatter.cs
- HtmlImage.cs
- BCLDebug.cs
- GlyphRunDrawing.cs
- TraceLog.cs
- CancellationHandler.cs
- SqlDataSourceStatusEventArgs.cs
- MenuItemStyleCollection.cs
- DataGridItemEventArgs.cs
- TextBoxRenderer.cs
- LinqToSqlWrapper.cs
- RefExpr.cs
- BufferedWebEventProvider.cs
- FreeFormPanel.cs
- DesignerVerbCollection.cs
- DataContractAttribute.cs
- ContractReference.cs
- CqlQuery.cs
- VisualCollection.cs
- PartialCachingControl.cs
- DiscoveryReference.cs
- DataGridViewBand.cs
- PolyLineSegmentFigureLogic.cs
- EventMap.cs
- DataKeyArray.cs
- MetadataCacheItem.cs
- CellConstantDomain.cs
- ISAPIRuntime.cs
- WindowManager.cs
- ProbeMatchesCD1.cs
- ListItemConverter.cs
- InvalidEnumArgumentException.cs
- Visitor.cs
- Int64KeyFrameCollection.cs
- SourceSwitch.cs
- StreamHelper.cs
- EntitySetBaseCollection.cs
- OutputCacheSettings.cs
- TransformedBitmap.cs
- NetStream.cs
- WorkflowViewElement.cs
- ReflectionServiceProvider.cs
- HtmlString.cs
- KnownBoxes.cs
- SqlRetyper.cs
- RequestResizeEvent.cs
- XslVisitor.cs
- AddressingProperty.cs
- FrameDimension.cs
- _ChunkParse.cs
- COAUTHIDENTITY.cs
- HostedHttpTransportManager.cs
- IntSecurity.cs
- Restrictions.cs
- UInt16.cs
- SchemaSetCompiler.cs
- MethodBuilder.cs
- DbConnectionPoolIdentity.cs
- SyndicationSerializer.cs
- SnapshotChangeTrackingStrategy.cs
- FlowDocumentPaginator.cs
- PageContentCollection.cs
- DataBoundControlHelper.cs
- SlipBehavior.cs
- FacetDescription.cs