Added AsStream to PipeReader and PipeWriter

src/System.IO.Pipelines/ref/System.IO.Pipelines.cs

@@ -44,6 +44,7 @@ public abstract partial class PipeReader
         protected PipeReader() { }
         public abstract void AdvanceTo(System.SequencePosition consumed);
         public abstract void AdvanceTo(System.SequencePosition consumed, System.SequencePosition examined);
+        public virtual System.IO.Stream AsStream() { throw null; }
         public abstract void CancelPendingRead();
         public abstract void Complete(System.Exception exception = null);
         public virtual System.Threading.Tasks.Task CopyToAsync(System.IO.Stream destination, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
@@ -62,6 +63,7 @@ public abstract partial class PipeWriter : System.Buffers.IBufferWriter<byte>
     {
         protected PipeWriter() { }
         public abstract void Advance(int bytes);
+        public virtual System.IO.Stream AsStream() { throw null; }
         public abstract void CancelPendingFlush();
         public abstract void Complete(System.Exception exception = null);
         protected internal virtual System.Threading.Tasks.Task CopyFromAsync(System.IO.Stream source, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }

src/System.IO.Pipelines/src/Resources/Strings.resx

@@ -1,17 +1,17 @@
 <?xml version="1.0" encoding="utf-8"?>
 <root>
-  <!--
-    Microsoft ResX Schema
-
+  <!-- 
+    Microsoft ResX Schema 
+    
     Version 2.0
-
-    The primary goals of this format is to allow a simple XML format
-    that is mostly human readable. The generation and parsing of the
-    various data types are done through the TypeConverter classes
+    
+    The primary goals of this format is to allow a simple XML format 
+    that is mostly human readable. The generation and parsing of the 
+    various data types are done through the TypeConverter classes 
     associated with the data types.
-
+    
     Example:
-
+    
     ... ado.net/XML headers & schema ...
     <resheader name="resmimetype">text/microsoft-resx</resheader>
     <resheader name="version">2.0</resheader>
@@ -26,36 +26,36 @@
         <value>[base64 mime encoded string representing a byte array form of the .NET Framework object]</value>
         <comment>This is a comment</comment>
     </data>
-
-    There are any number of "resheader" rows that contain simple
+                
+    There are any number of "resheader" rows that contain simple 
     name/value pairs.
-
-    Each data row contains a name, and value. The row also contains a
-    type or mimetype. Type corresponds to a .NET class that support
-    text/value conversion through the TypeConverter architecture.
-    Classes that don't support this are serialized and stored with the
+    
+    Each data row contains a name, and value. The row also contains a 
+    type or mimetype. Type corresponds to a .NET class that support 
+    text/value conversion through the TypeConverter architecture. 
+    Classes that don't support this are serialized and stored with the 
     mimetype set.
-
-    The mimetype is used for serialized objects, and tells the
-    ResXResourceReader how to depersist the object. This is currently not
+    
+    The mimetype is used for serialized objects, and tells the 
+    ResXResourceReader how to depersist the object. This is currently not 
     extensible. For a given mimetype the value must be set accordingly:
-
-    Note - application/x-microsoft.net.object.binary.base64 is the format
-    that the ResXResourceWriter will generate, however the reader can
+    
+    Note - application/x-microsoft.net.object.binary.base64 is the format 
+    that the ResXResourceWriter will generate, however the reader can 
     read any of the formats listed below.
-
+    
     mimetype: application/x-microsoft.net.object.binary.base64
-    value   : The object must be serialized with
+    value   : The object must be serialized with 
             : System.Runtime.Serialization.Formatters.Binary.BinaryFormatter
             : and then encoded with base64 encoding.
-
+    
     mimetype: application/x-microsoft.net.object.soap.base64
-    value   : The object must be serialized with
+    value   : The object must be serialized with 
             : System.Runtime.Serialization.Formatters.Soap.SoapFormatter
             : and then encoded with base64 encoding.
 
     mimetype: application/x-microsoft.net.object.bytearray.base64
-    value   : The object must be serialized into a byte array
+    value   : The object must be serialized into a byte array 
             : using a System.ComponentModel.TypeConverter
             : and then encoded with base64 encoding.
     -->
@@ -135,15 +135,24 @@
   <data name="ConcurrentOperationsNotSupported" xml:space="preserve">
     <value>Concurrent reads or writes are not supported.</value>
   </data>
+  <data name="FlushCanceledOnPipeWriter" xml:space="preserve">
+    <value>Flush was canceled on underlying PipeWriter.</value>
+  </data>
   <data name="GetResultBeforeCompleted" xml:space="preserve">
     <value>Can't GetResult unless awaiter is completed.</value>
   </data>
+  <data name="InvalidZeroByteRead" xml:space="preserve">
+    <value>The PipeReader returned 0 bytes when the ReadResult was not completed or cancelled.</value>
+  </data>
   <data name="NoReadingOperationToComplete" xml:space="preserve">
     <value>No reading operation to complete.</value>
   </data>
   <data name="NoWritingOperation" xml:space="preserve">
     <value>No writing operation. Make sure GetMemory() was called.</value>
   </data>
+  <data name="ReadCanceledOnPipeReader" xml:space="preserve">
+    <value>Read was canceled on underlying PipeReader.</value>
+  </data>
   <data name="ReaderAndWriterHasToBeCompleted" xml:space="preserve">
     <value>Both reader and writer has to be completed to be able to reset the pipe.</value>
   </data>
@@ -156,4 +165,4 @@
   <data name="WritingAfterCompleted" xml:space="preserve">
     <value>Writing is not allowed after writer was completed.</value>
   </data>
-</root>
+</root>

src/System.IO.Pipelines/src/System.IO.Pipelines.csproj

@@ -20,8 +20,10 @@
     <Compile Include="System\IO\Pipelines\PipeOptions.cs" />
     <Compile Include="System\IO\Pipelines\PipeReader.cs" />
     <Compile Include="System\IO\Pipelines\PipeOperationState.cs" />
+    <Compile Include="System\IO\Pipelines\PipeReaderStream.cs" />
     <Compile Include="System\IO\Pipelines\PipeScheduler.cs" />
     <Compile Include="System\IO\Pipelines\PipeWriter.cs" />
+    <Compile Include="System\IO\Pipelines\PipeWriterStream.cs" />
     <Compile Include="System\IO\Pipelines\ReadResult.cs" />
     <Compile Include="System\IO\Pipelines\ResultFlags.cs" />
     <Compile Include="System\IO\Pipelines\StreamPipeExtensions.cs" />

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeReader.cs

@@ -13,6 +13,8 @@ namespace System.IO.Pipelines
     /// </summary>
     public abstract partial class PipeReader
     {
+        private PipeReaderStream _stream;
+
         /// <summary>
         /// Attempt to synchronously read data the <see cref="PipeReader"/>.
         /// </summary>
@@ -48,6 +50,15 @@ public abstract partial class PipeReader
         /// </remarks>
         public abstract void AdvanceTo(SequencePosition consumed, SequencePosition examined);
 
+        /// <summary>
+        /// 
+        /// </summary>
+        /// <returns></returns>
+        public virtual Stream AsStream()
+        {
+            return _stream ?? (_stream = new PipeReaderStream(this));
+        }
+
         /// <summary>
         /// Cancel to currently pending or if none is pending next call to <see cref="ReadAsync"/>, without completing the <see cref="PipeReader"/>.
         /// </summary>
@@ -99,7 +110,7 @@ private async Task CopyToAsyncCore(Stream destination, CancellationToken cancell
 
                     if (result.IsCanceled)
                     {
-                        throw new OperationCanceledException();
+                        ThrowHelper.ThrowOperationCanceledException_ReadCanceled();
                     }
 
                     while (buffer.TryGet(ref position, out ReadOnlyMemory<byte> memory))

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeReaderStream.cs

@@ -0,0 +1,103 @@
+using System;
+using System.Buffers;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.IO.Pipelines
+{
+    internal class PipeReaderStream : Stream
+    {
+        private readonly PipeReader _pipeReader;
+
+        public PipeReaderStream(PipeReader pipeReader)
+        {
+            Debug.Assert(pipeReader != null);
+            _pipeReader = pipeReader;
+        }
+
+        public override bool CanRead => true;
+
+        public override bool CanSeek => false;
+
+        public override bool CanWrite => false;
+
+        public override long Length => throw new NotSupportedException();
+
+        public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
+
+        public override void Flush() => throw new NotSupportedException();
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            return ReadAsync(buffer, offset, count).GetAwaiter().GetResult();
+        }
+
+        public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+
+        public override void SetLength(long value) => throw new NotSupportedException();
+
+        public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+
+        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            return ReadAsyncInternal(new Memory<byte>(buffer, offset, count), cancellationToken).AsTask();
+        }
+
+#if !netstandard
+        public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            return ReadAsyncInternal(buffer, cancellationToken);
+        }
+#endif
+
+        private async ValueTask<int> ReadAsyncInternal(Memory<byte> buffer, CancellationToken cancellationToken)
+        {
+            ReadResult result = await _pipeReader.ReadAsync(cancellationToken).ConfigureAwait(false);
+
+            if (result.IsCanceled)
+            {
+                ThrowHelper.ThrowOperationCanceledException_ReadCanceled();
+            }
+
+            ReadOnlySequence<byte> sequence = result.Buffer;
+            long bufferLength = sequence.Length;
+            SequencePosition consumed = sequence.End;
+
+            try
+            {
+                if (bufferLength != 0)
+                {
+                    int actual = (int)Math.Min(bufferLength, buffer.Length);
+
+                    ReadOnlySequence<byte> slice = actual == bufferLength ? sequence : sequence.Slice(0, actual);
+                    consumed = slice.End;
+                    slice.CopyTo(buffer.Span);
+
+                    return actual;
+                }
+
+                if (result.IsCompleted)
+                {
+                    return 0;
+                }
+            }
+            finally
+            {
+                _pipeReader.AdvanceTo(consumed);
+            }
+
+            // This is a buggy PipeReader implementation that returns 0 byte reads even though the PipeReader
+            // isn't completed or cancelled
+            ThrowHelper.ThrowInvalidOperationException_InvalidZeroByteRead();
+            return 0;
+        }
+
+        /// <inheritdoc />
+        public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
+        {
+            // Delegate to CopyToAsync on the PipeReader
+            return _pipeReader.CopyToAsync(destination, cancellationToken);
+        }
+    }
+}

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeWriter.cs

@@ -13,6 +13,8 @@ namespace System.IO.Pipelines
     /// </summary>
     public abstract partial class PipeWriter : IBufferWriter<byte>
     {
+        private PipeWriterStream _stream;
+
         /// <summary>
         /// Marks the <see cref="PipeWriter"/> as being complete, meaning no more items will be written to it.
         /// </summary>
@@ -43,6 +45,15 @@ public abstract partial class PipeWriter : IBufferWriter<byte>
         /// <inheritdoc />
         public abstract Span<byte> GetSpan(int sizeHint = 0);
 
+        /// <summary>
+        /// 
+        /// </summary>
+        /// <returns></returns>
+        public virtual Stream AsStream()
+        {
+            return _stream ?? (_stream = new PipeWriterStream(this));
+        }
+
         /// <summary>
         /// Writes <paramref name="source"/> to the pipe and makes data accessible to <see cref="PipeReader"/>
         /// </summary>
@@ -76,7 +87,7 @@ protected internal virtual async Task CopyFromAsync(Stream source, CancellationT
 
                 if (result.IsCanceled)
                 {
-                    throw new OperationCanceledException();
+                    ThrowHelper.ThrowOperationCanceledException_FlushCanceled();
                 }
 
                 if (result.IsCompleted)