/// <summary>
 /// Initializes a new instance of the BlobWriteStreamBase class.
 /// </summary>
 /// <param name="serviceClient">The service client.</param>        
 /// <param name="accessCondition">An <see cref="AccessCondition"/> object that represents the condition that must be met in order for the request to proceed. If <c>null</c>, no condition is used.</param>
 /// <param name="options">A <see cref="BlobRequestOptions"/> object that specifies additional options for the request.</param>
 /// <param name="operationContext">An <see cref="OperationContext"/> object that represents the context for the current operation.</param>
 private BlobWriteStreamBase(CloudBlobClient serviceClient, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
     : base()
 {
     this.internalBuffer = new MultiBufferMemoryStream(serviceClient.BufferManager);
     this.accessCondition = accessCondition;
     this.currentOffset = 0;
     this.options = options;
     this.operationContext = operationContext;
     this.noPendingWritesEvent = new CounterEvent();
     this.blobMD5 = this.options.StoreBlobContentMD5.Value ? new MD5Wrapper() : null;
     this.blockMD5 = this.options.UseTransactionalMD5.Value ? new MD5Wrapper() : null;
     this.parallelOperationSemaphore = new AsyncSemaphore(options.ParallelOperationThreadCount.Value);
     this.lastException = null;
     this.committed = false;
     this.disposed = false;
 }

        /// <summary>
        /// Initializes a new instance of the BlobReadStreamBase class.
        /// </summary>
        /// <param name="blob">Blob reference to read from</param>
        /// <param name="accessCondition">An <see cref="AccessCondition"/> object that represents the condition that must be met in order for the request to proceed. If <c>null</c>, no condition is used.</param>
        /// <param name="options">A <see cref="BlobRequestOptions"/> object that specifies additional options for the request.</param>
        /// <param name="operationContext">An <see cref="OperationContext"/> object that represents the context for the current operation.</param>
        protected BlobReadStreamBase(CloudBlob blob, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
        {
            if (options.UseTransactionalMD5.Value)
            {
                CommonUtility.AssertInBounds("StreamMinimumReadSizeInBytes", blob.StreamMinimumReadSizeInBytes, 1, Constants.MaxRangeGetContentMD5Size);
            }

            this.blob = blob;
            this.blobProperties = new BlobProperties(blob.Properties);
            this.currentOffset = 0;
            this.streamMinimumReadSizeInBytes = this.blob.StreamMinimumReadSizeInBytes;
            this.internalBuffer = new MultiBufferMemoryStream(blob.ServiceClient.BufferManager);
            this.accessCondition = accessCondition;
            this.options = options;
            this.operationContext = operationContext;
            this.blobMD5 = (this.options.DisableContentMD5Validation.Value || string.IsNullOrEmpty(this.blobProperties.ContentMD5)) ? null : new MD5Wrapper();
            this.lastException = null;
        }

 /// <summary>
 /// Initializes a new instance of the FileWriteStreamBase class for a file.
 /// </summary>
 /// <param name="file">File reference to write to.</param>
 /// <param name="fileSize">Size of the file.</param>
 /// <param name="createNew">Use <c>true</c> if the file is newly created, <c>false</c> otherwise.</param>
 /// <param name="accessCondition">An <see cref="AccessCondition"/> object that represents the access conditions for the file. If <c>null</c>, no condition is used.</param>
 /// <param name="options">An <see cref="FileRequestOptions"/> object that specifies additional options for the request.</param>
 /// <param name="operationContext">An <see cref="OperationContext"/> object for tracking the current operation.</param>
 protected FileWriteStreamBase(CloudFile file, long fileSize, bool createNew, AccessCondition accessCondition, FileRequestOptions options, OperationContext operationContext)
     : base()
 {
     this.internalBuffer = new MultiBufferMemoryStream(file.ServiceClient.BufferManager);
     this.currentOffset = 0;
     this.accessCondition = accessCondition;
     this.options = options;
     this.operationContext = operationContext;
     this.noPendingWritesEvent = new CounterEvent();
     this.fileMD5 = this.options.StoreFileContentMD5.Value ? new MD5Wrapper() : null;
     this.rangeMD5 = this.options.UseTransactionalMD5.Value ? new MD5Wrapper() : null;
     this.parallelOperationSemaphore = new AsyncSemaphore(options.ParallelOperationThreadCount.Value);
     this.lastException = null;
     this.committed = false;
     this.disposed = false;
     this.currentFileOffset = 0;
     this.file = file;
     this.fileSize = fileSize;
     this.streamWriteSizeInBytes = file.StreamWriteSizeInBytes;
     this.newFile = createNew;
 }

        public async Task WriteToMultiBufferMemoryStreamTestAsync()
        {
            OperationContext tempOperationContext = new OperationContext();
            RESTCommand<NullType> cmd = new RESTCommand<NullType>(TestBase.StorageCredentials, null);
            ExecutionState<NullType> tempExecutionState = new ExecutionState<NullType>(cmd, null, tempOperationContext);

            byte[] buffer = GetRandomBuffer(1 * 1024 * 1024);
            MemoryStream stream1 = new MemoryStream(buffer);

            MultiBufferMemoryStream stream2 = new MultiBufferMemoryStream(null /* bufferManager */);
            await stream1.WriteToAsync(stream2, null, null, false, tempExecutionState, null, CancellationToken.None);
            stream1.Seek(0, SeekOrigin.Begin);
            stream2.Seek(0, SeekOrigin.Begin);
            TestHelper.AssertStreamsAreEqual(stream1, stream2);

            MultiBufferMemoryStream stream3 = new MultiBufferMemoryStream(null /* bufferManager */);
            await TestHelper.ExpectedExceptionAsync<TimeoutException>(
                () => stream2.FastCopyToAsync(stream3, DateTime.Now.AddMinutes(-1)),
                "Past expiration time should immediately fail");
            stream2.Seek(0, SeekOrigin.Begin);
            stream3.Seek(0, SeekOrigin.Begin);
            await stream2.FastCopyToAsync(stream3, DateTime.Now.AddHours(1));
            stream2.Seek(0, SeekOrigin.Begin);
            stream3.Seek(0, SeekOrigin.Begin);
            TestHelper.AssertStreamsAreEqual(stream2, stream3);

            MultiBufferMemoryStream stream4 = new MultiBufferMemoryStream(null /* bufferManager */, 12345);
            await stream3.FastCopyToAsync(stream4, null);
            stream3.Seek(0, SeekOrigin.Begin);
            stream4.Seek(0, SeekOrigin.Begin);
            TestHelper.AssertStreamsAreEqual(stream3, stream4);

            MemoryStream stream5 = new MemoryStream();
            await stream4.WriteToAsync(stream5, null, null, false, tempExecutionState, null, CancellationToken.None);
            stream4.Seek(0, SeekOrigin.Begin);
            stream5.Seek(0, SeekOrigin.Begin);
            TestHelper.AssertStreamsAreEqual(stream4, stream5);

            TestHelper.AssertStreamsAreEqual(stream1, stream5);
        }

        /// <summary>
        /// Implementation for the SetPermissions method.
        /// </summary>
        /// <param name="acl">The permissions to set.</param>
        /// <param name="accessCondition">An <see cref="AccessCondition"/> object that represents the condition that must be met in order for the request to proceed. If <c>null</c>, no condition is used.</param>
        /// <param name="options">A <see cref="BlobRequestOptions"/> object that specifies additional options for the request.</param>
        /// <returns>A <see cref="RESTCommand{T}"/> that sets the permissions.</returns>
        private RESTCommand<NullType> SetPermissionsImpl(BlobContainerPermissions acl, AccessCondition accessCondition, BlobRequestOptions options)
        {
            MultiBufferMemoryStream memoryStream = new MultiBufferMemoryStream(null /* bufferManager */, (int)(1 * Constants.KB));
            BlobRequest.WriteSharedAccessIdentifiers(acl.SharedAccessPolicies, memoryStream);
            memoryStream.Seek(0, SeekOrigin.Begin);

            RESTCommand<NullType> putCmd = new RESTCommand<NullType>(this.ServiceClient.Credentials, this.StorageUri);

            options.ApplyToStorageCommand(putCmd);
            putCmd.BuildRequestDelegate = (uri, builder, serverTimeout, useVersionHeader, ctx) => ContainerHttpWebRequestFactory.SetAcl(uri, serverTimeout, acl.PublicAccess, accessCondition, useVersionHeader, ctx);
            putCmd.SendStream = memoryStream;
            putCmd.StreamToDispose = memoryStream;
            putCmd.RecoveryAction = RecoveryActions.RewindStream;
            putCmd.SignRequest = this.ServiceClient.AuthenticationHandler.SignRequest;
            putCmd.PreProcessResponse = (cmd, resp, ex, ctx) =>
            {
                HttpResponseParsers.ProcessExpectedStatusCodeNoException(HttpStatusCode.OK, resp, NullType.Value, cmd, ex);
                this.UpdateETagAndLastModified(resp);
                return NullType.Value;
            };

            return putCmd;
        }

        public virtual void WriteRange(Stream rangeData, long startOffset, string contentMD5 = null, AccessCondition accessCondition = null, FileRequestOptions options = null, OperationContext operationContext = null)
        {
            CommonUtility.AssertNotNull("rangeData", rangeData);

            FileRequestOptions modifiedOptions = FileRequestOptions.ApplyDefaults(options, this.ServiceClient);
            bool requiresContentMD5 = (contentMD5 == null) && modifiedOptions.UseTransactionalMD5.Value;
            operationContext = operationContext ?? new OperationContext();

            Stream seekableStream = rangeData;
            bool seekableStreamCreated = false;

            try
            {
                if (!rangeData.CanSeek || requiresContentMD5)
                {
                    ExecutionState<NullType> tempExecutionState = CommonUtility.CreateTemporaryExecutionState(modifiedOptions);

                    Stream writeToStream;
                    if (rangeData.CanSeek)
                    {
                        writeToStream = Stream.Null;
                    }
                    else
                    {
                        seekableStream = new MultiBufferMemoryStream(this.ServiceClient.BufferManager);
                        seekableStreamCreated = true;
                        writeToStream = seekableStream;
                    }

                    long startPosition = seekableStream.Position;
                    StreamDescriptor streamCopyState = new StreamDescriptor();
                    rangeData.WriteToSync(writeToStream, null /* copyLength */, Constants.MaxBlockSize, requiresContentMD5, true, tempExecutionState, streamCopyState);
                    seekableStream.Position = startPosition;

                    if (requiresContentMD5)
                    {
                        contentMD5 = streamCopyState.Md5;
                    }
                }

                Executor.ExecuteSync(
                    this.PutRangeImpl(seekableStream, startOffset, contentMD5, accessCondition, modifiedOptions),
                    modifiedOptions.RetryPolicy,
                    operationContext);
            }
            finally
            {
                if (seekableStreamCreated)
                {
                    seekableStream.Dispose();
                }
            }
        }

        public IAsyncAction WriteRangeAsync(IInputStream rangeData, long startOffset, string contentMD5, AccessCondition accessCondition, FileRequestOptions options, OperationContext operationContext)
#endif
        {
        CommonUtility.AssertNotNull("rangeData", rangeData);

            FileRequestOptions modifiedOptions = FileRequestOptions.ApplyDefaults(options, this.ServiceClient);
            bool requiresContentMD5 = (contentMD5 == null) && modifiedOptions.UseTransactionalMD5.Value;
            operationContext = operationContext ?? new OperationContext();
            ExecutionState<NullType> tempExecutionState = CommonUtility.CreateTemporaryExecutionState(modifiedOptions);

#if ASPNET_K
            return Task.Run(async () =>
#else
            return AsyncInfo.Run(async (cancellationToken) =>
#endif
            {
                DateTime streamCopyStartTime = DateTime.Now;

                Stream rangeDataAsStream = rangeData.AsStreamForRead();
                Stream seekableStream = rangeDataAsStream;
                if (!rangeDataAsStream.CanSeek || requiresContentMD5)
                {
                    Stream writeToStream;
                    if (rangeDataAsStream.CanSeek)
                    {
                        writeToStream = Stream.Null;
                    }
                    else
                    {
                        seekableStream = new MultiBufferMemoryStream(this.ServiceClient.BufferManager);
                        writeToStream = seekableStream;
                    }

                    StreamDescriptor streamCopyState = new StreamDescriptor();
                    long startPosition = seekableStream.Position;
                    await rangeDataAsStream.WriteToAsync(writeToStream, null /* copyLength */, Constants.MaxBlockSize, requiresContentMD5, tempExecutionState, streamCopyState, cancellationToken);
                    seekableStream.Position = startPosition;

                    if (requiresContentMD5)
                    {
                        contentMD5 = streamCopyState.Md5;
                    }

                    if (modifiedOptions.MaximumExecutionTime.HasValue)
                    {
                        modifiedOptions.MaximumExecutionTime -= DateTime.Now.Subtract(streamCopyStartTime);
                    }
                }

                await Executor.ExecuteAsyncNullReturn(
                    this.PutRangeImpl(seekableStream, startOffset, contentMD5, accessCondition, modifiedOptions),
                    modifiedOptions.RetryPolicy,
                    operationContext,
                    cancellationToken);
#if ASPNET_K
            }, cancellationToken);
#else
            });

        /// <summary>
        /// Uploads the block list.
        /// </summary>
        /// <param name="blocks">The blocks to upload.</param>
        /// <param name="accessCondition">An <see cref="AccessCondition"/> object that represents the access conditions for the blob. If <c>null</c>, no condition is used.</param>
        /// <param name="options">A <see cref="BlobRequestOptions"/> object that specifies any additional options for the request.</param>
        /// <returns>A <see cref="RESTCommand{T}"/> that uploads the block list.</returns>
        internal RESTCommand<NullType> PutBlockListImpl(IEnumerable<PutBlockListItem> blocks, AccessCondition accessCondition, BlobRequestOptions options)
        {
            MultiBufferMemoryStream memoryStream = new MultiBufferMemoryStream(this.ServiceClient.BufferManager);
            BlobRequest.WriteBlockListBody(blocks, memoryStream);
            memoryStream.Seek(0, SeekOrigin.Begin);
#if !WINDOWS_PHONE
            string contentMD5 = memoryStream.ComputeMD5Hash();
            memoryStream.Seek(0, SeekOrigin.Begin);
#endif

            RESTCommand<NullType> putCmd = new RESTCommand<NullType>(this.ServiceClient.Credentials, this.Uri);

            putCmd.ApplyRequestOptions(options);
            putCmd.BuildRequestDelegate = (uri, builder, serverTimeout, ctx) => BlobHttpWebRequestFactory.PutBlockList(uri, serverTimeout, this.Properties, accessCondition, ctx);
            putCmd.SetHeaders = (r, ctx) =>
            {
#if !WINDOWS_PHONE
                r.Headers[HttpRequestHeader.ContentMd5] = contentMD5;
#endif
                BlobHttpWebRequestFactory.AddMetadata(r, this.Metadata);
            };
            putCmd.SendStream = memoryStream;
            putCmd.RecoveryAction = RecoveryActions.RewindStream;
            putCmd.SignRequest = this.ServiceClient.AuthenticationHandler.SignRequest;
            putCmd.PreProcessResponse = (cmd, resp, ex, ctx) =>
            {
                HttpResponseParsers.ProcessExpectedStatusCodeNoException(HttpStatusCode.Created, resp, NullType.Value, cmd, ex);
                CloudBlobSharedImpl.UpdateETagLMTAndSequenceNumber(this.attributes, resp);
                this.Properties.Length = -1;
                return NullType.Value;
            };

            return putCmd;
        }

        public void MultiBufferMemoryStreamReadSeekSetLengthTestAPM()
        {
            byte[] outBuffer = new byte[2 * 1024 * 1024];
            byte[] buffer = GetRandomBuffer(outBuffer.Length);

            using (MemoryStream memStream = new MemoryStream())
            {
                memStream.Write(buffer, 0, buffer.Length);
                using (MultiBufferMemoryStream multiBufferStream = new MultiBufferMemoryStream(null /* bufferManager */))
                {
                    using (AutoResetEvent waitHandle = new AutoResetEvent(false))
                    {
                        IAsyncResult result = multiBufferStream.BeginWrite(buffer, 0, buffer.Length, ar => waitHandle.Set(), null);
                        waitHandle.WaitOne();
                        multiBufferStream.EndWrite(result);
                        multiBufferStream.Seek(0, SeekOrigin.Begin);
                        TestHelper.AssertStreamsAreEqual(memStream, multiBufferStream);

                        result = multiBufferStream.BeginRead(outBuffer, 0, buffer.Length, ar => waitHandle.Set(), null);
                        waitHandle.WaitOne();
                        multiBufferStream.EndRead(result);
                        TestHelper.AssertBuffersAreEqual(buffer, outBuffer);

                        multiBufferStream.Seek(-1, SeekOrigin.End);
                        Assert.AreEqual(buffer.Length - 1, multiBufferStream.Position);

                        multiBufferStream.Seek(-1024, SeekOrigin.End);
                        memStream.Seek(-1024, SeekOrigin.End);
                        TestHelper.AssertStreamsAreEqual(multiBufferStream, memStream);

                        multiBufferStream.SetLength(3 * 1024 * 1024);
                        memStream.SetLength(3 * 1024 * 1024);
                        multiBufferStream.Seek(0, SeekOrigin.Begin);
                        memStream.Seek(0, SeekOrigin.Begin);
                        TestHelper.AssertStreamsAreEqual(memStream, multiBufferStream);
                    }
                }
            }
        }

        /// <summary>
        /// Implementation for the SetPermissions method.
        /// </summary>
        /// <param name="acl">The permissions to set.</param>
        /// <param name="accessCondition">An <see cref="AccessCondition"/> object that represents the access conditions for the container. If <c>null</c>, no condition is used.</param>
        /// <param name="options">A <see cref="BlobRequestOptions"/> object that specifies additional options for the request.</param>
        /// <returns>A <see cref="RESTCommand"/> that sets the permissions.</returns>
        private RESTCommand<NullType> SetPermissionsImpl(BlobContainerPermissions acl, AccessCondition accessCondition, BlobRequestOptions options)
        {
            MultiBufferMemoryStream memoryStream = new MultiBufferMemoryStream(null /* bufferManager */, (int)(1 * Constants.KB));
            BlobRequest.WriteSharedAccessIdentifiers(acl.SharedAccessPolicies, memoryStream);

            RESTCommand<NullType> putCmd = new RESTCommand<NullType>(this.ServiceClient.Credentials, this.StorageUri);

            options.ApplyToStorageCommand(putCmd);
            putCmd.Handler = this.ServiceClient.AuthenticationHandler;
            putCmd.BuildClient = HttpClientFactory.BuildHttpClient;
            putCmd.BuildRequest = (cmd, uri, builder, cnt, serverTimeout, ctx) => ContainerHttpRequestMessageFactory.SetAcl(uri, serverTimeout, acl.PublicAccess, accessCondition, cnt, ctx);
            putCmd.BuildContent = (cmd, ctx) => HttpContentFactory.BuildContentFromStream(memoryStream, 0, memoryStream.Length, null /* md5 */, cmd, ctx);
            putCmd.PreProcessResponse = (cmd, resp, ex, ctx) =>
            {
                HttpResponseParsers.ProcessExpectedStatusCodeNoException(HttpStatusCode.OK, resp, NullType.Value, cmd, ex);
                this.UpdateETagAndLastModified(resp);
                return NullType.Value;
            };

            return putCmd;
        }

        public void PutBlock(string blockId, Stream blockData, string contentMD5, AccessCondition accessCondition = null, BlobRequestOptions options = null, OperationContext operationContext = null)
        {
            CommonUtility.AssertNotNull("blockData", blockData);

            BlobRequestOptions modifiedOptions = BlobRequestOptions.ApplyDefaults(options, BlobType.BlockBlob, this.ServiceClient);
            bool requiresContentMD5 = (contentMD5 == null) && modifiedOptions.UseTransactionalMD5.Value;
            operationContext = operationContext ?? new OperationContext();

            Stream seekableStream = blockData;
            if (!blockData.CanSeek || requiresContentMD5)
            {
                ExecutionState<NullType> tempExecutionState = CommonUtility.CreateTemporaryExecutionState(modifiedOptions);

                Stream writeToStream;
                if (blockData.CanSeek)
                {
                    writeToStream = Stream.Null;
                }
                else
                {
                    seekableStream = new MultiBufferMemoryStream(this.ServiceClient.BufferManager);
                    writeToStream = seekableStream;
                }

                long startPosition = seekableStream.Position;
                StreamDescriptor streamCopyState = new StreamDescriptor();
                blockData.WriteToSync(writeToStream, null /* copyLength */, Constants.MaxBlockSize, requiresContentMD5, true, tempExecutionState, streamCopyState);
                seekableStream.Position = startPosition;

                if (requiresContentMD5)
                {
                    contentMD5 = streamCopyState.Md5;
                }
            }

            Executor.ExecuteSync(
                this.PutBlockImpl(seekableStream, blockId, contentMD5, accessCondition, modifiedOptions),
                modifiedOptions.RetryPolicy,
                operationContext);
        }

        /// <summary>
        /// Uploads the block list.
        /// </summary>
        /// <param name="blocks">The blocks to upload.</param>
        /// <param name="accessCondition">An <see cref="AccessCondition"/> object that represents the access conditions for the blob. If <c>null</c>, no condition is used.</param>
        /// <param name="options">A <see cref="BlobRequestOptions"/> object that specifies any additional options for the request.</param>
        /// <returns>A <see cref="RESTCommand"/> that uploads the block list.</returns>
        internal RESTCommand<NullType> PutBlockListImpl(IEnumerable<PutBlockListItem> blocks, AccessCondition accessCondition, BlobRequestOptions options)
        {
            MultiBufferMemoryStream memoryStream = new MultiBufferMemoryStream(null /* bufferManager */, (int)(1 * Constants.KB));
            BlobRequest.WriteBlockListBody(blocks, memoryStream);
            memoryStream.Seek(0, SeekOrigin.Begin);
            string contentMD5 = memoryStream.ComputeMD5Hash();

            RESTCommand<NullType> putCmd = new RESTCommand<NullType>(this.ServiceClient.Credentials, this.Uri);

            putCmd.ApplyRequestOptions(options);
            putCmd.Handler = this.ServiceClient.AuthenticationHandler;
            putCmd.BuildClient = HttpClientFactory.BuildHttpClient;
            putCmd.BuildContent = (cmd, ctx) => HttpContentFactory.BuildContentFromStream(memoryStream, 0, memoryStream.Length, contentMD5, cmd, ctx);
            putCmd.BuildRequest = (cmd, cnt, ctx) =>
            {
                HttpRequestMessage msg = BlobHttpRequestMessageFactory.PutBlockList(cmd.Uri, cmd.ServerTimeoutInSeconds, this.Properties, accessCondition, cnt, ctx);
                BlobHttpRequestMessageFactory.AddMetadata(msg, this.Metadata);
                return msg;
            };
            putCmd.PreProcessResponse = (cmd, resp, ex, ctx) =>
            {
                HttpResponseParsers.ProcessExpectedStatusCodeNoException(HttpStatusCode.Created, resp, NullType.Value, cmd, ex);
                CloudBlobSharedImpl.UpdateETagLMTAndSequenceNumber(this.attributes, resp);
                this.Properties.Length = -1;
                return NullType.Value;
            };

            return putCmd;
        }

        public IAsyncAction PutBlockAsync(string blockId, IInputStream blockData, string contentMD5, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
        {
            BlobRequestOptions modifiedOptions = BlobRequestOptions.ApplyDefaults(options, BlobType.BlockBlob, this.ServiceClient);
            bool requiresContentMD5 = (contentMD5 == null) && modifiedOptions.UseTransactionalMD5.Value;
            operationContext = operationContext ?? new OperationContext();
            ExecutionState<NullType> tempExecutionState = CommonUtility.CreateTemporaryExecutionState(modifiedOptions);

            return AsyncInfo.Run(async (token) =>
            {
                Stream blockDataAsStream = blockData.AsStreamForRead();
                Stream seekableStream = blockDataAsStream;
                if (!blockDataAsStream.CanSeek || requiresContentMD5)
                {
                    Stream writeToStream;
                    if (blockDataAsStream.CanSeek)
                    {
                        writeToStream = Stream.Null;
                    }
                    else
                    {
                        seekableStream = new MultiBufferMemoryStream(this.ServiceClient.BufferManager);
                        writeToStream = seekableStream;
                    }

                    StreamDescriptor streamCopyState = new StreamDescriptor();
                    long startPosition = seekableStream.Position;
                    await blockDataAsStream.WriteToAsync(writeToStream, null /* copyLength */, Constants.MaxBlockSize, requiresContentMD5, tempExecutionState, streamCopyState, token);
                    seekableStream.Position = startPosition;

                    if (requiresContentMD5)
                    {
                        contentMD5 = streamCopyState.Md5;
                    }
                }

                await Executor.ExecuteAsyncNullReturn(
                    this.PutBlockImpl(seekableStream, blockId, contentMD5, accessCondition, modifiedOptions),
                    modifiedOptions.RetryPolicy,
                    operationContext,
                    token);
            });
        }

        /// <summary>
        /// Implementation for the AddMessage method.
        /// </summary>
        /// <param name="message">The message.</param>
        /// <param name="timeToLive">The maximum time to allow the message to be in the queue, or null.</param>
        /// <param name="initialVisibilityDelay">The length of time from now during which the message will be invisible.
        /// If <c>null</c> then the message will be visible immediately.</param>
        /// <param name="options">A <see cref="QueueRequestOptions"/> object that specifies additional options for the request.</param>
        /// <returns>A <see cref="RESTCommand"/> that sets the permissions.</returns>
        private RESTCommand<NullType> AddMessageImpl(CloudQueueMessage message, TimeSpan? timeToLive, TimeSpan? initialVisibilityDelay, QueueRequestOptions options)
        {
            int? timeToLiveInSeconds = null;
            int? initialVisibilityDelayInSeconds = null;

            if (timeToLive.HasValue)
            {
                CommonUtility.AssertInBounds("timeToLive", timeToLive.Value, TimeSpan.Zero, CloudQueueMessage.MaxTimeToLive);
                timeToLiveInSeconds = (int)timeToLive.Value.TotalSeconds;
            }

            if (initialVisibilityDelay.HasValue)
            {
                CommonUtility.AssertInBounds("initialVisibilityDelay", initialVisibilityDelay.Value, TimeSpan.Zero, timeToLive ?? CloudQueueMessage.MaxTimeToLive);
                initialVisibilityDelayInSeconds = (int)initialVisibilityDelay.Value.TotalSeconds;
            }

            MultiBufferMemoryStream memoryStream = new MultiBufferMemoryStream(null /* bufferManager */, (int)(1 * Constants.KB));
            QueueRequest.WriteMessageContent(message.GetMessageContentForTransfer(this.EncodeMessage), memoryStream);
            memoryStream.Seek(0, SeekOrigin.Begin);

            RESTCommand<NullType> putCmd = new RESTCommand<NullType>(this.ServiceClient.Credentials, this.GetMessageRequestAddress());

            options.ApplyToStorageCommand(putCmd);
            putCmd.BuildRequest = (cmd, uri, builder, cnt, serverTimeout, ctx) => QueueHttpRequestMessageFactory.AddMessage(uri, serverTimeout, timeToLiveInSeconds, initialVisibilityDelayInSeconds, cnt, ctx, this.ServiceClient.GetCanonicalizer(), this.ServiceClient.Credentials);
            putCmd.BuildContent = (cmd, ctx) => HttpContentFactory.BuildContentFromStream(memoryStream, 0, memoryStream.Length, null, cmd, ctx);
            putCmd.StreamToDispose = memoryStream;
            putCmd.PreProcessResponse = (cmd, resp, ex, ctx) =>
            {
                HttpResponseParsers.ProcessExpectedStatusCodeNoException(HttpStatusCode.Created, resp, NullType.Value, cmd, ex);
                GetMessageCountAndMetadataFromResponse(resp);
                return NullType.Value;
            };

            return putCmd;
        }

        /// <summary>
        /// Uploads the block list.
        /// </summary>
        /// <param name="blocks">The blocks to upload.</param>
        /// <param name="accessCondition">An <see cref="AccessCondition"/> object that represents the access conditions for the blob. If <c>null</c>, no condition is used.</param>
        /// <param name="options">A <see cref="BlobRequestOptions"/> object that specifies additional options for the request.</param>
        /// <returns>A <see cref="RESTCommand"/> that uploads the block list.</returns>
        internal RESTCommand<NullType> PutBlockListImpl(IEnumerable<PutBlockListItem> blocks, AccessCondition accessCondition, BlobRequestOptions options)
        {
            MultiBufferMemoryStream memoryStream = new MultiBufferMemoryStream(null /* bufferManager */, (int)(1 * Constants.KB));
            BlobRequest.WriteBlockListBody(blocks, memoryStream);
            memoryStream.Seek(0, SeekOrigin.Begin);
            string contentMD5 = null;

            if (options.UseTransactionalMD5.HasValue && options.UseTransactionalMD5.Value)
            {
                contentMD5 = memoryStream.ComputeMD5Hash();
            }

            RESTCommand<NullType> putCmd = new RESTCommand<NullType>(this.ServiceClient.Credentials, this.attributes.StorageUri);

            options.ApplyToStorageCommand(putCmd);
            putCmd.BuildContent = (cmd, ctx) => HttpContentFactory.BuildContentFromStream(memoryStream, 0, memoryStream.Length, contentMD5, cmd, ctx);
            putCmd.StreamToDispose = memoryStream;
            putCmd.BuildRequest = (cmd, uri, builder, cnt, serverTimeout, ctx) =>
            {
                StorageRequestMessage msg = BlobHttpRequestMessageFactory.PutBlockList(uri, serverTimeout, this.Properties, accessCondition, cnt, ctx, this.ServiceClient.GetCanonicalizer(), this.ServiceClient.Credentials);
                BlobHttpRequestMessageFactory.AddMetadata(msg, this.Metadata);
                return msg;
            };
            putCmd.PreProcessResponse = (cmd, resp, ex, ctx) =>
            {
                HttpResponseParsers.ProcessExpectedStatusCodeNoException(HttpStatusCode.Created, resp, NullType.Value, cmd, ex);
                CloudBlob.UpdateETagLMTLengthAndSequenceNumber(this.attributes, resp, false);
                cmd.CurrentResult.IsRequestServerEncrypted = CloudBlob.ParseServerRequestEncrypted(resp);
                this.Properties.Length = -1;
                return NullType.Value;
            };

            return putCmd;
        }

        private RESTCommand<NullType> SetServicePropertiesImpl(ServiceProperties properties, TableRequestOptions requestOptions)
        {
            MultiBufferMemoryStream str = new MultiBufferMemoryStream(null /* bufferManager */, (int)(1 * Constants.KB));
            try
            {
                properties.WriteServiceProperties(str);
            }
            catch (InvalidOperationException invalidOpException)
            {
                str.Dispose();
                throw new ArgumentException(invalidOpException.Message, "properties");
            }

            str.Seek(0, SeekOrigin.Begin);

            RESTCommand<NullType> retCmd = new RESTCommand<NullType>(this.Credentials, this.StorageUri);
            retCmd.SendStream = str;
            retCmd.StreamToDispose = str;
            retCmd.BuildRequestDelegate = TableHttpWebRequestFactory.SetServiceProperties;
            retCmd.RecoveryAction = RecoveryActions.RewindStream;
            retCmd.SignRequest = this.AuthenticationHandler.SignRequest;
            retCmd.ParseError = StorageExtendedErrorInformation.ReadFromStreamUsingODataLib;
            retCmd.PreProcessResponse =
                (cmd, resp, ex, ctx) => HttpResponseParsers.ProcessExpectedStatusCodeNoException(HttpStatusCode.Accepted, resp, NullType.Value, cmd, ex);

            requestOptions.ApplyToStorageCommand(retCmd);
            return retCmd;
        }

        public void WriteToMultiBufferMemoryStreamTestAPM()
        {
            using (AutoResetEvent waitHandle = new AutoResetEvent(false))
            {
                byte[] buffer = GetRandomBuffer(1 * 1024 * 1024);
                MemoryStream stream1 = new MemoryStream(buffer);

                RESTCommand<NullType> cmd = new RESTCommand<NullType>(TestBase.StorageCredentials, null);
                ExecutionState<NullType> state = new ExecutionState<NullType>(cmd, new NoRetry(), new OperationContext());
                StreamDescriptor copyState = new StreamDescriptor();

                MultiBufferMemoryStream stream2 = new MultiBufferMemoryStream(null /* bufferManager */);
                stream1.WriteToAsync(stream2, null, null, false, state, copyState, _ => waitHandle.Set());
                waitHandle.WaitOne();
                if (state.ExceptionRef != null)
                {
                    throw state.ExceptionRef;
                }

                stream1.Seek(0, SeekOrigin.Begin);
                stream2.Seek(0, SeekOrigin.Begin);
                TestHelper.AssertStreamsAreEqual(stream1, stream2);

                MultiBufferMemoryStream stream3 = new MultiBufferMemoryStream(null /* bufferManager */);
                IAsyncResult ar = stream2.BeginFastCopyTo(stream3, DateTime.Now.AddMinutes(-1), _ => waitHandle.Set(), null);
                waitHandle.WaitOne();
                TestHelper.ExpectedException<TimeoutException>(
                    () => stream3.EndFastCopyTo(ar),
                    "Past expiration time should immediately fail");
                stream2.Seek(0, SeekOrigin.Begin);
                stream3.Seek(0, SeekOrigin.Begin);
                ar = stream2.BeginFastCopyTo(stream3, DateTime.Now.AddHours(1), _ => waitHandle.Set(), null);
                waitHandle.WaitOne();
                stream2.EndFastCopyTo(ar);
                stream2.Seek(0, SeekOrigin.Begin);
                stream3.Seek(0, SeekOrigin.Begin);
                TestHelper.AssertStreamsAreEqual(stream2, stream3);

                MultiBufferMemoryStream stream4 = new MultiBufferMemoryStream(null, 12345);
                ar = stream3.BeginFastCopyTo(stream4, null, _ => waitHandle.Set(), null);
                waitHandle.WaitOne();
                stream3.EndFastCopyTo(ar);
                stream3.Seek(0, SeekOrigin.Begin);
                stream4.Seek(0, SeekOrigin.Begin);
                TestHelper.AssertStreamsAreEqual(stream3, stream4);

                state = new ExecutionState<NullType>(cmd, new NoRetry(), new OperationContext());
                copyState = new StreamDescriptor();

                MemoryStream stream5 = new MemoryStream();
                stream4.WriteToAsync(stream5, null, null, false, state, copyState, _ => waitHandle.Set());
                waitHandle.WaitOne();
                if (state.ExceptionRef != null)
                {
                    throw state.ExceptionRef;
                }

                stream4.Seek(0, SeekOrigin.Begin);
                stream5.Seek(0, SeekOrigin.Begin);
                TestHelper.AssertStreamsAreEqual(stream4, stream5);

                TestHelper.AssertStreamsAreEqual(stream1, stream5);
            }
        }