Implement Stream Flush+FlushAsync fully

Streams should pass through the flush and not assume the underlying
Stream's Flush behaviour

src/Microsoft.AspNetCore.Server.Kestrel/Adapter/Internal/LoggingStream.cs

@@ -71,6 +71,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Adapter.Internal
             _inner.Flush();
         }
 
+        public override Task FlushAsync(CancellationToken cancellationToken)
+        {
+            return _inner.FlushAsync(cancellationToken);
+        }
+
         public override int Read(byte[] buffer, int offset, int count)
         {
             int read = _inner.Read(buffer, offset, count);

src/Microsoft.AspNetCore.Server.Kestrel/Adapter/Internal/RawStream.cs

@@ -117,15 +117,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Adapter.Internal
 
         public override void Flush()
         {
-            // No-op since writes are immediate.
+            _output.Flush();
         }
 
         public override Task FlushAsync(CancellationToken cancellationToken)
         {
-            // No-op since writes are immediate.
-            return TaskCache.CompletedTask;
+            return _output.FlushAsync(cancellationToken);
         }
 
+
         private ValueTask<int> ReadAsync(ArraySegment<byte> buffer)
         {
             return _input.ReadAsync(buffer.Array, buffer.Offset, buffer.Count);

src/Microsoft.AspNetCore.Server.Kestrel/Adapter/Internal/StreamSocketOutput.cs

@@ -8,7 +8,6 @@ using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Server.Kestrel.Internal.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Internal.Infrastructure;
-using Microsoft.Extensions.Internal;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Adapter.Internal
 {
@@ -114,14 +113,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Adapter.Internal
             end.Block.Pool.Return(end.Block);
         }
 
-        // Flush no-ops. We rely on connection filter streams to auto-flush.
         public void Flush()
         {
+            _outputStream.Flush();
         }
 
         public Task FlushAsync(CancellationToken cancellationToken)
         {
-            return TaskCache.CompletedTask;
+            return _outputStream.FlushAsync(cancellationToken);
         }
     }
 }

test/Microsoft.AspNetCore.Server.KestrelTests/ConnectionAdapterTests.cs

@@ -186,7 +186,12 @@ namespace Microsoft.AspNetCore.Server.KestrelTests
 
             public override void Flush()
             {
-                // No-op
+                _innerStream.Flush();
+            }
+
+            public override Task FlushAsync(CancellationToken cancellationToken)
+            {
+                return _innerStream.FlushAsync(cancellationToken);
             }
 
             public override int Read(byte[] buffer, int offset, int count)