ResponseCaching: started conversion to pipes

Author: alefranz

src/Middleware/ResponseCaching/src/CacheEntry/CachedResponse.cs

@@ -2,7 +2,6 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
-using System.IO;
 using Microsoft.AspNetCore.Http;
 
 namespace Microsoft.AspNetCore.ResponseCaching
@@ -15,6 +14,6 @@ internal class CachedResponse : IResponseCacheEntry
 
         public IHeaderDictionary Headers { get; set; }
 
-        public Stream Body { get; set; }
+        public CachedResponseBody Body { get; set; }
     }
 }

src/Middleware/ResponseCaching/src/CacheEntry/CachedResponseBody.cs

@@ -0,0 +1,47 @@
+// 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.Collections.Generic;
+using System.IO.Pipelines;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.AspNetCore.ResponseCaching
+{
+    internal class CachedResponseBody
+    {
+        public CachedResponseBody(List<byte[]> segments, long length)
+        {
+            Segments = segments;
+            Length = length;
+        }
+
+        public List<byte[]> Segments { get; }
+
+        public long Length { get; }
+
+        public async Task CopyToAsync(PipeWriter destination, CancellationToken cancellationToken)
+        {
+            if (destination == null)
+            {
+                throw new ArgumentNullException(nameof(destination));
+            }
+
+            for (var segmentIndex = 0; segmentIndex < Segments.Count; segmentIndex++)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                var segmentLength = Segments[segmentIndex].Length;
+
+                var memory = destination.GetMemory(segmentLength);
+                Segments[segmentIndex].AsMemory().CopyTo(memory);
+
+                destination.Advance(segmentLength);
+
+                await destination.FlushAsync();
+            }
+
+            await destination.CompleteAsync();
+        }
+    }
+}

src/Middleware/ResponseCaching/src/MemoryCachedResponse.cs

@@ -1,4 +1,4 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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;
@@ -15,8 +15,6 @@ internal class MemoryCachedResponse
 
         public IHeaderDictionary Headers { get; set; } = new HeaderDictionary();
 
-        public List<byte[]> BodySegments { get; set; }
-
-        public long BodyLength { get; set; }
+        public CachedResponseBody Body { get; set; }
     }
 }

src/Middleware/ResponseCaching/src/MemoryResponseCache.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;
+using System.Buffers;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Caching.Memory;
 
@@ -27,7 +28,7 @@ public IResponseCacheEntry Get(string key)
                     Created = memoryCachedResponse.Created,
                     StatusCode = memoryCachedResponse.StatusCode,
                     Headers = memoryCachedResponse.Headers,
-                    Body = new SegmentReadStream(memoryCachedResponse.BodySegments, memoryCachedResponse.BodyLength)
+                    Body = memoryCachedResponse.Body
                 };
             }
             else
@@ -40,18 +41,14 @@ public void Set(string key, IResponseCacheEntry entry, TimeSpan validFor)
         {
             if (entry is CachedResponse cachedResponse)
             {
-                var segmentStream = new SegmentWriteStream(StreamUtilities.BodySegmentSize);
-                cachedResponse.Body.CopyTo(segmentStream);
-
                 _cache.Set(
                     key,
                     new MemoryCachedResponse
                     {
                         Created = cachedResponse.Created,
                         StatusCode = cachedResponse.StatusCode,
                         Headers = cachedResponse.Headers,
-                        BodySegments = segmentStream.GetSegments(),
-                        BodyLength = segmentStream.Length
+                        Body = cachedResponse.Body
                     },
                     new MemoryCacheEntryOptions
                     {

src/Middleware/ResponseCaching/src/ResponseCachingMiddleware.cs

@@ -193,13 +193,13 @@ internal async Task<bool> TryServeCachedResponseAsync(ResponseCachingContext con
                     {
                         try
                         {
-                            await body.CopyToAsync(response.Body, StreamUtilities.BodySegmentSize, context.HttpContext.RequestAborted);
+                            await body.CopyToAsync(response.BodyWriter, context.HttpContext.RequestAborted);
                         }
-                        catch (OperationCanceledException)
+                            catch (OperationCanceledException)
                         {
                             context.HttpContext.Abort();
                         }
-                    }
+                }
                     _logger.CachedResponseServed();
                 }
                 return true;
@@ -344,17 +344,17 @@ internal void FinalizeCacheBody(ResponseCachingContext context)
             if (context.ShouldCacheResponse && context.ResponseCachingStream.BufferingEnabled)
             {
                 var contentLength = context.HttpContext.Response.ContentLength;
-                var bufferStream = context.ResponseCachingStream.GetBufferStream();
-                if (!contentLength.HasValue || contentLength == bufferStream.Length)
+                var cachedResponseBody = context.ResponseCachingStream.GetCachedResponseBody();
+                if (!contentLength.HasValue || contentLength == cachedResponseBody.Length)
                 {
                     var response = context.HttpContext.Response;
                     // Add a content-length if required
                     if (!response.ContentLength.HasValue && StringValues.IsNullOrEmpty(response.Headers[HeaderNames.TransferEncoding]))
                     {
-                        context.CachedResponse.Headers[HeaderNames.ContentLength] = HeaderUtilities.FormatNonNegativeInt64(bufferStream.Length);
+                        context.CachedResponse.Headers[HeaderNames.ContentLength] = HeaderUtilities.FormatNonNegativeInt64(cachedResponseBody.Length);
                     }
 
-                    context.CachedResponse.Body = bufferStream;
+                    context.CachedResponse.Body = cachedResponseBody;
                     _logger.ResponseCached();
                     _cache.Set(context.StorageVaryKey ?? context.BaseKey, context.CachedResponse, context.CachedResponseValidFor);
                 }

src/Middleware/ResponseCaching/src/Streams/ResponseCachingStream.cs

@@ -45,13 +45,13 @@ public override long Position
             }
         }
 
-        internal Stream GetBufferStream()
+        internal CachedResponseBody GetCachedResponseBody()
         {
             if (!BufferingEnabled)
             {
                 throw new InvalidOperationException("Buffer stream cannot be retrieved since buffering is disabled.");
             }
-            return new SegmentReadStream(_segmentWriteStream.GetSegments(), _segmentWriteStream.Length);
+            return new CachedResponseBody(_segmentWriteStream.GetSegments(), _segmentWriteStream.Length);
         }
 
         internal void DisableBuffering()