Change IPipe* to abstract classes

corefxlab.sln

@@ -2,7 +2,7 @@ Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio 15
 VisualStudioVersion = 15.0.27130.2020
 MinimumVisualStudioVersion = 10.0.40219.1
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{5E7EB061-B9BC-4DA2-B5E5-859AA7C67695}"
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "MultiSegmentBytesReaderNumbers", "MultiSegmentBytesReaderNumbers", "{5E7EB061-B9BC-4DA2-B5E5-859AA7C67695}"
 	ProjectSection(SolutionItems) = preProject
 		global.json = global.json
 		NuGet.Config = NuGet.Config

samples/System.IO.Pipelines.Samples/CompressionSample.cs

@@ -34,7 +34,7 @@ public Task Run()
 
                 // Wrap the console in a pipeline writer
 
-                var outputPipe = new Pipe(options);
+                var outputPipe = new ResetablePipe(options);
                 outputPipe.Reader.CopyToEndAsync(Console.OpenStandardOutput());
 
                 // Copy from the file reader to the console writer

samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpClientHandler.cs

@@ -243,7 +243,7 @@ private static async Task ProduceResponse(ConnectionState state, IPipeConnection
             }
         }
 
-        private static void WriteHeaders(HttpHeaders headers, IPipeWriter buffer)
+        private static void WriteHeaders(HttpHeaders headers, PipeWriter buffer)
         {
             foreach (var header in headers)
             {

samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpContent.cs

@@ -10,9 +10,9 @@ namespace System.IO.Pipelines.Samples
 {
     public class PipelineHttpContent : HttpContent
     {
-        private readonly IPipeReader _output;
+        private readonly PipeReader _output;
 
-        public PipelineHttpContent(IPipeReader output)
+        public PipelineHttpContent(PipeReader output)
         {
             _output = output;
         }

samples/System.IO.Pipelines.Samples/HttpServer/HttpConnection.cs

@@ -17,8 +17,8 @@ public partial class HttpConnection<TContext>
         private static readonly byte[] _chunkedEndBytes = Encoding.UTF8.GetBytes("0\r\n\r\n");
         private static readonly byte[] _endChunkBytes = Encoding.ASCII.GetBytes("\r\n");
 
-        private readonly IPipeReader _input;
-        private readonly IPipeWriter _output;
+        private readonly PipeReader _input;
+        private readonly PipeWriter _output;
         private readonly IHttpApplication<TContext> _application;
 
         public RequestHeaderDictionary RequestHeaders => _parser.RequestHeaders;
@@ -41,7 +41,7 @@ public partial class HttpConnection<TContext>
 
         private HttpRequestParser _parser = new HttpRequestParser();
 
-        public HttpConnection(IHttpApplication<TContext> application, IPipeReader input, IPipeWriter output)
+        public HttpConnection(IHttpApplication<TContext> application, PipeReader input, PipeWriter output)
         {
             _application = application;
             _input = input;
@@ -50,9 +50,9 @@ public HttpConnection(IHttpApplication<TContext> application, IPipeReader input,
             _responseBody = new HttpResponseStream<TContext>(this);
         }
 
-        public IPipeReader Input => _input;
+        public PipeReader Input => _input;
 
-        public IPipeWriter Output => _output;
+        public PipeWriter Output => _output;
 
         public HttpRequestStream<TContext> RequestBody { get; set; }
 
@@ -194,12 +194,12 @@ public Task WriteAsync(Span<byte> data)
             return FlushAsync(buffer);
         }
 
-        public async Task FlushAsync(IPipeWriter buffer)
+        public async Task FlushAsync(PipeWriter buffer)
         {
             await buffer.FlushAsync();
         }
 
-        private void WriteBeginResponseHeaders(IPipeWriter buffer)
+        private void WriteBeginResponseHeaders(PipeWriter buffer)
         {
             if (HasStarted)
             {
@@ -217,7 +217,7 @@ private void WriteBeginResponseHeaders(IPipeWriter buffer)
             ResponseHeaders.CopyTo(_autoChunk, buffer);
         }
 
-        private void WriteEndResponse(IPipeWriter buffer)
+        private void WriteEndResponse(PipeWriter buffer)
         {
             buffer.Write(_chunkedEndBytes);
         }

samples/System.IO.Pipelines.Samples/SampleBase/RawHttpClientSampleBase.cs

@@ -39,7 +39,7 @@ public async Task Run()
 
         protected abstract Task<IPipeConnection> GetConnection();
 
-        private async Task CopyCompletedAsync(IPipeReader input, IPipeWriter output)
+        private async Task CopyCompletedAsync(PipeReader input, PipeWriter output)
         {
             var result = await input.ReadAsync();
             var inputBuffer = result.Buffer;

samples/System.IO.Pipelines.Samples/SampleBase/RawHttpServerSampleBase.cs

@@ -65,7 +65,7 @@ protected async Task ProcessConnection(IPipeConnection connection)
 
                     // Writing directly to pooled buffers
                     var output = connection.Output;
-                    var formatter = new OutputFormatter<IPipeWriter>(output, SymbolTable.InvariantUtf8);
+                    var formatter = new OutputFormatter<PipeWriter>(output, SymbolTable.InvariantUtf8);
                     formatter.Append("HTTP/1.1 200 OK");
                     formatter.Append("\r\nContent-Length: 13");
                     formatter.Append("\r\nContent-Type: text/plain");

src/System.IO.Pipelines.Compression/CompressionPipelineExtensions.cs

@@ -10,82 +10,82 @@ namespace System.IO.Pipelines.Compression
 {
     public static class CompressionPipelineExtensions
     {
-        public static IPipeReader DeflateDecompress(this IPipeReader reader, PipeOptions options)
+        public static PipeReader DeflateDecompress(this PipeReader reader, PipeOptions options)
         {
             var inflater = new ReadableDeflateTransform(ZLibNative.Deflate_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = inflater.Execute(reader, pipe.Writer);
             return pipe.Reader;
         }
 
-        public static IPipeReader DeflateCompress(this IPipeReader reader, PipeOptions options, CompressionLevel compressionLevel)
+        public static PipeReader DeflateCompress(this PipeReader reader, PipeOptions options, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.Deflate_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = deflater.Execute(reader, pipe.Writer);
             return pipe.Reader;
         }
 
-        public static IPipeReader GZipDecompress(this IPipeReader reader, PipeOptions options)
+        public static PipeReader GZipDecompress(this PipeReader reader, PipeOptions options)
         {
             var inflater = new ReadableDeflateTransform(ZLibNative.GZip_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = inflater.Execute(reader, pipe.Writer);
             return pipe.Reader;
         }
 
-        public static IPipeWriter GZipCompress(this IPipeWriter writer, PipeOptions options, CompressionLevel compressionLevel)
+        public static PipeWriter GZipCompress(this PipeWriter writer, PipeOptions options, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = deflater.Execute(pipe.Reader, writer);
             return pipe.Writer;
         }
 
-        public static IPipeReader GZipCompress(this IPipeReader reader, PipeOptions options, CompressionLevel compressionLevel)
+        public static PipeReader GZipCompress(this PipeReader reader, PipeOptions options, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = deflater.Execute(reader, pipe.Writer);
             return pipe.Reader;
         }
 
-        public static IPipeReader CreateDeflateDecompressReader(PipeOptions options, IPipeReader reader)
+        public static PipeReader CreateDeflateDecompressReader(PipeOptions options, PipeReader reader)
         {
             var inflater = new ReadableDeflateTransform(ZLibNative.Deflate_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = inflater.Execute(reader, pipe.Writer);
             return pipe.Reader;
         }
 
-        public static IPipeReader CreateDeflateCompressReader(PipeOptions options, IPipeReader reader, CompressionLevel compressionLevel)
+        public static PipeReader CreateDeflateCompressReader(PipeOptions options, PipeReader reader, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.Deflate_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = deflater.Execute(reader, pipe.Writer);
             return pipe.Reader;
         }
 
-        public static IPipeReader CreateGZipDecompressReader(PipeOptions options, IPipeReader reader)
+        public static PipeReader CreateGZipDecompressReader(PipeOptions options, PipeReader reader)
         {
             var inflater = new ReadableDeflateTransform(ZLibNative.GZip_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = inflater.Execute(reader, pipe.Writer);
             return pipe.Reader;
         }
 
-        public static IPipeWriter CreateGZipCompressWriter(PipeOptions options, IPipeWriter writer, CompressionLevel compressionLevel)
+        public static PipeWriter CreateGZipCompressWriter(PipeOptions options, PipeWriter writer, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = deflater.Execute(pipe.Reader, writer);
             return pipe.Writer;
         }
 
-        public static IPipeReader CreateGZipCompressReader(PipeOptions options, IPipeReader reader, CompressionLevel compressionLevel)
+        public static PipeReader CreateGZipCompressReader(PipeOptions options, PipeReader reader, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits);
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = deflater.Execute(reader, pipe.Writer);
             return pipe.Reader;
         }
@@ -99,7 +99,7 @@ public WritableDeflateTransform(CompressionLevel compressionLevel, int bits)
                 _deflater = new Deflater(compressionLevel, bits);
             }
 
-            public async Task Execute(IPipeReader reader, IPipeWriter writer)
+            public async Task Execute(PipeReader reader, PipeWriter writer)
             {
                 List<MemoryHandle> handles = new List<MemoryHandle>();
 
@@ -210,7 +210,7 @@ public ReadableDeflateTransform(int bits)
                 _inflater = new Inflater(bits);
             }
 
-            public async Task Execute(IPipeReader reader, IPipeWriter writer)
+            public async Task Execute(PipeReader reader, PipeWriter writer)
             {
                 List<MemoryHandle> handles = new List<MemoryHandle>();
 

src/System.IO.Pipelines.Extensions/PipelineReaderExtensions.cs

@@ -9,7 +9,7 @@ namespace System.IO.Pipelines
 {
     public static class PipelineReaderExtensions
     {
-        public static ValueTask<int> ReadAsync(this IPipeReader input, ArraySegment<byte> destination)
+        public static ValueTask<int> ReadAsync(this PipeReader input, ArraySegment<byte> destination)
         {
             while (true)
             {
@@ -44,7 +44,7 @@ public static ValueTask<int> ReadAsync(this IPipeReader input, ArraySegment<byte
             return new ValueTask<int>(input.ReadAsyncAwaited(destination));
         }
 
-        public static async Task CopyToAsync(this IPipeReader input, IPipeWriter output)
+        public static async Task CopyToAsync(this PipeReader input, PipeWriter output)
         {
             while (true)
             {
@@ -75,7 +75,7 @@ public static async Task CopyToAsync(this IPipeReader input, IPipeWriter output)
             }
         }
 
-        private static async Task<int> ReadAsyncAwaited(this IPipeReader input, ArraySegment<byte> destination)
+        private static async Task<int> ReadAsyncAwaited(this PipeReader input, ArraySegment<byte> destination)
         {
             while (true)
             {
@@ -99,12 +99,12 @@ private static async Task<int> ReadAsyncAwaited(this IPipeReader input, ArraySeg
             }
         }
 
-        public static Task CopyToAsync(this IPipeReader input, Stream stream)
+        public static Task CopyToAsync(this PipeReader input, Stream stream)
         {
             return input.CopyToAsync(stream, 4096, CancellationToken.None);
         }
 
-        public static async Task CopyToAsync(this IPipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken)
+        public static async Task CopyToAsync(this PipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken)
         {
             // TODO: Use bufferSize argument
             while (!cancellationToken.IsCancellationRequested)

src/System.IO.Pipelines.Extensions/ReadWriteExtensions.cs

@@ -98,7 +98,7 @@ static void WriteLittleEndian<[Primitive]T>(this Span<byte> buffer, T value) whe
             WriteMachineEndian(buffer, ref value);
         }
 
-        public static async Task<ReadOnlyBuffer<byte>> ReadToEndAsync(this IPipeReader input)
+        public static async Task<ReadOnlyBuffer<byte>> ReadToEndAsync(this PipeReader input)
         {
             while (true)
             {
@@ -158,7 +158,7 @@ private static T ReadMultiLittle<[Primitive]T>(ReadOnlyBuffer<byte> buffer, int
         /// Reads a structure of type T out of a buffer of bytes.
         /// </summary>
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public static void WriteBigEndian<[Primitive]T>(this IPipeWriter buffer, T value) where T : struct
+        public static void WriteBigEndian<[Primitive]T>(this PipeWriter buffer, T value) where T : struct
         {
             int len = Unsafe.SizeOf<T>();
             buffer.GetMemory(len).Span.WriteBigEndian(value);
@@ -169,7 +169,7 @@ public static void WriteBigEndian<[Primitive]T>(this IPipeWriter buffer, T value
         /// Reads a structure of type T out of a buffer of bytes.
         /// </summary>
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public static void WriteLittleEndian<[Primitive]T>(this IPipeWriter buffer, T value) where T : struct
+        public static void WriteLittleEndian<[Primitive]T>(this PipeWriter buffer, T value) where T : struct
         {
             int len = Unsafe.SizeOf<T>();
             buffer.GetMemory(len).Span.WriteLittleEndian(value);

src/System.IO.Pipelines.Extensions/StreamExtensions.cs

@@ -11,19 +11,19 @@ namespace System.IO.Pipelines
     public static class StreamExtensions
     {
         /// <summary>
-        /// Copies the content of a <see cref="Stream"/> into a <see cref="IPipeWriter"/>.
+        /// Copies the content of a <see cref="Stream"/> into a <see cref="PipeWriter"/>.
         /// </summary>
         /// <param name="stream"></param>
         /// <param name="writer"></param>
         /// <param name="cancellationToken"></param>
         /// <returns></returns>
-        public static Task CopyToAsync(this Stream stream, IPipeWriter writer, CancellationToken cancellationToken = default)
+        public static Task CopyToAsync(this Stream stream, PipeWriter writer, CancellationToken cancellationToken = default)
         {
             // 81920 is the default bufferSize, there is not stream.CopyToAsync overload that takes only a cancellationToken
             return stream.CopyToAsync(new PipelineWriterStream(writer), bufferSize: 81920, cancellationToken: cancellationToken);
         }
 
-        public static async Task CopyToEndAsync(this Stream stream, IPipeWriter writer, CancellationToken cancellationToken = default)
+        public static async Task CopyToEndAsync(this Stream stream, PipeWriter writer, CancellationToken cancellationToken = default)
         {
             try
             {
@@ -77,12 +77,12 @@ await stream.WriteAsync(data.Array, data.Offset, data.Count)
             }
         }
 
-        public static Task CopyToEndAsync(this IPipeReader input, Stream stream)
+        public static Task CopyToEndAsync(this PipeReader input, Stream stream)
         {
             return input.CopyToEndAsync(stream, 4096, CancellationToken.None);
         }
 
-        public static async Task CopyToEndAsync(this IPipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken)
+        public static async Task CopyToEndAsync(this PipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken)
         {
             try
             {
@@ -98,9 +98,9 @@ public static async Task CopyToEndAsync(this IPipeReader input, Stream stream, i
 
         private class PipelineWriterStream : Stream
         {
-            private readonly IPipeWriter _writer;
+            private readonly PipeWriter _writer;
 
-            public PipelineWriterStream(IPipeWriter writer)
+            public PipelineWriterStream(PipeWriter writer)
             {
                 _writer = writer;
             }

src/System.IO.Pipelines.Extensions/StreamPipeConnection.cs

@@ -11,37 +11,37 @@ public StreamPipeConnection(PipeOptions options, Stream stream)
             Output = CreateWriter(options, stream);
         }
 
-        public IPipeReader Input { get; }
+        public PipeReader Input { get; }
 
-        public IPipeWriter Output { get; }
+        public PipeWriter Output { get; }
 
         public void Dispose()
         {
             Input.Complete();
             Output.Complete();
         }
 
-        public static IPipeReader CreateReader(PipeOptions options, Stream stream)
+        public static PipeReader CreateReader(PipeOptions options, Stream stream)
         {
             if (!stream.CanRead)
             {
                 throw new NotSupportedException();
             }
 
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = stream.CopyToEndAsync(pipe.Writer);
 
             return pipe.Reader;
         }
 
-        public static IPipeWriter CreateWriter(PipeOptions options, Stream stream)
+        public static PipeWriter CreateWriter(PipeOptions options, Stream stream)
         {
             if (!stream.CanWrite)
             {
                 throw new NotSupportedException();
             }
 
-            var pipe = new Pipe(options);
+            var pipe = new ResetablePipe(options);
             var ignore = pipe.Reader.CopyToEndAsync(stream);
 
             return pipe.Writer;