Add IPipe interface, Rename IPipeline* to IPipe* (#1182)

* Add IPipe interface, Rename IPipeline* to IPipe*
* Fix all build errors
* move ReadingStarted to pipe
* Do state changes after argument validation

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

@@ -10,7 +10,7 @@ public class CompressionSample
     {
         public static void Run()
         {
-            using (var factory = new PipelineFactory())
+            using (var factory = new PipeFactory())
             {
                 var filePath = Path.GetFullPath("Program.cs");
 

samples/System.IO.Pipelines.Samples/Framing/Codec.cs

@@ -81,7 +81,7 @@ public static void Run()
             thread.Dispose();
         }
 
-        public static IPipelineConnection MakePipeline(IPipelineConnection connection)
+        public static IPipeConnection MakePipeline(IPipeConnection connection)
         {
             // Do something fancy here to wrap the connection, SSL etc
             return connection;
@@ -97,7 +97,7 @@ public class LineHandler : IFrameHandler<Line>
     {
         private PipelineTextOutput _textOutput;
 
-        public void Initialize(IPipelineConnection connection)
+        public void Initialize(IPipeConnection connection)
         {
             _textOutput = new PipelineTextOutput(connection.Output, EncodingData.InvariantUtf8);
         }
@@ -135,7 +135,7 @@ public interface IFrameDecoder<TInput>
 
     public interface IFrameHandler<TInput>
     {
-        void Initialize(IPipelineConnection connection);
+        void Initialize(IPipeConnection connection);
 
         Task HandleAsync(TInput message);
     }

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

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

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

@@ -14,8 +14,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 IPipelineReader _input;
-        private readonly IPipelineWriter _output;
+        private readonly IPipeReader _input;
+        private readonly IPipeWriter _output;
         private readonly IHttpApplication<TContext> _application;
 
         public RequestHeaderDictionary RequestHeaders => _parser.RequestHeaders;
@@ -38,7 +38,7 @@ public partial class HttpConnection<TContext>
 
         private HttpRequestParser _parser = new HttpRequestParser();
 
-        public HttpConnection(IHttpApplication<TContext> application, IPipelineReader input, IPipelineWriter output)
+        public HttpConnection(IHttpApplication<TContext> application, IPipeReader input, IPipeWriter output)
         {
             _application = application;
             _input = input;
@@ -47,9 +47,9 @@ public HttpConnection(IHttpApplication<TContext> application, IPipelineReader in
             _responseBody = new HttpResponseStream<TContext>(this);
         }
 
-        public IPipelineReader Input => _input;
+        public IPipeReader Input => _input;
 
-        public IPipelineWriter Output => _output;
+        public IPipeWriter Output => _output;
 
         public HttpRequestStream<TContext> RequestBody { get; set; }
 

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

@@ -91,7 +91,7 @@ private async void StartAcceptingConnections<TContext>(IHttpApplication<TContext
             _listenSocket.Bind(new IPEndPoint(ip, port));
             _listenSocket.Listen(10);
 
-            using (var factory = new PipelineFactory())
+            using (var factory = new PipeFactory())
             {
                 while (true)
                 {
@@ -154,20 +154,20 @@ private static async Task ProcessRIOConnection<TContext>(IHttpApplication<TConte
             }
         }
 
-        private static async Task ProcessConnection<TContext>(IHttpApplication<TContext> application, PipelineFactory pipelineFactory, Socket socket)
+        private static async Task ProcessConnection<TContext>(IHttpApplication<TContext> application, PipeFactory pipeFactory, Socket socket)
         {
             using (var ns = new NetworkStream(socket))
             {
-                using (var connection = pipelineFactory.CreateConnection(ns))
+                using (var connection = pipeFactory.CreateConnection(ns))
                 {
                     await ProcessClient(application, connection);
                 }
             }
         }
 
-        private static async Task ProcessClient<TContext>(IHttpApplication<TContext> application, IPipelineConnection pipelineConnection)
+        private static async Task ProcessClient<TContext>(IHttpApplication<TContext> application, IPipeConnection pipeConnection)
         {
-            var connection = new HttpConnection<TContext>(application, pipelineConnection.Input, pipelineConnection.Output);
+            var connection = new HttpConnection<TContext>(application, pipeConnection.Input, pipeConnection.Output);
 
             await connection.ProcessAllRequests();
         }

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

@@ -17,7 +17,7 @@ public static async Task Run()
             var thread = new UvThread();
             var client = new UvTcpClient(thread, new IPEndPoint(IPAddress.Loopback, 5000));
 
-            var consoleOutput = thread.PipelineFactory.CreateWriter(Console.OpenStandardOutput());
+            var consoleOutput = thread.PipeFactory.CreateWriter(Console.OpenStandardOutput());
 
             var connection = await client.ConnectAsync();
 
@@ -36,7 +36,7 @@ public static async Task Run()
                 await Task.Delay(1000);
             }
         }
-        private static async Task CopyCompletedAsync(IPipelineReader input, IPipelineWriter output)
+        private static async Task CopyCompletedAsync(IPipeReader input, IPipeWriter output)
         {
             var result = await input.ReadAsync();
             var inputBuffer = result.Buffer;

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

@@ -9,61 +9,61 @@ namespace System.IO.Pipelines.Compression
 {
     public static class CompressionPipelineExtensions
     {
-        public static IPipelineReader DeflateDecompress(this IPipelineReader reader, PipelineFactory factory)
+        public static IPipeReader DeflateDecompress(this IPipeReader reader, PipeFactory factory)
         {
             var inflater = new ReadableDeflateTransform(ZLibNative.Deflate_DefaultWindowBits);
             return factory.CreateReader(reader, inflater.Execute);
         }
 
-        public static IPipelineReader DeflateCompress(this IPipelineReader reader, PipelineFactory factory, CompressionLevel compressionLevel)
+        public static IPipeReader DeflateCompress(this IPipeReader reader, PipeFactory factory, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.Deflate_DefaultWindowBits);
             return factory.CreateReader(reader, deflater.Execute);
         }
 
-        public static IPipelineReader GZipDecompress(this IPipelineReader reader, PipelineFactory factory)
+        public static IPipeReader GZipDecompress(this IPipeReader reader, PipeFactory factory)
         {
             var inflater = new ReadableDeflateTransform(ZLibNative.GZip_DefaultWindowBits);
             return factory.CreateReader(reader, inflater.Execute);
         }
 
-        public static IPipelineWriter GZipCompress(this IPipelineWriter writer, PipelineFactory factory, CompressionLevel compressionLevel)
+        public static IPipeWriter GZipCompress(this IPipeWriter writer, PipeFactory factory, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits);
             return factory.CreateWriter(writer, deflater.Execute);
         }
 
-        public static IPipelineReader GZipCompress(this IPipelineReader reader, PipelineFactory factory, CompressionLevel compressionLevel)
+        public static IPipeReader GZipCompress(this IPipeReader reader, PipeFactory factory, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits);
             return factory.CreateReader(reader, deflater.Execute);
         }
 
-        public static IPipelineReader CreateDeflateDecompressReader(this PipelineFactory factory, IPipelineReader reader)
+        public static IPipeReader CreateDeflateDecompressReader(this PipeFactory factory, IPipeReader reader)
         {
             var inflater = new ReadableDeflateTransform(ZLibNative.Deflate_DefaultWindowBits);
             return factory.CreateReader(reader, inflater.Execute);
         }
 
-        public static IPipelineReader CreateDeflateCompressReader(this PipelineFactory factory, IPipelineReader reader, CompressionLevel compressionLevel)
+        public static IPipeReader CreateDeflateCompressReader(this PipeFactory factory, IPipeReader reader, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.Deflate_DefaultWindowBits);
             return factory.CreateReader(reader, deflater.Execute);
         }
 
-        public static IPipelineReader CreateGZipDecompressReader(this PipelineFactory factory, IPipelineReader reader)
+        public static IPipeReader CreateGZipDecompressReader(this PipeFactory factory, IPipeReader reader)
         {
             var inflater = new ReadableDeflateTransform(ZLibNative.GZip_DefaultWindowBits);
             return factory.CreateReader(reader, inflater.Execute);
         }
 
-        public static IPipelineWriter CreateGZipCompressWriter(this PipelineFactory factory, IPipelineWriter writer, CompressionLevel compressionLevel)
+        public static IPipeWriter CreateGZipCompressWriter(this PipeFactory factory, IPipeWriter writer, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits);
             return factory.CreateWriter(writer, deflater.Execute);
         }
 
-        public static IPipelineReader CreateGZipCompressReader(this PipelineFactory factory, IPipelineReader reader, CompressionLevel compressionLevel)
+        public static IPipeReader CreateGZipCompressReader(this PipeFactory factory, IPipeReader reader, CompressionLevel compressionLevel)
         {
             var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits);
             return factory.CreateReader(reader, deflater.Execute);
@@ -78,7 +78,7 @@ public WritableDeflateTransform(CompressionLevel compressionLevel, int bits)
                 _deflater = new Deflater(compressionLevel, bits);
             }
 
-            public async Task Execute(IPipelineReader reader, IPipelineWriter writer)
+            public async Task Execute(IPipeReader reader, IPipeWriter writer)
             {
                 while (true)
                 {
@@ -207,7 +207,7 @@ public ReadableDeflateTransform(int bits)
                 _inflater = new Inflater(bits);
             }
 
-            public async Task Execute(IPipelineReader reader, IPipelineWriter writer)
+            public async Task Execute(IPipeReader reader, IPipeWriter writer)
             {
                 while (true)
                 {

src/System.IO.Pipelines.File/FileReader.cs

@@ -9,13 +9,13 @@
 
 namespace System.IO.Pipelines.File
 {
-    public class FileReader : PipelineReader
+    public class FileReader : PipeReader
     {
         public FileReader(MemoryPool pool) : base(pool)
         {
         }
 
-        public FileReader(Pipe pipe) : base(pipe)
+        public FileReader(IPipe pipe) : base(pipe)
         {
         }
 
@@ -29,7 +29,7 @@ public unsafe void OpenReadFile(string path)
 
             var readOperation = new ReadOperation
             {
-                Writer = _pipe,
+                Writer = Pipe.Writer,
                 FileHandle = fileHandle,
                 ThreadPoolBoundHandle = handle,
                 IOCallback = IOCallback
@@ -38,7 +38,7 @@ public unsafe void OpenReadFile(string path)
             var overlapped = new PreAllocatedOverlapped(IOCallback, readOperation, null);
             readOperation.PreAllocatedOverlapped = overlapped;
 
-            _pipe.ReadingStarted.ContinueWith((t, state) =>
+            Pipe.ReadingStarted.ContinueWith((t, state) =>
             {
                 ((ReadOperation)state).Read();
             },
@@ -103,7 +103,7 @@ private class ReadOperation
 
             public unsafe NativeOverlapped* Overlapped { get; set; }
 
-            public IPipelineWriter Writer { get; set; }
+            public IPipeWriter Writer { get; set; }
 
             public StrongBox<WritableBuffer> BoxedBuffer { get; set; }
 

src/System.IO.Pipelines.File/ReadableFilePipelineFactoryExtensions.cs

@@ -10,7 +10,7 @@ namespace System.IO.Pipelines.File
 {
     public static class ReadableFilePipelineFactoryExtensions
     {
-        public static IPipelineReader ReadFile(this PipelineFactory factory, string path)
+        public static IPipeReader ReadFile(this PipeFactory factory, string path)
         {
             var pipe = factory.Create();