Ensure PipeWriter is flushed

src/Microsoft.AspNetCore.SystemWebAdapters.CoreServices/Features/HttpResponseAdapterFeature.cs

@@ -91,6 +91,11 @@ bool IHttpResponseBufferingFeature.IsEnabled
 
     private async ValueTask FlushInternalAsync()
     {
+        if (_pipeWriter is { })
+        {
+            await _pipeWriter.FlushAsync();
+        }
+
         if (_state is StreamState.Buffering && _bufferedStream is not null && !SuppressContent)
         {
             if (_filter is { } filter)
@@ -111,7 +116,23 @@ private async ValueTask FlushInternalAsync()
 
     Stream IHttpResponseBodyFeature.Stream => this;
 
-    PipeWriter IHttpResponseBodyFeature.Writer => _pipeWriter ??= PipeWriter.Create(this, new StreamPipeWriterOptions(leaveOpen: true));
+    PipeWriter IHttpResponseBodyFeature.Writer
+    {
+        get
+        {
+            if (_pipeWriter is null)
+            {
+                _pipeWriter = PipeWriter.Create(this, new StreamPipeWriterOptions(leaveOpen: true));
+
+                if (_state is StreamState.Complete)
+                {
+                    _pipeWriter.Complete();
+                }
+            }
+
+            return _pipeWriter;
+        }
+    }
 
     public bool SuppressContent
     {
@@ -229,6 +250,11 @@ private async Task CompleteAsync()
 
         _state = StreamState.Complete;
 
+        if (_pipeWriter is { })
+        {
+            await _pipeWriter.CompleteAsync();
+        }
+
         await _responseBodyFeature.CompleteAsync();
     }
 

test/Microsoft.AspNetCore.SystemWebAdapters.CoreServices.Tests/ResponseStreamTests.cs

@@ -2,6 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 
 using System;
+using System.Buffers;
 using System.Collections.Generic;
 using System.IO;
 using System.Text;
@@ -42,6 +43,32 @@ public async Task BufferedOutputIsFlushed()
         Assert.Equal(ContentValue, result);
     }
 
+    [Fact]
+    public async Task OutputPipeIsFlushed()
+    {
+        const string Message = "hello world from pipe!";
+
+        var result = await RunAsync(context =>
+        {
+            context.AsAspNetCore().Response.BodyWriter.Write(Encoding.UTF8.GetBytes(Message));
+        });
+
+        Assert.Equal(Message, result);
+    }
+
+    [Fact]
+    public async Task OutputPipeIsMarkedCompleteIfRequestIsComplete()
+    {
+        var result = await RunAsync(async context =>
+        {
+            await context.AsAspNetCore().Response.CompleteAsync();
+
+            Assert.Throws<InvalidOperationException>(() => context.AsAspNetCore().Response.BodyWriter.Write("Hello world"u8));
+        });
+
+        Assert.Empty(result);
+    }
+
     [Fact]
     public async Task BufferedOutputIsFlushedOnceWithStart()
     {