BatchStream.cs source code in C# .NET

Source code for the .NET framework in C#

                        

Code:

/ 4.0 / 4.0 / untmp / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / DataWeb / Client / System / Data / Services / Client / BatchStream.cs / 1305376 / BatchStream.cs

                            //---------------------------------------------------------------------- 
// 
//      Copyright (c) Microsoft Corporation.  All rights reserved.
// 
//  
// break a batch stream into its multiple parts
// text reading parts grabbed from System.IO.StreamReader 
//  
//---------------------------------------------------------------------
 
#if ASTORIA_CLIENT
namespace System.Data.Services.Client
#else
namespace System.Data.Services 
#endif
{ 
    using System; 
    using System.Collections.Generic;
    using System.Diagnostics; 
    using System.Globalization;
    using System.IO;
    using System.Text;
 
#if ASTORIA_CLIENT
#if !ASTORIA_LIGHT // Data.Services http stack 
    using System.Net; 
#else
    using System.Data.Services.Http; 
#endif
#endif

    ///  
    /// materialize objects from an application/atom+xml stream
    ///  
    internal class BatchStream : Stream 
    {
        /// Default buffer size, should be larger than buffer size of StreamReader 
        private const int DefaultBufferSize = 8000;

        /// Is this a batch resquest or batch response
        private readonly bool batchRequest; 

        /// Buffered bytes from the stream. 
        private readonly byte[] byteBuffer; 

        /// Underlying stream being buffered. 
        private Stream reader;

        /// Number of valid bytes in the byteBuffer.
        private int byteLength; 

        /// Position in the byteBuffer. 
        private int bytePosition; 

        /// Discovered encoding of underlying stream. 
        private Encoding batchEncoding;

        /// check preamble.
        private bool checkPreamble; 

        /// batch boundary. 
        private string batchBoundary; 

        /// batch length 
        private int batchLength;

        /// running total byte count
        private int totalCount; 

        /// changeset boundary. 
        private string changesetBoundary; 

        /// Discovered encoding of underlying neseted stream. 
        private Encoding changesetEncoding;

        /// content headers
        private Dictionary contentHeaders; 

        /// content stream 
        private Stream contentStream; 

        /// stream dispose delayed until the contentStream is disposed 
        private bool disposeWithContentStreamDispose;

#if ASTORIA_SERVER
        /// content uri 
        private string contentUri;
#else 
        /// status code of the response. 
        private string statusCode;
#endif 

        /// batch state
        private BatchStreamState batchState;
 
#if DEBUG && !ASTORIA_LIGHT
        /// everything batch reads to help debugging 
        private MemoryStream writer = new MemoryStream(); 
#else
#pragma warning disable 649 
        /// everything batch reads to help debugging
        private MemoryStream writer;
#pragma warning restore 649
#endif 

        /// Wrap a stream for batching. 
        /// underlying stream 
        /// batch boundary
        /// encoding of batch 
        /// is request stream or response stream
        internal BatchStream(Stream stream, string boundary, Encoding batchEncoding, bool requestStream)
        {
            Debug.Assert(null != stream, "null stream"); 

            this.reader = stream; 
            this.byteBuffer = new byte[DefaultBufferSize]; 
            this.batchBoundary = VerifyBoundary(boundary);
            this.batchState = BatchStreamState.StartBatch; 
            this.batchEncoding = batchEncoding;
            this.checkPreamble = (null != batchEncoding);
            this.batchRequest = requestStream;
        } 

        #region batch properties ContentHeaders, ContentStream, Encoding, Sate 
        /// content headers 
        public Dictionary ContentHeaders
        { 
            get { return this.contentHeaders; }
        }

#if ASTORIA_SERVER 
        /// Content URI.
        public string ContentUri 
        { 
            get { return this.contentUri; }
        } 
#endif

        /// encoding
        public Encoding Encoding 
        {
            get { return this.changesetEncoding ?? this.batchEncoding; } 
        } 

        /// batch state 
        public BatchStreamState State
        {
            get { return this.batchState; }
        } 
        #endregion
 
        #region Stream properties 
        /// Delegate to underlying stream
        public override bool CanRead 
        {
            get { return (null != this.reader && this.reader.CanRead); }
        }
 
        /// False
        public override bool CanSeek 
        { 
            get { return false; }
        } 

        /// False
        public override bool CanWrite
        { 
            get { return false; }
        } 
 
        /// Not supported.
        public override long Length 
        {
            get { throw Error.NotSupported(); }
        }
 
        /// Not supported.
        public override long Position 
        { 
            get { throw Error.NotSupported(); }
            set { throw Error.NotSupported(); } 
        }
        #endregion

        #region Stream methods 
        /// Does nothing.
        public override void Flush() 
        { 
            this.reader.Flush();
        } 

        /// Not supported.
        /// The parameter is not used.
        /// The parameter is not used. 
        /// The parameter is not used.
        /// The parameter is not used. 
        /// The parameter is not used. 
        /// nothing
        public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state) 
        {
            throw Error.NotSupported();
        }
 
        /// Not supported.
        /// The parameter is not used. 
        /// The parameter is not used. 
        /// The parameter is not used.
        /// nothing 
        public override int Read(byte[] buffer, int offset, int count)
        {
            throw Error.NotSupported();
        } 

        ///  
        /// Forward seek in buffered underlying stream. 
        /// 
        /// non-negative bytes to forward seek 
        /// must be Current
        /// underlying stream forward seek result.
        public override long Seek(long offset, SeekOrigin origin)
        { 
            this.AssertOpen();
 
            if (offset < 0) 
            {
                throw Error.ArgumentOutOfRange("offset"); 
            }

            if (SeekOrigin.Current != origin)
            { 
                throw Error.ArgumentOutOfRange("origin");
            } 
 
            if (Int32.MaxValue == offset)
            {   // special case - read to end of delimiter 
                byte[] buffer = new byte[256]; // should be at least 70 for minimum boundary length
                while (0 < this.ReadDelimiter(buffer, 0, buffer.Length))
                {
                    /* ignore data */ 
                }
            } 
            else if (0 < offset) 
            {   // underlying stream may not support seek, so just move forward the buffered bytes
                do 
                {
                    int count = Math.Min(checked((int)offset), Math.Min(this.byteLength, this.batchLength));
                    this.totalCount += count;
                    this.bytePosition += count; 
                    this.byteLength -= count;
                    this.batchLength -= count; 
                    offset -= count; 

                    // underlying stream doesn't support Seek, so we just need to fill our buffer. 
                }
                while ((0 < offset) && (this.batchLength != 0) && this.ReadBuffer());
            }
 
            Debug.Assert(0 <= this.byteLength, "negative byteLength");
            Debug.Assert(0 <= this.batchLength, "negative batchLength"); 
            return 0; 
        }
 
        /// Not supported.
        /// The parameter is not used.
        public override void SetLength(long value)
        { 
            throw Error.NotSupported();
        } 
 
        /// Not supported.
        /// The parameter is not used. 
        /// The parameter is not used.
        /// The parameter is not used.
        public override void Write(byte[] buffer, int offset, int count)
        { 
            throw Error.NotSupported();
        } 
        #endregion 

        ///  
        /// Get the boundary string and encoding if the content type is multipart/mixed.
        /// 
        /// content type specified in the request.
        /// returns the boundary string specified in the content type. 
        /// returns the encoding specified in the content type.
        /// true if multipart/mixed with boundary 
        /// if multipart/mixed without boundary 
        internal static bool GetBoundaryAndEncodingFromMultipartMixedContentType(string contentType, out string boundary, out Encoding encoding)
        { 
            boundary = null;
            encoding = null;

            string mime; 
            KeyValuePair[] parameters = HttpProcessUtility.ReadContentType(contentType, out mime, out encoding);
 
            if (String.Equals(XmlConstants.MimeMultiPartMixed, mime, StringComparison.OrdinalIgnoreCase)) 
            {
                if (null != parameters) 
                {
                    foreach (KeyValuePair parameter in parameters)
                    {
                        if (String.Equals(parameter.Key, XmlConstants.HttpMultipartBoundary, StringComparison.OrdinalIgnoreCase)) 
                        {
                            if (boundary != null) 
                            {   // detect multiple boundary parameters 
                                boundary = null;
                                break; 
                            }

                            boundary = parameter.Value;
                        } 
                    }
                } 
 
                // if an invalid boundary string is specified or no boundary string is specified
                if (String.IsNullOrEmpty(boundary)) 
                {   // however, empty string is considered a valid boundary
                    throw Error.BatchStreamMissingBoundary();
                }
            } 

            return (null != boundary); 
        } 

#if !ASTORIA_CLIENT 
        /// 
        /// Checks whether the given stream instance is a internal batch stream class or not.
        /// 
        /// stream instance. 
        /// returns true if the given stream instance is a internal batch stream class. Otherwise returns false.
        internal static bool IsBatchStream(Stream stream) 
        { 
            return (stream is StreamWithDelimiter || stream is StreamWithLength);
        } 

        /// 
        /// Validates that there is no extreaneous data in the stream beyond the end of batch
        ///  
        /// Exception if there is remaining data after the end boundary of the batch. Otherwise null.
        internal Exception ValidateNoDataBeyondEndOfBatch() 
        { 
            if (this.reader.ReadByte() >= 0)
            { 
                return Error.BatchStreamMoreDataAfterEndOfBatch();
            }

            return null; 
        }
#endif 
 
#if ASTORIA_CLIENT
        /// Gets the version from content-headers if available. 
        /// The value for the DataServiceVersion header.
        internal string GetResponseVersion()
        {
            string result; 
            this.ContentHeaders.TryGetValue(XmlConstants.HttpDataServiceVersion, out result);
            return result; 
        } 

        /// Get and parse status code from content-headers 
        /// status code
        internal HttpStatusCode GetStatusCode()
        {
            return (HttpStatusCode)(null != this.statusCode ? Int32.Parse(this.statusCode, CultureInfo.InvariantCulture) : 500); 
        }
#endif 
 
        /// start a multipart content section with a specific boundary
        /// true if this is content to process 
        /// 
        /// 5.1.2
        /// an improperly truncated "multipart" entity may not have
        /// any terminating boundary marker. 
        /// MIME implementations are required to recognize outer level
        /// boundary markers at ANY level of inner nesting. 
        /// 5.1.3 
        /// The "mixed" subtype of "multipart" is intended for use when the body
        /// parts are independent and need to be bundled in a particular order. 
        /// Any "multipart" subtypes that an implementation does not recognize
        /// must be treated as being of subtype "mixed".
        /// 
        internal bool MoveNext() 
        {
            #region dispose previous content stream 
            if (null == this.reader || this.disposeWithContentStreamDispose) 
            {
                return false; 
            }

            if (null != this.contentStream)
            { 
                this.contentStream.Dispose();
            } 
 
            Debug.Assert(0 <= this.byteLength, "negative byteLength");
            Debug.Assert(0 <= this.batchLength, "negative batchLength"); 
            #endregion

            #region initialize start state to EndBatch or EndChangeSet
            switch (this.batchState) 
            {
                case BatchStreamState.EndBatch: 
                    // already finished 
                    Debug.Assert(null == this.batchBoundary, "non-null batch boundary");
                    Debug.Assert(null == this.changesetBoundary, "non-null changesetBoundary boundary"); 
                    throw Error.BatchStreamInvalidBatchFormat();

                case BatchStreamState.Get:
                case BatchStreamState.GetResponse: 
                    // Since there is no content specified for Get operations,
                    // after the operation is performed, we need to clear out the headers and uri information 
                    // specified for the Get operation 
                    this.ClearPreviousOperationInformation();
                    goto case BatchStreamState.StartBatch; 

                case BatchStreamState.StartBatch:
                case BatchStreamState.EndChangeSet:
                    Debug.Assert(null != this.batchBoundary, "null batch boundary"); 
                    Debug.Assert(null == this.changesetBoundary, "non-null changeset boundary");
                    this.batchState = BatchStreamState.EndBatch; 
                    this.batchLength = Int32.MaxValue; 
                    break;
 
                case BatchStreamState.BeginChangeSet:
                    Debug.Assert(null != this.batchBoundary, "null batch boundary");
                    Debug.Assert(null != this.contentHeaders, "null contentHeaders");
                    Debug.Assert(null != this.changesetBoundary, "null changeset boundary"); 
                    this.contentHeaders = null;
                    this.changesetEncoding = null; 
                    this.batchState = BatchStreamState.EndChangeSet; 
                    break;
 
                case BatchStreamState.ChangeResponse:
                case BatchStreamState.Delete:
                    Debug.Assert(null != this.changesetBoundary, "null changeset boundary");
                    this.ClearPreviousOperationInformation(); 
                    this.batchState = BatchStreamState.EndChangeSet;
                    break; 
 
                case BatchStreamState.Post:
                case BatchStreamState.Put: 
                case BatchStreamState.Merge:
                    // Since there is no content specified for DELETE operations or PUT response
                    // after the operation is performed, we need to clear out the headers and uri information
                    // specified for the DELETE operation 
                    Debug.Assert(null != this.changesetBoundary, "null changeset boundary");
                    this.batchState = BatchStreamState.EndChangeSet; 
                    break; 

                default: 
                    Debug.Assert(false, "unknown state");
                    throw Error.BatchStreamInvalidBatchFormat();
            }
 
            Debug.Assert(null == this.contentHeaders, "non-null content headers");
            Debug.Assert(null == this.contentStream, "non-null content stream"); 
#if ASTORIA_SERVER 
            Debug.Assert(null == this.contentUri, "non-null content uri");
#endif 
#if ASTORIA_CLIENT
            Debug.Assert(null == this.statusCode, "non-null statusCode");
#endif
 
            Debug.Assert(
                this.batchState == BatchStreamState.EndBatch || 
                this.batchState == BatchStreamState.EndChangeSet, 
                "unexpected state at start");
            #endregion 

            #region read --delimiter
            string delimiter = this.ReadLine();
            if (String.IsNullOrEmpty(delimiter)) 
            {   // was the \r\n not included in the previous section's content-length?
                delimiter = this.ReadLine(); 
            } 

            if (String.IsNullOrEmpty(delimiter)) 
            {
                throw Error.BatchStreamInvalidBatchFormat();
            }
 
            if (delimiter.EndsWith("--", StringComparison.Ordinal))
            { 
                delimiter = delimiter.Substring(0, delimiter.Length - 2); 

                if ((null != this.changesetBoundary) && (delimiter == this.changesetBoundary)) 
                {
                    Debug.Assert(this.batchState == BatchStreamState.EndChangeSet, "bad changeset boundary state");

                    this.changesetBoundary = null; 
                    return true;
                } 
                else if (delimiter == this.batchBoundary) 
                {
                    if (BatchStreamState.EndChangeSet == this.batchState) 
                    {   // we should technically recover, but we are not going to.
                        throw Error.BatchStreamMissingEndChangesetDelimiter();
                    }
 
                    this.changesetBoundary = null;
                    this.batchBoundary = null; 
                    if (this.byteLength != 0) 
                    {
                        throw Error.BatchStreamMoreDataAfterEndOfBatch(); 
                    }

                    return false;
                } 
                else
                { 
                    throw Error.BatchStreamInvalidDelimiter(delimiter); 
                }
            } 
            else if ((null != this.changesetBoundary) && (delimiter == this.changesetBoundary))
            {
                Debug.Assert(this.batchState == BatchStreamState.EndChangeSet, "bad changeset boundary state");
            } 
            else if (delimiter == this.batchBoundary)
            { 
                if (this.batchState != BatchStreamState.EndBatch) 
                {
                    if (this.batchState == BatchStreamState.EndChangeSet) 
                    {   // we should technically recover, but we are not going to.
                        throw Error.BatchStreamMissingEndChangesetDelimiter();
                    }
                    else 
                    {
                        throw Error.BatchStreamInvalidBatchFormat(); 
                    } 
                }
            } 
            else
            {   // unknown delimiter
                throw Error.BatchStreamInvalidDelimiter(delimiter);
            } 

            #endregion 
 
            #region read header with values in this form (([^:]*:.*)\r\n)*\r\n
            this.ReadContentHeaders(); 
            #endregion

            #region should start changeset?
            string contentType; 
            bool readHttpHeaders = false;
            if (this.contentHeaders.TryGetValue(XmlConstants.HttpContentType, out contentType)) 
            { 
                if (String.Equals(contentType, XmlConstants.MimeApplicationHttp, StringComparison.OrdinalIgnoreCase))
                { 
                    // We don't allow custom headers at the start of changeset or get batch request.
                    // One can always specify custom headers along with the other http headers that
                    // follows these headers
                    if (this.contentHeaders.Count != 2) 
                    {
                        throw Error.BatchStreamInvalidNumberOfHeadersAtOperationStart( 
                            XmlConstants.HttpContentType, 
                            XmlConstants.HttpContentTransferEncoding);
                    } 

                    string transferEncoding;
                    if (!this.contentHeaders.TryGetValue(XmlConstants.HttpContentTransferEncoding, out transferEncoding) ||
                        XmlConstants.BatchRequestContentTransferEncoding != transferEncoding) 
                    {
                        throw Error.BatchStreamMissingOrInvalidContentEncodingHeader( 
                            XmlConstants.HttpContentTransferEncoding, 
                            XmlConstants.BatchRequestContentTransferEncoding);
                    } 

                    readHttpHeaders = true;
                }
                else if (BatchStreamState.EndBatch == this.batchState) 
                {
                    string boundary; 
                    Encoding encoding; 
                    if (GetBoundaryAndEncodingFromMultipartMixedContentType(contentType, out boundary, out encoding))
                    { 
                        this.changesetBoundary = VerifyBoundary(boundary);
                        this.changesetEncoding = encoding;
                        this.batchState = BatchStreamState.BeginChangeSet;
                    } 
                    else
                    { 
                        throw Error.BatchStreamInvalidContentTypeSpecified( 
                            XmlConstants.HttpContentType,
                            contentType, 
                            XmlConstants.MimeApplicationHttp,
                            XmlConstants.MimeMultiPartMixed);
                    }
 
                    // We don't allow custom headers at the start of batch operation.
                    // One can always specify custom headers along with the other http headers that 
                    // are present in the changeset. 
                    if (this.contentHeaders.Count > 2 ||
                        (this.contentHeaders.Count == 2 && !this.contentHeaders.ContainsKey(XmlConstants.HttpContentLength))) 
                    {
                        throw Error.BatchStreamInvalidNumberOfHeadersAtChangeSetStart(XmlConstants.HttpContentType, XmlConstants.HttpContentLength);
                    }
                } 
                else
                { 
                    throw Error.BatchStreamInvalidContentTypeSpecified( 
                        XmlConstants.HttpContentType,
                        contentType, 
                        XmlConstants.MimeApplicationHttp,
                        XmlConstants.MimeMultiPartMixed);
                }
            } 
            else
            { 
                throw Error.BatchStreamMissingContentTypeHeader(XmlConstants.HttpContentType); 
            }
            #endregion 

            #region what is the operation and uri?
            if (readHttpHeaders)
            { 
                this.ReadHttpHeaders();
 
                // read the content type to clear the value 
                // of the content type
                this.contentHeaders.TryGetValue(XmlConstants.HttpContentType, out contentType); 
            }
            #endregion

            //// stream is now positioned on content 
            //// or its on the start of the actual headers
 
            #region does content have a fixed length? 
            string text = null;
            int length = -1; 
            if (this.contentHeaders.TryGetValue(XmlConstants.HttpContentLength, out text))
            {
                length = Int32.Parse(text, CultureInfo.InvariantCulture);
                if (length < 0) 
                {
                    throw Error.BatchStreamInvalidContentLengthSpecified(text); 
                } 

                if (this.batchState == BatchStreamState.BeginChangeSet) 
                {
                    this.batchLength = length;
                }
                else if (length != 0) 
                {
                    Debug.Assert( 
                        this.batchState == BatchStreamState.Delete || 
                        this.batchState == BatchStreamState.Get ||
                        this.batchState == BatchStreamState.Post || 
                        this.batchState == BatchStreamState.Put ||
                        this.batchState == BatchStreamState.Merge,
                        "unexpected contentlength location");
                    this.contentStream = new StreamWithLength(this, length); 
                }
            } 
            else 
            {
                if (this.batchState == BatchStreamState.EndBatch) 
                {
                    this.batchLength = Int32.MaxValue;
                }
 
                if (this.batchState != BatchStreamState.BeginChangeSet)
                { 
                    this.contentStream = new StreamWithDelimiter(this); 
                }
            } 

            #endregion

            Debug.Assert( 
                this.batchState == BatchStreamState.BeginChangeSet ||
                (this.batchRequest && (this.batchState == BatchStreamState.Delete || 
                                       this.batchState == BatchStreamState.Get || 
                                       this.batchState == BatchStreamState.Post ||
                                       this.batchState == BatchStreamState.Put || 
                                       this.batchState == BatchStreamState.Merge)) ||
                (!this.batchRequest && (this.batchState == BatchStreamState.GetResponse ||
                                        this.batchState == BatchStreamState.ChangeResponse)),
                "unexpected state at return"); 

            #region enforce if contentStream is expected, caller needs to enforce if contentStream is not expected 
            if (null == this.contentStream) 
            {
                switch (this.batchState) 
                {
                    case BatchStreamState.BeginChangeSet:
                    case BatchStreamState.Delete:
                    case BatchStreamState.Get: 
                    case BatchStreamState.ChangeResponse:   // example DELETE /Customers(1)
                    case BatchStreamState.GetResponse:      // example GET /Customers(1)/BestFriend 
                        break; 

                    case BatchStreamState.Post: 
                    case BatchStreamState.Put:
                    case BatchStreamState.Merge:
                    default:
                        // we expect a content stream 
                        throw Error.BatchStreamContentExpected(this.batchState);
                } 
            } 
            #endregion
 
            #region enforce if contentType not is expected, caller needs to enforce if contentType is expected
            if (!String.IsNullOrEmpty(contentType))
            {
                switch (this.batchState) 
                {
                    case BatchStreamState.BeginChangeSet: 
                    case BatchStreamState.Post: 
                    case BatchStreamState.Put:
                    case BatchStreamState.Merge: 
                    case BatchStreamState.GetResponse:
                    case BatchStreamState.ChangeResponse:
                        // we do allow content-type to be defined
                        break; 

                    case BatchStreamState.Get:              // request does not expect content-type 
                    case BatchStreamState.Delete:           // request does not expect content-type 
                    default:
                        // we do NOT expect content-type to be defined 
                        throw Error.BatchStreamContentUnexpected(this.batchState);
                }
            }
            #endregion 

            return true; 
        } 

        /// Method to get content stream instead of property so it can be passed as function 
        /// ContentStream
        internal Stream GetContentStream()
        {
            return this.contentStream; 
        }
 
        /// Dispose underlying stream 
        /// true if active dispose, false if finalizer
        protected override void Dispose(bool disposing) 
        {
            if (disposing)
            {
                if (null != this.contentStream) 
                {   // delay disposing of the reader until content stream is disposed
                    this.disposeWithContentStreamDispose = true; 
                } 
                else
                { 
                    this.byteLength = 0;
                    if (null != this.reader)
                    {
                        this.reader.Dispose(); 
                        this.reader = null;
                    } 
 
                    this.contentHeaders = null;
                    if (null != this.contentStream) 
                    {
                        this.contentStream.Dispose();
                    }
 
                    if (null != this.writer)
                    { 
                        this.writer.Dispose(); 
                    }
                } 
            }

            // Stream.Dispose(bool) does nothing, but for completeness w/ FxCop
            base.Dispose(disposing); 
        }
 
        ///  
        /// Validates the method name and returns the state based on the method name
        ///  
        /// method name to be validated
        /// state based on the method name
        private static BatchStreamState GetStateBasedOnHttpMethodName(string methodName)
        { 
            if (XmlConstants.HttpMethodGet.Equals(methodName, StringComparison.Ordinal))
            { 
                return BatchStreamState.Get; 
            }
            else if (XmlConstants.HttpMethodDelete.Equals(methodName, StringComparison.Ordinal)) 
            {
                return BatchStreamState.Delete;
            }
            else if (XmlConstants.HttpMethodPost.Equals(methodName, StringComparison.Ordinal)) 
            {
                return BatchStreamState.Post; 
            } 
            else if (XmlConstants.HttpMethodPut.Equals(methodName, StringComparison.Ordinal))
            { 
                return BatchStreamState.Put;
            }
            else if (XmlConstants.HttpMethodMerge.Equals(methodName, StringComparison.Ordinal))
            { 
                return BatchStreamState.Merge;
            } 
            else 
            {
                throw Error.BatchStreamInvalidHttpMethodName(methodName); 
            }
        }

        ///  
        /// verify boundary delimiter if valid
        ///  
        /// boundary to test 
        /// "--" + boundary
        private static string VerifyBoundary(string boundary) 
        {
            if ((null == boundary) || (70 < boundary.Length))
            {
                throw Error.BatchStreamInvalidDelimiter(boundary); 
            }
 
            foreach (char c in boundary) 
            {
                if ((127 < (int)c) || Char.IsWhiteSpace(c) || Char.IsControl(c)) 
                {   // must be 7-bit, non-whitespace (including newline char), non-control character
                    throw Error.BatchStreamInvalidDelimiter(boundary);
                }
            } 

            return "--" + boundary; 
        } 

        ///  
        /// Clears the headers, contentUri and stream of the previous operation
        /// 
        private void ClearPreviousOperationInformation()
        { 
            this.contentHeaders = null;
            this.contentStream = null; 
#if ASTORIA_SERVER 
            this.contentUri = null;
#endif 
#if ASTORIA_CLIENT
            this.statusCode = null;
#endif
        } 

        /// appends bytes from byteBuffer to buffer 
        /// buffer to append to, grows as necessary 
        /// count of bytes to append
        private void Append(ref byte[] buffer, int count) 
        {
            int oldSize = (null != buffer) ? buffer.Length : 0;

            byte[] tmp = new byte[oldSize + count]; 
            if (0 < oldSize)
            { 
                Buffer.BlockCopy(buffer, 0, tmp, 0, oldSize); 
            }
 
            Buffer.BlockCopy(this.byteBuffer, this.bytePosition, tmp, oldSize, count);
            buffer = tmp;

            this.totalCount += count; 
            this.bytePosition += count;
            this.byteLength -= count; 
            this.batchLength -= count; 

            Debug.Assert(0 <= this.byteLength, "negative byteLength"); 
            Debug.Assert(0 <= this.batchLength, "negative batchLength");
        }

        /// verify reader is open 
        /// if reader is used after dispose
        private void AssertOpen() 
        { 
            if (null == this.reader)
            { 
                Error.ThrowObjectDisposed(this.GetType());
            }
        }
 
        /// Fill the buffer from the underlying stream.
        /// true if any data was read. 
        private bool ReadBuffer() 
        {
            this.AssertOpen(); 

            if (0 == this.byteLength)
            {
                this.bytePosition = 0; 
                this.byteLength = this.reader.Read(this.byteBuffer, this.bytePosition, this.byteBuffer.Length);
                if (null != this.writer) 
                { 
                    this.writer.Write(this.byteBuffer, this.bytePosition, this.byteLength);
                } 

                if (null == this.batchEncoding)
                {
                    this.batchEncoding = this.DetectEncoding(); 
                }
                else if (null != this.changesetEncoding) 
                { 
                    this.changesetEncoding = this.DetectEncoding();
                } 
                else if (this.checkPreamble)
                {
                    bool match = true;
                    byte[] preamble = this.batchEncoding.GetPreamble(); 
                    if (preamble.Length <= this.byteLength)
                    { 
                        for (int i = 0; i < preamble.Length; ++i) 
                        {
                            if (preamble[i] != this.byteBuffer[i]) 
                            {
                                match = false;
                                break;
                            } 
                        }
 
                        if (match) 
                        {
                            this.byteLength -= preamble.Length; 
                            this.bytePosition += preamble.Length;
                        }
                    }
 
                    this.checkPreamble = false;
                } 
 
                return (0 < this.byteLength);
            } 

            return true;
        }
 
        /// 
        /// Reads a line. 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.
        /// 
        /// line from the buffered stream
        private String ReadLine() 
        {
            if ((0 == this.batchLength) || !this.ReadBuffer()) 
            { 
                return null;
            } 

            byte[] buffer = null;
            do
            { 
                Debug.Assert(0 < this.byteLength, "out of bytes");
                Debug.Assert(this.bytePosition + this.byteLength <= this.byteBuffer.Length, "byte tracking out of range"); 
                int i = this.bytePosition; 
                int end = i + Math.Min(this.byteLength, this.batchLength);
                do 
                {
                    char ch = (char)this.byteBuffer[i];

                    // Note the following common line feed chars: 
                    // \n - UNIX   \r\n - DOS   \r - Mac
                    if (('\r' == ch) || ('\n' == ch)) 
                    { 
                        string s;
 
                        i -= this.bytePosition;
                        if (null != buffer)
                        {
                            this.Append(ref buffer, i); 
                            s = this.Encoding.GetString(buffer, 0, buffer.Length);
                        } 
                        else 
                        {
                            s = this.Encoding.GetString(this.byteBuffer, this.bytePosition, i); 

                            this.totalCount += i;
                            this.bytePosition += i;
                            this.byteLength -= i; 
                            this.batchLength -= i;
                        } 
 
                        this.totalCount++;
                        this.bytePosition++; 
                        this.byteLength--;
                        this.batchLength--;
                        if (('\r' == ch) && ((0 < this.byteLength) || this.ReadBuffer()) && (0 < this.batchLength))
                        { 
                            ch = (char)this.byteBuffer[this.bytePosition];
                            if ('\n' == ch) 
                            { 
                                this.totalCount++;
                                this.bytePosition++; 
                                this.byteLength--;
                                this.batchLength--;
                            }
                        } 

                        Debug.Assert(0 <= this.byteLength, "negative byteLength"); 
                        Debug.Assert(0 <= this.batchLength, "negative batchLength"); 
                        return s;
                    } 

                    i++;
                }
                while (i < end); 

                i -= this.bytePosition; 
                this.Append(ref buffer, i); 
            }
            while (this.ReadBuffer() && (0 < this.batchLength)); 

            Debug.Assert(0 <= this.byteLength, "negative byteLength");
            Debug.Assert(0 <= this.batchLength, "negative batchLength");
            return this.Encoding.GetString(buffer, 0, buffer.Length); 
        }
 
        /// Detect the encoding based data from the stream. 
        /// discovered encoding
        private Encoding DetectEncoding() 
        {
            if (this.byteLength < 2)
            {
#if !ASTORIA_LIGHT  // ASCII not available 
                return Encoding.ASCII;
#else 
                return HttpProcessUtility.FallbackEncoding; 
#endif
            } 
            else if (this.byteBuffer[0] == 0xFE && this.byteBuffer[1] == 0xFF)
            {   // Big Endian Unicode
                this.bytePosition = 2;
                this.byteLength -= 2; 
                return new UnicodeEncoding(true, true);
            } 
            else if (this.byteBuffer[0] == 0xFF && this.byteBuffer[1] == 0xFE) 
            {   // Little Endian Unicode, or possibly little endian UTF32
                if (this.byteLength >= 4 && 
                    this.byteBuffer[2] == 0 &&
                    this.byteBuffer[3] == 0)
                {
#if !ASTORIA_LIGHT  // Little Endian UTF32 not available 
                    this.bytePosition = 4;
                    this.byteLength -= 4; 
                    return new UTF32Encoding(false, true); 
#else
                    throw Error.NotSupported(); 
#endif
                }
                else
                { 
                    this.bytePosition = 2;
                    this.byteLength -= 2; 
                    return new UnicodeEncoding(false, true); 
                }
            } 
            else if (this.byteLength >= 3 &&
                     this.byteBuffer[0] == 0xEF &&
                     this.byteBuffer[1] == 0xBB &&
                     this.byteBuffer[2] == 0xBF) 
            {   // UTF-8
                this.bytePosition = 3; 
                this.byteLength -= 3; 
                return Encoding.UTF8;
            } 
            else if (this.byteLength >= 4 &&
                     this.byteBuffer[0] == 0 &&
                     this.byteBuffer[1] == 0 &&
                     this.byteBuffer[2] == 0xFE && 
                     this.byteBuffer[3] == 0xFF)
            {   // Big Endian UTF32 
#if !ASTORIA_LIGHT  // Big Endian UTF32 not available 
                this.bytePosition = 4;
                this.byteLength -= 4; 
                return new UTF32Encoding(true, true);
#else
                throw Error.NotSupported();
#endif 
            }
            else 
            { 
#if !ASTORIA_LIGHT  // ASCII not available
                return Encoding.ASCII; 
#else
                return HttpProcessUtility.FallbackEncoding;
#endif
            } 
        }
 
        ///  
        /// read from BatchStream buffer into user buffer, stopping when a boundary delimiter is found
        ///  
        /// place to copy bytes read from underlying stream
        /// offset in buffer to start writing
        /// count of bytes to read from buffered underlying stream
        /// count of bytes actualy copied into buffer 
        private int ReadDelimiter(byte[] buffer, int offset, int count)
        { 
            Debug.Assert(null != buffer, "null != buffer"); 
            Debug.Assert(0 <= offset, "0 <= offset");
            Debug.Assert(0 <= count, "0 <= count"); 
            Debug.Assert(offset + count <= buffer.Length, "offset + count <= buffer.Length");
            int copied = 0;

            // which boundary are we looking for 
            string boundary = null;
            string boundary1 = this.batchBoundary; 
            string boundary2 = this.changesetBoundary; 

            while ((0 < count) && (0 < this.batchLength) && this.ReadBuffer()) 
            {
                // if a boundary spanned to actually buffer reads, we shifted and restart boundary match
                // how many bytes have we matched in the boundary
                int boundaryIndex = 0; 
                int boundary1Index = 0;
                int boundary2Index = 0; 
 
                // how many bytes can we search for
                int size = Math.Min(Math.Min(count, this.byteLength), this.batchLength) + this.bytePosition; 

                byte[] data = this.byteBuffer;
                for (int i = this.bytePosition; i < size; ++i)
                { 
                    byte value = data[i];
                    buffer[offset++] = value; // copy value to caller's buffer 
 
                    if ((char)value == boundary1[boundary1Index])
                    { 
                        if (boundary1.Length == ++boundary1Index)
                        {   // found full match
                            size = (1 + i) - boundary1Index;
                            offset -= boundary1Index; 
                            Debug.Assert(this.bytePosition <= size, "negative size");
                            break; 
                        } 
                    }
                    else 
                    {
                        boundary1Index = 0;
                    }
 
                    if ((null != boundary2) && ((char)value == boundary2[boundary2Index]))
                    { 
                        if (boundary2.Length == ++boundary2Index) 
                        {   // found full match
                            size = (1 + i) - boundary2Index; 
                            offset -= boundary2Index;
                            Debug.Assert(this.bytePosition <= size, "negative size");
                            break;
                        } 
                    }
                    else 
                    { 
                        boundary2Index = 0;
                    } 
                }

                // Here the size is going to be the amount of data just read before hitting the boundary
                // Basically it indicates the payload size for the operation. 
                size -= this.bytePosition;
                Debug.Assert(0 <= size, "negative size"); 
 
                if (boundary1Index < boundary2Index)
                { 
                    boundaryIndex = boundary2Index;
                    boundary = boundary2;
                }
                else 
                {
                    Debug.Assert(null != boundary1, "batch boundary shouldn't be null"); 
                    boundaryIndex = boundary1Index; 
                    boundary = boundary1;
                } 

                if (size == this.batchLength)
                {   // outer batch stream has reached its limit - there will be no more data for this delimiter
                    // partial match at EOF is not a match 
                    boundaryIndex = 0;
                } 
 
                // boundaryIndex either represents either
                // full match 
                // partial match and we just need more data in the buffer to continue
                // partial match in the requested count buffer (count maybe < boundary.Length)
                if ((0 < boundaryIndex) && (boundary.Length != boundaryIndex))
                {   // partial boundary in stream - but hit the end (compress and continue) 
                    if ((size + copied == boundaryIndex) && (boundaryIndex < this.byteLength))
                    { 
                        // The issue here is that someone is asking us to read a chunk of data which is 
                        // smaller than the boundary length, and we cannot return partial boundary content
                        // requested smaller amount than we buffered - have partial match - is it real? 
                        // the count caller is requesting is too small without look ahead
                        throw Error.BatchStreamInternalBufferRequestTooSmall();
                    }
                    else 
                    {   // we need more data before we can determine if match
                        size -= boundaryIndex; 
                        offset -= boundaryIndex; 
                    }
                } 

                this.totalCount += size;
                this.bytePosition += size;
                this.byteLength -= size; 
                this.batchLength -= size;
 
                count -= size; 
                copied += size;
 
                // Each boundary should begin in a new line.  We should remove the "\r\n" right before the boundary.
                // Having the extra "\r\n" corrupts the data if the content before the delimiter is binary.
                if (boundaryIndex > 0 && copied >= 2 && buffer[copied - 2] == '\r' && buffer[copied - 1] == '\n')
                { 
                    copied -= 2;
                } 
 
                if (boundary.Length == boundaryIndex)
                { 
                    break;
                }
                else if (0 < boundaryIndex)
                { 
                    if (boundaryIndex == this.byteLength)
                    {   // we need more data from underlying stream 
                        if (0 < this.bytePosition) 
                        {   // compress the buffer
                            Buffer.BlockCopy(data, this.bytePosition, data, 0, this.byteLength); 
                            this.bytePosition = 0;
                        }

                        int tmp = this.reader.Read(this.byteBuffer, this.byteLength, this.byteBuffer.Length - this.byteLength); 
                        if (null != this.writer)
                        { 
                            this.writer.Write(this.byteBuffer, this.byteLength, tmp); 
                        }
 
                        if (0 == tmp)
                        {   // partial boundary is at EOF
                            this.totalCount += boundaryIndex;
                            this.bytePosition += boundaryIndex; 
                            this.byteLength -= boundaryIndex;
                            this.batchLength -= boundaryIndex; 
 
                            offset += boundaryIndex;
                            count -= boundaryIndex; 
                            copied += boundaryIndex;
                            break;
                        }
 
                        // partial boundary not at EOF, restart the boundary match
                        this.byteLength += tmp; 
                    } 
                    else
                    {   // return smaller than requested buffer to user 
                        break;
                    }
                }
            } 

            return copied; 
        } 

        /// Read from internal buffer or use unbuffered read from underlying stream. 
        /// place to copy bytes read from underlying stream
        /// offset in buffer to start writing
        /// count of bytes to read from buffered underlying stream
        /// count of bytes actualy copied into buffer 
        private int ReadLength(byte[] buffer, int offset, int count)
        { 
            Debug.Assert(null != buffer, "null != buffer"); 
            Debug.Assert(0 <= offset, "0 <= offset");
            Debug.Assert(0 <= count, "0 <= count"); 
            Debug.Assert(offset + count <= buffer.Length, "offset + count <= buffer.Length");
            int copied = 0;

            if (0 < this.byteLength) 
            {   // initial read drains from our buffer
                int size = Math.Min(Math.Min(count, this.byteLength), this.batchLength); 
                Buffer.BlockCopy(this.byteBuffer, this.bytePosition, buffer, offset, size); 
                this.totalCount += size;
                this.bytePosition += size; 
                this.byteLength -= size;
                this.batchLength -= size;

                offset += size; 
                count -= size;
                copied = size; 
            } 

            if (0 < count && this.batchLength > 0) 
            {   // read remainder directly from stream
                int size = this.reader.Read(buffer, offset, Math.Min(count, this.batchLength));
                if (null != this.writer)
                { 
                    this.writer.Write(buffer, offset, size);
                } 
 
                this.totalCount += size;
                this.batchLength -= size; 
                copied += size;
            }

            Debug.Assert(0 <= this.byteLength, "negative byteLength"); 
            Debug.Assert(0 <= this.batchLength, "negative batchLength");
            return copied; 
        } 

        ///  
        /// Read the content headers
        /// 
        private void ReadContentHeaders()
        { 
            // Read the content headers u
            this.contentHeaders = new Dictionary(StringComparer.OrdinalIgnoreCase); 
            while (true) 
            {
                string line = this.ReadLine(); 
                if (0 < line.Length)
                {
                    int colon = line.IndexOf(':');
                    if (colon <= 0) 
                    {   // expecting "name: value"
                        throw Error.BatchStreamInvalidHeaderValueSpecified(line); 
                    } 

                    string name = line.Substring(0, colon).Trim(); 
                    string value = line.Substring(colon + 1).Trim();
                    this.contentHeaders.Add(name, value);
                }
                else 
                {
                    break; 
                } 
            }
        } 

        /// 
        /// Validate that the first header is the http method name, followed by url followed by http version
        /// E.g. POST /Customers HTTP/1.1 
        /// 
        private void ReadHttpHeaders() 
        { 
            // read the header line
            string line = this.ReadLine(); 

            // Batch Request: POST /Customers HTTP/1.1
            // Since the uri can contain spaces, the only way to read the request url, is to
            // check for first space character and last space character and anything between 
            // them.
 
            // Batch Response: HTTP/1.1 200 Ok 
            // Since the http status code strings have spaces in them, we cannot use the same
            // logic. We need to check for the second space and anything after that is the error 
            // message.
            int index1 = line.IndexOf(' ');
            if ((index1 <= 0) || ((line.Length - 3) <= index1))
            { 
                // only 1 segment or empty first segment or not enough left for 2nd and 3rd segments
                throw Error.BatchStreamInvalidMethodHeaderSpecified(line); 
            } 

            int index2 = (this.batchRequest ? line.LastIndexOf(' ') : line.IndexOf(' ', index1 + 1)); 
            if ((index2 < 0) || (index2 - index1 - 1 <= 0) || ((line.Length - 1) <= index2))
            {
                // only 2 segments or empty 2nd or 3rd segments
                throw Error.BatchStreamInvalidMethodHeaderSpecified(line); 
            }
 
            string segment1 = line.Substring(0, index1);                        // Request - Http method,  Response - Http version 
            string segment2 = line.Substring(index1 + 1, index2 - index1 - 1);  // Request - Request uri,  Response - Http status code
            string segment3 = line.Substring(index2 + 1);                       // Request - Http version, Response - Http status description 

            #region validate HttpVersion
            string httpVersion = this.batchRequest ? segment3 : segment1;
            if (httpVersion != XmlConstants.HttpVersionInBatching) 
            {
                throw Error.BatchStreamInvalidHttpVersionSpecified(httpVersion, XmlConstants.HttpVersionInBatching); 
            } 
            #endregion
 
            // read the actual http headers now
            this.ReadContentHeaders();

            BatchStreamState state; 
            if (this.batchRequest)
            { 
                state = GetStateBasedOnHttpMethodName(segment1); 
#if ASTORIA_SERVER
                this.contentUri = segment2; 
#endif
            }
            else
            { 
                // caller must use content-id to correlate response to action
                state = (BatchStreamState.EndBatch == this.batchState) ? BatchStreamState.GetResponse : BatchStreamState.ChangeResponse; 
#if ASTORIA_CLIENT 
                this.statusCode = segment2;
#endif 
            }

            #region validate state change
            Debug.Assert( 
                BatchStreamState.EndBatch == this.batchState ||
                BatchStreamState.EndChangeSet == this.batchState, 
                "unexpected BatchStreamState"); 

            if (this.batchState == BatchStreamState.EndBatch) 
            {
                if ((this.batchRequest && (state == BatchStreamState.Get)) ||
                    (!this.batchRequest && (state == BatchStreamState.GetResponse)))
                { 
                    this.batchState = state;
                } 
                else 
                {
                    throw Error.BatchStreamOnlyGETOperationsCanBeSpecifiedInBatch(); 
                }
            }
            else if (this.batchState == BatchStreamState.EndChangeSet)
            { 
                if ((this.batchRequest && ((BatchStreamState.Post == state) || (BatchStreamState.Put == state) || (BatchStreamState.Delete == state) || (BatchStreamState.Merge == state))) ||
                    (!this.batchRequest && (state == BatchStreamState.ChangeResponse))) 
                { 
                    this.batchState = state;
                } 
                else
                {
                    // setting the batch state to POST so that in the next round, we can have the correct
                    // state to start with. 
                    this.batchState = BatchStreamState.Post;
 
                    // bad http method verb for changeset 
                    throw Error.BatchStreamGetMethodNotSupportInChangeset();
                } 
            }
            else
            {   // bad state for operation to exist
                throw Error.BatchStreamInvalidOperationHeaderSpecified(); 
            }
            #endregion 
        } 

        ///  
        /// sub stream of BatchStream that reads up to a boundary delimiter
        /// 
        private sealed class StreamWithDelimiter : StreamWithLength
        { 
            /// 
            /// constructor 
            ///  
            /// underlying stream
            internal StreamWithDelimiter(BatchStream stream) 
                : base(stream, Int32.MaxValue)
            {
            }
 
            /// read bytes from stream
            /// buffer to store bytes being read 
            /// offset in buffer to start storing bytes 
            /// count of bytes to read
            /// count of bytes actualy read into the buffer 
            public override int Read(byte[] buffer, int offset, int count)
            {
                if (null == this.Target)
                { 
                    Error.ThrowObjectDisposed(this.GetType());
                } 
 
                int result = this.Target.ReadDelimiter(buffer, offset, count);
                return result; 
            }
        }

        ///  
        /// sub stream of BatchStream that reads a specific length from underlying stream
        ///  
        ///  
        /// Allows users of stream to call Dispose multiple times
        /// without affecting the BatchStream 
        /// 
        private class StreamWithLength : Stream
        {
            /// Underlying batch stream 
            private BatchStream target;
 
            /// Max remaining byte length to read from underlying stream 
            private int length;
 
            /// 
            /// constructor
            /// 
            /// underlying stream 
            /// max byte length to read
            internal StreamWithLength(BatchStream stream, int contentLength) 
            { 
                Debug.Assert(null != stream, "null != stream");
                Debug.Assert(0 < contentLength, "0 < contentLength"); 
                this.target = stream;
                this.length = contentLength;
            }
 
            /// Delegate to underlying stream
            public override bool CanRead 
            { 
                get { return (null != this.target && this.target.CanRead); }
            } 

            /// False
            public override bool CanSeek
            { 
                get { return false; }
            } 
 
            /// False
            public override bool CanWrite 
            {
                get { return false; }
            }
 
            /// Not supported.
            public override long Length 
            { 
                get { throw Error.NotSupported(); }
            } 

            /// Not supported.
            public override long Position
            { 
                get { throw Error.NotSupported(); }
                set { throw Error.NotSupported(); } 
            } 

            /// Underlying batch stream 
            internal BatchStream Target
            {
                get { return this.target; }
            } 

            /// Does nothing. 
            public override void Flush() 
            {
            } 

#if DEBUG && !ASTORIA_LIGHT // Synchronous methods not available
            /// Not supported.
            /// ignored 
            /// ignored
            /// ignored 
            /// ignored 
            /// ignored
            /// nothing 
            public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
            {
                throw Error.NotSupported();
            } 
#endif
 
            /// read bytes from stream 
            /// buffer to store bytes being read
            /// offset in buffer to start storing bytes 
            /// count of bytes to read
            /// count of bytes actualy read into the buffer
            public override int Read(byte[] buffer, int offset, int count)
            { 
                if (null == this.target)
                { 
                    Error.ThrowObjectDisposed(this.GetType()); 
                }
 
                int result = this.target.ReadLength(buffer, offset, Math.Min(count, this.length));
                this.length -= result;
                Debug.Assert(0 <= this.length, "Read beyond expected length");
                return result; 
            }
 
            /// Not supported. 
            /// The parameter is not used.
            /// The parameter is not used. 
            /// nothing
            public override long Seek(long offset, SeekOrigin origin)
            {
                throw Error.NotSupported(); 
            }
 
            /// Not supported. 
            /// ignored
            public override void SetLength(long value) 
            {
                throw Error.NotSupported();
            }
 
            /// Not supported.
            /// The parameter is not used. 
            /// The parameter is not used. 
            /// The parameter is not used.
            public override void Write(byte[] buffer, int offset, int count) 
            {
                throw Error.NotSupported();
            }
 
            /// Dispose of this nested stream.
            /// true if active dispose 
            protected override void Dispose(bool disposing) 
            {
                base.Dispose(disposing); 

                if (disposing && (null != this.target))
                {
                    if (this.target.disposeWithContentStreamDispose) 
                    {
                        this.target.contentStream = null; 
                        this.target.Dispose(); 
                    }
                    else if (0 < this.length) 
                    {
                        if (null != this.target.reader)
                        {
                            this.target.Seek(this.length, SeekOrigin.Current); 
                        }
 
                        this.length = 0; 
                    }
 
                    this.target.ClearPreviousOperationInformation();
                }

                this.target = null; 
            }
        } 
    } 
}

// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
                        

Link Menu

Network programming in C#, Network Programming in VB.NET, Network Programming in .NET
This book is available now!
Buy at Amazon US or
Buy at Amazon UK