No end stream on ws connect and flush every message

src/libraries/System.Net.Http/src/System/Net/Http/HttpRequestMessage.cs

@@ -170,6 +170,8 @@ public override string ToString()
 
         internal bool IsWebSocketH2Request() => _version.Major == 2 && Method == HttpMethod.Connect && HasHeaders && string.Equals(Headers.Protocol, "websocket", StringComparison.OrdinalIgnoreCase);
 
+        internal bool IsExtendedConnectRequest => Method == HttpMethod.Connect && _headers?.Protocol != null;
+
         #region IDisposable Members
 
         protected virtual void Dispose(bool disposing)

src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http2Connection.cs

@@ -1600,7 +1600,7 @@ private async ValueTask<Http2Stream> SendHeadersAsync(HttpRequestMessage request
                 // Start the write.  This serializes access to write to the connection, and ensures that HEADERS
                 // and CONTINUATION frames stay together, as they must do. We use the lock as well to ensure new
                 // streams are created and started in order.
-                await PerformWriteAsync(totalSize, (thisRef: this, http2Stream, headerBytes, endStream: (request.Content == null && !http2Stream.ConnectProtocolEstablished), mustFlush), static (s, writeBuffer) =>
+                await PerformWriteAsync(totalSize, (thisRef: this, http2Stream, headerBytes, endStream: (request.Content == null && !request.IsExtendedConnectRequest), mustFlush), static (s, writeBuffer) =>
                 {
                     if (NetEventSource.Log.IsEnabled()) s.thisRef.Trace(s.http2Stream.StreamId, $"Started writing. Total header bytes={s.headerBytes.Length}");
 
@@ -1962,8 +1962,8 @@ public async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, boo
             try
             {
                 // Send request headers
-                bool shouldExpectContinue = request.Content != null && request.HasHeaders && request.Headers.ExpectContinue == true;
-                Http2Stream http2Stream = await SendHeadersAsync(request, cancellationToken, mustFlush: shouldExpectContinue).ConfigureAwait(false);
+                bool shouldExpectContinue = (request.Content != null && request.HasHeaders && request.Headers.ExpectContinue == true);
+                Http2Stream http2Stream = await SendHeadersAsync(request, cancellationToken, mustFlush: shouldExpectContinue || request.IsExtendedConnectRequest).ConfigureAwait(false);
 
                 bool duplex = request.Content != null && request.Content.AllowDuplex;
 

src/libraries/System.Net.WebSockets.Client/tests/SendReceiveTest.Http2.cs

@@ -57,6 +57,9 @@ await Http2LoopbackServer.CreateClientAndServerAsync(async uri =>
                 // send status 200 OK to establish websocket
                 await connection.SendResponseHeadersAsync(streamId, endStream: false).ConfigureAwait(false);
 
+                WebSocket websocket = WebSocket.CreateFromStream(connection.Stream, isServer: true, null, Timeout.InfiniteTimeSpan);
+                Assert.Equal(WebSocketState.Open, websocket.State);
+
                 // send reply
                 byte binaryMessageType = 2;
                 var prefix = new byte[] { binaryMessageType, (byte)serverMessage.Length };
@@ -97,6 +100,9 @@ await Http2LoopbackServer.CreateClientAndServerAsync(async uri =>
                 // send status 200 OK to establish websocket
                 await connection.SendResponseHeadersAsync(streamId, endStream: false).ConfigureAwait(false);
 
+                WebSocket websocket = WebSocket.CreateFromStream(connection.Stream, isServer: true, null, Timeout.InfiniteTimeSpan);
+                Assert.Equal(WebSocketState.Open, websocket.State);
+
                 // send reply
                 byte binaryMessageType = 2;
                 var prefix = new byte[] { binaryMessageType, (byte)serverMessage.Length };

src/libraries/System.Net.WebSockets/src/System/Net/WebSockets/ManagedWebSocket.cs

@@ -423,6 +423,8 @@ private ValueTask SendFrameLockAcquiredNonCancelableAsync(MessageOpcode opcode,
                 // the task, and we're done.
                 if (writeTask.IsCompleted)
                 {
+                    writeTask.GetAwaiter().GetResult();
+                    writeTask = new ValueTask(_stream.FlushAsync());
                     return writeTask;
                 }
 
@@ -455,6 +457,7 @@ private async ValueTask WaitForWriteTaskAsync(ValueTask writeTask)
             try
             {
                 await writeTask.ConfigureAwait(false);
+                await _stream.FlushAsync().ConfigureAwait(false);
             }
             catch (Exception exc) when (!(exc is OperationCanceledException))
             {
@@ -478,6 +481,7 @@ private async ValueTask SendFrameFallbackAsync(MessageOpcode opcode, bool endOfM
                 using (cancellationToken.Register(static s => ((ManagedWebSocket)s!).Abort(), this))
                 {
                     await _stream.WriteAsync(new ReadOnlyMemory<byte>(_sendBuffer, 0, sendBytes), cancellationToken).ConfigureAwait(false);
+                    await _stream.FlushAsync(cancellationToken).ConfigureAwait(false);
                 }
             }
             catch (Exception exc) when (!(exc is OperationCanceledException))

src/libraries/System.Net.WebSockets/tests/WebSocketTestStream.cs

@@ -206,7 +206,7 @@ public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, Cancella
             Write(buffer.Span);
         }
 
-        public override void Flush() => throw new NotSupportedException();
+        public override void Flush() { }
 
         public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();