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/Filter/Internal/LibuvStream.cs

@@ -117,13 +117,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Filter.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 _output.FlushAsync(cancellationToken);
-            return TaskCache.CompletedTask;
         }
 
         private ValueTask<int> ReadAsync(ArraySegment<byte> buffer)

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

@@ -71,6 +71,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Filter.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/Filter/Internal/StreamSocketOutput.cs

@@ -114,14 +114,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Filter.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/ConnectionFilterTests.cs

@@ -160,7 +160,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)