Implement PipeBody Features and add to HttpContext (#6394)

src/Http/Http.Abstractions/src/HttpRequest.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System.IO;
+using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Routing;
@@ -102,6 +103,11 @@ namespace Microsoft.AspNetCore.Http
         /// <returns>The RequestBody Stream.</returns>
         public abstract Stream Body { get; set; }
 
+        /// <summary>
+        /// Gets or sets the request body pipe <see cref="PipeReader"/>.
+        /// </summary>
+        public abstract PipeReader BodyPipe { get; set; }
+
         /// <summary>
         /// Checks the Content-Type header for form types.
         /// </summary>

src/Http/Http.Abstractions/src/HttpResponse.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.IO;
+using System.IO.Pipelines;
 using System.Threading.Tasks;
 
 namespace Microsoft.AspNetCore.Http
@@ -39,6 +40,11 @@ namespace Microsoft.AspNetCore.Http
         /// </summary>
         public abstract Stream Body { get; set; }
 
+        /// <summary>
+        /// Gets or sets the response body pipe <see cref="PipeWriter"/>
+        /// </summary>
+        public abstract PipeWriter BodyPipe { get; set; }
+
         /// <summary>
         /// Gets or sets the value for the <c>Content-Length</c> response header.
         /// </summary>

src/Http/Http.Abstractions/src/Microsoft.AspNetCore.Http.Abstractions.csproj

@@ -22,6 +22,7 @@ Microsoft.AspNetCore.Http.HttpResponse</Description>
     <Reference Include="Microsoft.AspNetCore.Http.Features" />
     <Reference Include="Microsoft.Extensions.ActivatorUtilities.Sources" PrivateAssets="All" />
     <Reference Include="System.Text.Encodings.Web" />
+    <Reference Include="System.IO.Pipelines" />
   </ItemGroup>
 
 </Project>

src/Http/Http.Features/src/IRequestBodyPipeFeature.cs

@@ -0,0 +1,18 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System.IO.Pipelines;
+
+namespace Microsoft.AspNetCore.Http.Features
+{
+    /// <summary>
+    /// Represents the HttpRequestBody as a PipeReader.
+    /// </summary>
+    public interface IRequestBodyPipeFeature
+    {
+        /// <summary>
+        /// A <see cref="PipeWriter"/> representing the request body, if any.
+        /// </summary>
+        PipeReader RequestBodyPipe { get; set; }
+    }
+}

src/Http/Http.Features/src/IResponseBodyPipeFeature.cs

@@ -0,0 +1,19 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.IO.Pipelines;
+
+namespace Microsoft.AspNetCore.Http.Features
+{
+    /// <summary>
+    ///     Represents the HttpResponseBody as a PipeWriter
+    /// </summary>
+    public interface IResponseBodyPipeFeature
+    {
+        /// <summary>
+        /// A <see cref="PipeWriter"/> representing the response body, if any.
+        /// </summary>
+        PipeWriter ResponseBodyPipe { get; set; }
+    }
+}

src/Http/Http.Features/src/Microsoft.AspNetCore.Http.Features.csproj

@@ -11,6 +11,7 @@
 
   <ItemGroup>
     <Reference Include="Microsoft.Extensions.Primitives" />
+    <Reference Include="System.IO.Pipelines" />
   </ItemGroup>
 
 </Project>

src/Http/Http/src/Features/RequestBodyPipeFeature.cs

@@ -0,0 +1,49 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.IO.Pipelines;
+
+namespace Microsoft.AspNetCore.Http.Features
+{
+    public class RequestBodyPipeFeature : IRequestBodyPipeFeature
+    {
+        private StreamPipeReader _internalPipeReader;
+        private PipeReader _userSetPipeReader;
+        private HttpContext _context;
+
+        public RequestBodyPipeFeature(HttpContext context)
+        {
+            if (context == null)
+            {
+                throw new ArgumentNullException(nameof(context));
+            }
+            _context = context;
+        }
+
+        public PipeReader RequestBodyPipe
+        {
+            get
+            {
+                if (_userSetPipeReader != null)
+                {
+                    return _userSetPipeReader;
+                }
+
+                if (_internalPipeReader == null ||
+                    !object.ReferenceEquals(_internalPipeReader.InnerStream, _context.Request.Body))
+                {
+                    _internalPipeReader = new StreamPipeReader(_context.Request.Body);
+                    _context.Response.RegisterForDispose(_internalPipeReader);
+                }
+
+                return _internalPipeReader;
+            }
+            set
+            {
+                _userSetPipeReader = value ?? throw new ArgumentNullException(nameof(value));
+                // TODO set the request body Stream to an adapted pipe https://github.com/aspnet/AspNetCore/issues/3971
+            }
+        }
+    }
+}

src/Http/Http/src/Features/ResponseBodyPipeFeature.cs

@@ -0,0 +1,49 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.IO.Pipelines;
+
+namespace Microsoft.AspNetCore.Http.Features
+{
+    public class ResponseBodyPipeFeature : IResponseBodyPipeFeature
+    {
+        private StreamPipeWriter _internalPipeWriter;
+        private PipeWriter _userSetPipeWriter;
+        private HttpContext _context;
+
+        public ResponseBodyPipeFeature(HttpContext context)
+        {
+            if (context == null)
+            {
+                throw new ArgumentNullException(nameof(context));
+            }
+            _context = context;
+        }
+
+        public PipeWriter ResponseBodyPipe
+        {
+            get
+            {
+                if (_userSetPipeWriter != null)
+                {
+                    return _userSetPipeWriter;
+                }
+
+                if (_internalPipeWriter == null ||
+                    !object.ReferenceEquals(_internalPipeWriter.InnerStream, _context.Response.Body))
+                {
+                    _internalPipeWriter = new StreamPipeWriter(_context.Response.Body);
+                    _context.Response.RegisterForDispose(_internalPipeWriter);
+                }
+
+                return _internalPipeWriter;
+            }
+            set
+            {
+                _userSetPipeWriter = value ?? throw new ArgumentNullException(nameof(value));
+                // TODO set the response body Stream to an adapted pipe https://github.com/aspnet/AspNetCore/issues/3971
+            }
+        }
+    }
+}

src/Http/Http/src/Internal/DefaultHttpRequest.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.IO;
+using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http.Features;
@@ -19,6 +20,7 @@ namespace Microsoft.AspNetCore.Http.Internal
         private readonly static Func<HttpRequest, IFormFeature> _newFormFeature = r => new FormFeature(r);
         private readonly static Func<IFeatureCollection, IRequestCookiesFeature> _newRequestCookiesFeature = f => new RequestCookiesFeature(f);
         private readonly static Func<IFeatureCollection, IRouteValuesFeature> _newRouteValuesFeature = f => new RouteValuesFeature();
+        private readonly static Func<HttpContext, IRequestBodyPipeFeature> _newRequestBodyPipeFeature = context => new RequestBodyPipeFeature(context);
 
         private HttpContext _context;
         private FeatureReferences<FeatureInterfaces> _features;
@@ -57,6 +59,9 @@ namespace Microsoft.AspNetCore.Http.Internal
         private IRouteValuesFeature RouteValuesFeature =>
             _features.Fetch(ref _features.Cache.RouteValues, _newRouteValuesFeature);
 
+        private IRequestBodyPipeFeature RequestBodyPipeFeature =>
+            _features.Fetch(ref _features.Cache.BodyPipe, this.HttpContext, _newRequestBodyPipeFeature);
+
         public override PathString PathBase
         {
             get { return new PathString(HttpRequestFeature.PathBase); }
@@ -162,6 +167,12 @@ namespace Microsoft.AspNetCore.Http.Internal
             set { RouteValuesFeature.RouteValues = value; }
         }
 
+        public override PipeReader BodyPipe
+        {
+            get { return RequestBodyPipeFeature.RequestBodyPipe; }
+            set { RequestBodyPipeFeature.RequestBodyPipe = value; }
+        }
+
         struct FeatureInterfaces
         {
             public IHttpRequestFeature Request;
@@ -169,6 +180,7 @@ namespace Microsoft.AspNetCore.Http.Internal
             public IFormFeature Form;
             public IRequestCookiesFeature Cookies;
             public IRouteValuesFeature RouteValues;
+            public IRequestBodyPipeFeature BodyPipe;
         }
     }
 }

src/Http/Http/src/Internal/DefaultHttpResponse.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.IO;
+using System.IO.Pipelines;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.Net.Http.Headers;
@@ -14,6 +15,7 @@ namespace Microsoft.AspNetCore.Http.Internal
         // Lambdas hoisted to static readonly fields to improve inlining https://github.com/dotnet/roslyn/issues/13624
         private readonly static Func<IFeatureCollection, IHttpResponseFeature> _nullResponseFeature = f => null;
         private readonly static Func<IFeatureCollection, IResponseCookiesFeature> _newResponseCookiesFeature = f => new ResponseCookiesFeature(f);
+        private readonly static Func<HttpContext, IResponseBodyPipeFeature> _newResponseBodyPipeFeature = context => new ResponseBodyPipeFeature(context);
 
         private HttpContext _context;
         private FeatureReferences<FeatureInterfaces> _features;
@@ -41,6 +43,8 @@ namespace Microsoft.AspNetCore.Http.Internal
         private IResponseCookiesFeature ResponseCookiesFeature =>
             _features.Fetch(ref _features.Cache.Cookies, _newResponseCookiesFeature);
 
+        private IResponseBodyPipeFeature ResponseBodyPipeFeature =>
+            _features.Fetch(ref _features.Cache.BodyPipe, this.HttpContext, _newResponseBodyPipeFeature);
 
         public override HttpContext HttpContext { get { return _context; } }
 
@@ -96,6 +100,12 @@ namespace Microsoft.AspNetCore.Http.Internal
             get { return HttpResponseFeature.HasStarted; }
         }
 
+        public override PipeWriter BodyPipe
+        {
+            get { return ResponseBodyPipeFeature.ResponseBodyPipe; }
+            set { ResponseBodyPipeFeature.ResponseBodyPipe = value; }
+        }
+
         public override void OnStarting(Func<object, Task> callback, object state)
         {
             if (callback == null)
@@ -134,6 +144,7 @@ namespace Microsoft.AspNetCore.Http.Internal
         {
             public IHttpResponseFeature Response;
             public IResponseCookiesFeature Cookies;
+            public IResponseBodyPipeFeature BodyPipe;
         }
     }
 }

src/Http/Http/src/Internal/ReusableHttpRequest.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.IO.Pipelines;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -18,6 +19,8 @@ namespace Microsoft.AspNetCore.Http.Internal
         private readonly static Func<HttpRequest, IFormFeature> _newFormFeature = r => new FormFeature(r);
         private readonly static Func<IFeatureCollection, IRequestCookiesFeature> _newRequestCookiesFeature = f => new RequestCookiesFeature(f);
         private readonly static Func<IFeatureCollection, IRouteValuesFeature> _newRouteValuesFeature = f => new RouteValuesFeature();
+        private readonly static Func<HttpContext, IRequestBodyPipeFeature> _newRequestBodyPipeFeature = context => new RequestBodyPipeFeature(context);
+
 
         private HttpContext _context;
         private FeatureReferences<FeatureInterfaces> _features;
@@ -56,6 +59,9 @@ namespace Microsoft.AspNetCore.Http.Internal
         private IRouteValuesFeature RouteValuesFeature =>
             _features.Fetch(ref _features.Cache.RouteValues, _newRouteValuesFeature);
 
+        private IRequestBodyPipeFeature RequestBodyPipeFeature =>
+            _features.Fetch(ref _features.Cache.BodyPipe, this.HttpContext, _newRequestBodyPipeFeature);
+
         public override PathString PathBase
         {
             get { return new PathString(HttpRequestFeature.PathBase); }
@@ -161,6 +167,12 @@ namespace Microsoft.AspNetCore.Http.Internal
             set { RouteValuesFeature.RouteValues = value; }
         }
 
+        public override PipeReader BodyPipe
+        {
+            get { return RequestBodyPipeFeature.RequestBodyPipe; }
+            set { RequestBodyPipeFeature.RequestBodyPipe = value; }
+        }
+
         struct FeatureInterfaces
         {
             public IHttpRequestFeature Request;
@@ -168,6 +180,7 @@ namespace Microsoft.AspNetCore.Http.Internal
             public IFormFeature Form;
             public IRequestCookiesFeature Cookies;
             public IRouteValuesFeature RouteValues;
+            public IRequestBodyPipeFeature BodyPipe;
         }
     }
 }

src/Http/Http/src/Internal/ReusableHttpResponse.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.IO.Pipelines;
 using System.Text;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http.Features;
@@ -13,6 +14,7 @@ namespace Microsoft.AspNetCore.Http.Internal
         // Lambdas hoisted to static readonly fields to improve inlining https://github.com/dotnet/roslyn/issues/13624
         private readonly static Func<IFeatureCollection, IHttpResponseFeature> _nullResponseFeature = f => null;
         private readonly static Func<IFeatureCollection, IResponseCookiesFeature> _newResponseCookiesFeature = f => new ResponseCookiesFeature(f);
+        private readonly static Func<HttpContext, IResponseBodyPipeFeature> _newResponseBodyPipeFeature = context => new ResponseBodyPipeFeature(context);
 
         private HttpContext _context;
         private FeatureReferences<FeatureInterfaces> _features;
@@ -39,7 +41,8 @@ namespace Microsoft.AspNetCore.Http.Internal
 
         private IResponseCookiesFeature ResponseCookiesFeature =>
             _features.Fetch(ref _features.Cache.Cookies, _newResponseCookiesFeature);
-
+        private IResponseBodyPipeFeature ResponseBodyPipeFeature =>
+            _features.Fetch(ref _features.Cache.BodyPipe, this.HttpContext, _newResponseBodyPipeFeature);
 
         public override HttpContext HttpContext { get { return _context; } }
 
@@ -90,6 +93,12 @@ namespace Microsoft.AspNetCore.Http.Internal
             get { return ResponseCookiesFeature.Cookies; }
         }
 
+        public override PipeWriter BodyPipe
+        {
+            get { return ResponseBodyPipeFeature.ResponseBodyPipe; }
+            set { ResponseBodyPipeFeature.ResponseBodyPipe = value; }
+        }
+
         public override bool HasStarted
         {
             get { return HttpResponseFeature.HasStarted; }
@@ -133,6 +142,7 @@ namespace Microsoft.AspNetCore.Http.Internal
         {
             public IHttpResponseFeature Response;
             public IResponseCookiesFeature Cookies;
+            public IResponseBodyPipeFeature BodyPipe;
         }
     }
 }

src/Http/Http/src/StreamPipeReader.cs

@@ -17,7 +17,7 @@ namespace System.IO.Pipelines
     /// <summary>
     /// Implements PipeReader using an underlying stream.
     /// </summary>
-    public class StreamPipeReader : PipeReader
+    public class StreamPipeReader : PipeReader, IDisposable
     {
         private readonly int _minimumSegmentSize;
         private readonly int _minimumReadThreshold;
@@ -70,6 +70,11 @@ namespace System.IO.Pipelines
             _pool = options.MemoryPool;
         }
 
+        /// <summary>
+        /// Gets the inner stream that is being read from.
+        /// </summary>
+        public Stream InnerStream => _readingStream;
+
         /// <inheritdoc />
         public override void AdvanceTo(SequencePosition consumed)
         {

src/Http/Http/src/StreamPipeWriter.cs

@@ -64,6 +64,11 @@ namespace System.IO.Pipelines
             _pool = pool ?? MemoryPool<byte>.Shared;
         }
 
+        /// <summary>
+        /// Gets the inner stream that is being read from.
+        /// </summary>
+        public Stream InnerStream => _writingStream;
+
         /// <inheritdoc />
         public override void Advance(int count)
         {

src/Http/Http/test/Features/RequestBodyPipeFeatureTests.cs

@@ -0,0 +1,134 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System;
+using System.Buffers;
+using System.IO;
+using System.IO.Pipelines;
+using System.Text;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Microsoft.AspNetCore.Http.Features
+{
+    public class RequestBodyPipeFeatureTests
+    {
+        [Fact]
+        public void RequestBodyReturnsStreamPipeReader()
+        {
+            var context = new DefaultHttpContext();
+            var expectedStream = new MemoryStream();
+            context.Request.Body = expectedStream;
+
+            var provider = new RequestBodyPipeFeature(context);
+
+            var pipeBody = provider.RequestBodyPipe;
+
+            Assert.True(pipeBody is StreamPipeReader);
+            Assert.Equal(expectedStream, (pipeBody as StreamPipeReader).InnerStream);
+        }
+
+        [Fact]
+        public async Task RequestBodyReadCanWorkWithPipe()
+        {
+            var expectedString = "abcdef";
+            var provider = InitializeFeatureWithData(expectedString);
+
+            var data = await provider.RequestBodyPipe.ReadAsync();
+            Assert.Equal(expectedString, GetStringFromReadResult(data));
+        }
+
+        [Fact]
+        public void RequestBodySetPipeReaderReturnsSameValue()
+        {
+            var context = new DefaultHttpContext();
+
+            var provider = new RequestBodyPipeFeature(context);
+
+            var pipeReader = new Pipe().Reader;
+            provider.RequestBodyPipe = pipeReader;
+
+            Assert.Equal(pipeReader, provider.RequestBodyPipe);
+        }
+
+        [Fact]
+        public void RequestBodySetPipeReadReturnsUserSetValueAlways()
+        {
+            var context = new DefaultHttpContext();
+
+            var provider = new RequestBodyPipeFeature(context);
+
+            var expectedPipeReader = new Pipe().Reader;
+            provider.RequestBodyPipe = expectedPipeReader;
+
+            // Because the user set the RequestBodyPipe, this will return the user set pipeReader
+            context.Request.Body = new MemoryStream();
+
+            Assert.Equal(expectedPipeReader, provider.RequestBodyPipe);
+        }
+
+        [Fact]
+        public async Task RequestBodyDoesNotAffectUserSetPipe()
+        {
+            var expectedString = "abcdef";
+            var provider = InitializeFeatureWithData("hahaha");
+            provider.RequestBodyPipe = await GetPipeReaderWithData(expectedString);
+
+            var data = await provider.RequestBodyPipe.ReadAsync();
+            Assert.Equal(expectedString, GetStringFromReadResult(data));
+        }
+
+        [Fact]
+        public void RequestBodyGetPipeReaderAfterSettingBodyTwice()
+        {
+            var context = new DefaultHttpContext();
+
+            context.Request.Body = new MemoryStream();
+
+            var provider = new RequestBodyPipeFeature(context);
+
+            var pipeBody = provider.RequestBodyPipe;
+
+            // Requery the PipeReader after setting the body again.
+            var expectedStream = new MemoryStream();
+            context.Request.Body = expectedStream;
+            pipeBody = provider.RequestBodyPipe;
+
+            Assert.True(pipeBody is StreamPipeReader);
+            Assert.Equal(expectedStream, (pipeBody as StreamPipeReader).InnerStream);
+        }
+
+        [Fact]
+        public async Task RequestBodyGetsDataFromSecondStream()
+        {
+            var context = new DefaultHttpContext();
+            context.Request.Body = new MemoryStream(Encoding.ASCII.GetBytes("hahaha"));
+            var provider = new RequestBodyPipeFeature(context);
+            var _ = provider.RequestBodyPipe;
+
+            var expectedString = "abcdef";
+            context.Request.Body = new MemoryStream(Encoding.ASCII.GetBytes(expectedString));
+            var data = await provider.RequestBodyPipe.ReadAsync();
+            Assert.Equal(expectedString, GetStringFromReadResult(data));
+        }
+
+        private RequestBodyPipeFeature InitializeFeatureWithData(string input)
+        {
+            var context = new DefaultHttpContext();
+            context.Request.Body = new MemoryStream(Encoding.ASCII.GetBytes(input));
+            return new RequestBodyPipeFeature(context);
+        }
+
+        private static string GetStringFromReadResult(ReadResult data)
+        {
+            return Encoding.ASCII.GetString(data.Buffer.ToArray());
+        }
+
+        private async Task<PipeReader> GetPipeReaderWithData(string input)
+        {
+            var pipe = new Pipe();
+            await pipe.Writer.WriteAsync(Encoding.ASCII.GetBytes(input));
+            return pipe.Reader;
+        }
+    }
+}

src/Http/Http/test/Features/ResponseBodyPipeFeatureTests.cs

@@ -0,0 +1,56 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using System.IO;
+using System.IO.Pipelines;
+using Xunit;
+
+namespace Microsoft.AspNetCore.Http.Features
+{
+    public class ResponseBodyPipeFeatureTests
+    {
+        [Fact]
+        public void ResponseBodyReturnsStreamPipeReader()
+        {
+            var context = new DefaultHttpContext();
+            var expectedStream = new MemoryStream();
+            context.Response.Body = expectedStream;
+
+            var provider = new ResponseBodyPipeFeature(context);
+
+            var pipeBody = provider.ResponseBodyPipe;
+
+            Assert.True(pipeBody is StreamPipeWriter);
+            Assert.Equal(expectedStream, (pipeBody as StreamPipeWriter).InnerStream);
+        }
+
+        [Fact]
+        public void ResponseBodySetPipeReaderReturnsSameValue()
+        {
+            var context = new DefaultHttpContext();
+            var provider = new ResponseBodyPipeFeature(context);
+
+            var pipeWriter = new Pipe().Writer;
+            provider.ResponseBodyPipe = pipeWriter;
+
+            Assert.Equal(pipeWriter, provider.ResponseBodyPipe);
+        }
+
+        [Fact]
+        public void ResponseBodyGetPipeWriterAfterSettingBodyTwice()
+        {
+            var context = new DefaultHttpContext();
+            var expectedStream = new MemoryStream();
+            context.Response.Body = new MemoryStream();
+
+            var provider = new ResponseBodyPipeFeature(context);
+
+            var pipeBody = provider.ResponseBodyPipe;
+            context.Response.Body = expectedStream;
+            pipeBody = provider.ResponseBodyPipe;
+
+            Assert.True(pipeBody is StreamPipeWriter);
+            Assert.Equal(expectedStream, (pipeBody as StreamPipeWriter).InnerStream);
+        }
+    }
+}

src/Http/Http/test/Internal/DefaultHttpRequestTests.cs

@@ -4,6 +4,8 @@
 using System;
 using System.Collections.Generic;
 using System.Globalization;
+using System.IO;
+using System.IO.Pipelines;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Routing;
 using Microsoft.Extensions.Primitives;
@@ -241,6 +243,44 @@ namespace Microsoft.AspNetCore.Http.Internal
             Assert.Empty(request.RouteValues);
         }
 
+        [Fact]
+        public void BodyPipe_CanGet()
+        {
+            var context = new DefaultHttpContext();
+            var bodyPipe = context.Request.BodyPipe;
+            Assert.NotNull(bodyPipe);
+        }
+
+        [Fact]
+        public void BodyPipe_CanSet()
+        {
+            var pipeReader = new Pipe().Reader;
+            var context = new DefaultHttpContext();
+
+            context.Request.BodyPipe = pipeReader;
+
+            Assert.Equal(pipeReader, context.Request.BodyPipe);
+        }
+
+        [Fact]
+        public void BodyPipe_WrapsStream()
+        {
+            var context = new DefaultHttpContext();
+            var expectedStream = new MemoryStream();
+            context.Request.Body = expectedStream;
+
+            var bodyPipe = context.Request.BodyPipe as StreamPipeReader;
+
+            Assert.Equal(expectedStream, bodyPipe.InnerStream);
+        }
+
+        [Fact]
+        public void BodyPipe_ThrowsWhenSettingNull()
+        {
+            var context = new DefaultHttpContext();
+            Assert.Throws<ArgumentNullException>(() => context.Request.BodyPipe = null);
+        }
+
         private class CustomRouteValuesFeature : IRouteValuesFeature
         {
             public RouteValueDictionary RouteValues { get; set; }

src/Http/Http/test/Internal/DefaultHttpResponseTests.cs

@@ -4,6 +4,8 @@
 using System;
 using System.Collections.Generic;
 using System.Globalization;
+using System.IO;
+using System.IO.Pipelines;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.Extensions.Primitives;
 using Xunit;
@@ -59,6 +61,44 @@ namespace Microsoft.AspNetCore.Http.Internal
             Assert.Null(response.ContentType);
         }
 
+        [Fact]
+        public void BodyPipe_CanGet()
+        {
+            var response = new DefaultHttpContext();
+            var bodyPipe = response.Response.BodyPipe;
+
+            Assert.NotNull(bodyPipe);
+        }
+
+        [Fact]
+        public void BodyPipe_CanSet()
+        {
+            var response = new DefaultHttpContext();
+            var pipeWriter = new Pipe().Writer;
+            response.Response.BodyPipe = pipeWriter;
+
+            Assert.Equal(pipeWriter, response.Response.BodyPipe);
+        }
+
+        [Fact]
+        public void BodyPipe_WrapsStream()
+        {
+            var context = new DefaultHttpContext();
+            var expectedStream = new MemoryStream();
+            context.Response.Body = expectedStream;
+
+            var bodyPipe = context.Response.BodyPipe as StreamPipeWriter;
+
+            Assert.Equal(expectedStream, bodyPipe.InnerStream);
+        }
+
+        [Fact]
+        public void BodyPipe_ThrowsWhenSettingNull()
+        {
+            var context = new DefaultHttpContext();
+            Assert.Throws<ArgumentNullException>(() => context.Response.BodyPipe = null);
+        }
+
         private static HttpResponse CreateResponse(IHeaderDictionary headers)
         {
             var context = new DefaultHttpContext();

src/Http/Http/test/PipeTest.cs

@@ -4,6 +4,7 @@
 using System;
 using System.IO;
 using System.IO.Pipelines;
+using System.Text;
 
 namespace System.IO.Pipelines.Tests
 {
@@ -38,6 +39,12 @@ namespace System.IO.Pipelines.Tests
             return ReadWithoutFlush();
         }
 
+        public string ReadAsString()
+        {
+            Writer.FlushAsync().GetAwaiter().GetResult();
+            return Encoding.ASCII.GetString(ReadWithoutFlush());
+        }
+
         public void Write(byte[] data)
         {
             MemoryStream.Write(data, 0, data.Length);

src/Http/Http/test/StreamPipeReaderTests.cs

@@ -1,11 +1,8 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
-using System;
 using System.Buffers;
 using System.Collections.Generic;
-using System.IO;
-using System.IO.Pipelines;
 using System.Linq;
 using System.Text;
 using System.Threading;
@@ -533,6 +530,91 @@ namespace System.IO.Pipelines.Tests
             Assert.Throws<ArgumentNullException>(() => new StreamPipeReader(MemoryStream, null));
         }
 
+        [Fact]
+        public async Task UseBothStreamAndPipeToReadConfirmSameSize()
+        {
+            Write(new byte[8]);
+            var buffer = new byte[4];
+
+            MemoryStream.Read(buffer, 0, buffer.Length);
+            var readResult = await Reader.ReadAsync();
+
+            Assert.Equal(buffer, readResult.Buffer.ToArray());
+        }
+
+        [Fact]
+        public async Task UseStreamThenPipeToReadNoBytesLost()
+        {
+            CreateReader(minimumSegmentSize: 1, minimumReadThreshold: 1);
+
+            var expectedString = WriteString("abcdef");
+            var accumulatedResult = "";
+            var buffer = new byte[1];
+
+            for (var i = 0; i < expectedString.Length / 2; i++)
+            {
+                // Read from stream then pipe to guarantee no bytes are lost.
+                accumulatedResult += ReadFromStreamAsString(buffer);
+                accumulatedResult += await ReadFromPipeAsString();
+            }
+
+            Assert.Equal(expectedString, accumulatedResult);
+        }
+
+        [Fact]
+        public async Task UsePipeThenStreamToReadNoBytesLost()
+        {
+            CreateReader(minimumSegmentSize: 1, minimumReadThreshold: 1);
+
+            var expectedString = WriteString("abcdef");
+            var accumulatedResult = "";
+            var buffer = new byte[1];
+
+            for (var i = 0; i < expectedString.Length / 2; i++)
+            {
+                // Read from pipe then stream to guarantee no bytes are lost.
+                accumulatedResult += await ReadFromPipeAsString();
+                accumulatedResult += ReadFromStreamAsString(buffer);
+            }
+
+            Assert.Equal(expectedString, accumulatedResult);
+        }
+
+        [Fact]
+        public async Task UseBothStreamAndPipeToReadWithoutAdvance_StreamIgnoresAdvance()
+        {
+            var buffer = new byte[1];
+            CreateReader(minimumSegmentSize: 1, minimumReadThreshold: 1);
+
+            WriteString("abc");
+            ReadFromStreamAsString(buffer);
+            var readResult = await Reader.ReadAsync();
+
+            // No Advance
+            // Next call to Stream.Read will get the next 4 bytes rather than the bytes already read by the pipe
+            Assert.Equal("c", ReadFromStreamAsString(buffer));
+        }
+
+        private async Task<string> ReadFromPipeAsString()
+        {
+            var readResult = await Reader.ReadAsync();
+            var result = Encoding.ASCII.GetString(readResult.Buffer.ToArray());
+            Reader.AdvanceTo(readResult.Buffer.End);
+            return result;
+        }
+
+        private string ReadFromStreamAsString(byte[] buffer)
+        {
+            var res = MemoryStream.Read(buffer, 0, buffer.Length);
+            return Encoding.ASCII.GetString(buffer);
+        }
+
+        private string WriteString(string expectedString)
+        {
+            Write(Encoding.ASCII.GetBytes(expectedString));
+            return expectedString;
+        }
+
         private void CreateReader(int minimumSegmentSize = 16, int minimumReadThreshold = 4, MemoryPool<byte> memoryPool = null)
         {
             Reader = new StreamPipeReader(MemoryStream,
@@ -566,7 +648,8 @@ namespace System.IO.Pipelines.Tests
                 return await base.ReadAsync(buffer, offset, count, cancellationToken);
             }
 
-#if NETCOREAPP2_2
+            // Keeping as this code will eventually be ported to corefx
+#if NETCOREAPP3_0
             public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
             {
                 await Task.Yield();

src/Http/Http/test/StreamPipeWriterTests.cs

@@ -269,6 +269,64 @@ namespace System.IO.Pipelines.Tests
             Assert.Equal(16 * 10 * 2, Read().Length);
         }
 
+        [Fact]
+        public async Task UseBothStreamAndPipeToWrite()
+        {
+            await WriteStringToPipeWriter("a");
+            WriteStringToStream("c");
+
+            Assert.Equal("ac", ReadAsString());
+        }
+
+        [Fact]
+        public async Task UsePipeThenStreamToWriteMultipleTimes()
+        {
+            var expectedString = "abcdef";
+            for (var i = 0; i < expectedString.Length; i++)
+            {
+                if (i % 2 == 0)
+                {
+                    WriteStringToStream(expectedString[i].ToString());
+                }
+                else
+                {
+                    await WriteStringToPipeWriter(expectedString[i].ToString());
+                }
+            }
+
+            Assert.Equal(expectedString, ReadAsString());
+        }
+
+        [Fact]
+        public async Task UseStreamThenPipeToWriteMultipleTimes()
+        {
+            var expectedString = "abcdef";
+            for (var i = 0; i < expectedString.Length; i++)
+            {
+                if (i % 2 == 0)
+                {
+                    await WriteStringToPipeWriter(expectedString[i].ToString());
+                }
+                else
+                {
+                    WriteStringToStream(expectedString[i].ToString());
+                }
+            }
+
+            Assert.Equal(expectedString, ReadAsString());
+        }
+
+        private void WriteStringToStream(string input)
+        {
+            var buffer = Encoding.ASCII.GetBytes(input);
+            MemoryStream.Write(buffer, 0, buffer.Length);
+        }
+
+        private async Task WriteStringToPipeWriter(string input)
+        {
+            await Writer.WriteAsync(Encoding.ASCII.GetBytes(input));
+        }
+
         private async Task CheckWriteIsNotCanceled()
         {
             var flushResult = await Writer.WriteAsync(Encoding.ASCII.GetBytes("data"));
@@ -291,7 +349,6 @@ namespace System.IO.Pipelines.Tests
 
     internal class HangingStream : MemoryStream
     {
-
         public HangingStream()
         {
         }
@@ -311,7 +368,9 @@ namespace System.IO.Pipelines.Tests
             await Task.Delay(30000, cancellationToken);
             return 0;
         }
-#if NETCOREAPP2_2
+
+        // Keeping as this code will eventually be ported to corefx
+#if NETCOREAPP3_0
         public override async ValueTask<int> ReadAsync(Memory<byte> destination, CancellationToken cancellationToken = default)
         {
             await Task.Delay(30000, cancellationToken);
@@ -326,8 +385,8 @@ namespace System.IO.Pipelines.Tests
 
         public bool AllowAllWrites { get; set; }
 
-
+        // Keeping as this code will eventually be ported to corefx
-#if NETCOREAPP2_2
+#if NETCOREAPP3_0
         public override async ValueTask WriteAsync(ReadOnlyMemory<byte> source, CancellationToken cancellationToken = default)
         {
             try

src/Http/build.cmd

@@ -0,0 +1,3 @@
+@ECHO OFF
+SET RepoRoot=%~dp0..\..
+%RepoRoot%\build.cmd -projects %~dp0\**\*.*proj %*

src/Middleware/WebSockets/test/UnitTests/Microsoft.AspNetCore.WebSockets.Tests.csproj

@@ -5,8 +5,8 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.AspNetCore.Server.IntegrationTesting" Version="$(MicrosoftAspNetCoreServerIntegrationTestingPackageVersion)" />
+    <Reference Include="Microsoft.AspNetCore.Server.IntegrationTesting" />
-    <PackageReference Include="Microsoft.AspNetCore.Server.Kestrel" Version="$(MicrosoftAspNetCoreServerKestrelPackageVersion)" />
+    <Reference Include="Microsoft.AspNetCore.Server.Kestrel" />
     <Reference Include="Microsoft.AspNetCore.WebSockets" />
     <Reference Include="Microsoft.Extensions.Logging.Testing" />
   </ItemGroup>