@@ -105,7 +105,9 @@ public Http2Stream(HttpRequestMessage request, Http2Connection connection)
105105
106106 _headerBudgetRemaining = connection . _pool . Settings . MaxResponseHeadersByteLength ;
107107
108- if ( _request . Content == null )
108+ // Extended connect requests will use the response content stream for bidirectional communication.
109+ // We will ignore any content set for such requests in SendRequestBodyAsync, as it has no defined semantics.
110+ if ( _request . Content == null || _request . IsExtendedConnectRequest )
109111 {
110112 _requestCompletionState = StreamCompletionState . Completed ;
111113 if ( _request . IsExtendedConnectRequest )
@@ -173,7 +175,9 @@ public HttpResponseMessage GetAndClearResponse()
173175
174176 public async Task SendRequestBodyAsync ( CancellationToken cancellationToken )
175177 {
176- if ( _request . Content == null )
178+ // Extended connect requests will use the response content stream for bidirectional communication.
179+ // Ignore any content set for such requests, as it has no defined semantics.
180+ if ( _request . Content == null || _request . IsExtendedConnectRequest )
177181 {
178182 Debug . Assert ( _requestCompletionState == StreamCompletionState . Completed ) ;
179183 return ;
@@ -250,6 +254,7 @@ public async Task SendRequestBodyAsync(CancellationToken cancellationToken)
250254 // and we also don't want to propagate any error to the caller, in particular for non-duplex scenarios.
251255 Debug . Assert ( _responseCompletionState == StreamCompletionState . Completed ) ;
252256 _requestCompletionState = StreamCompletionState . Completed ;
257+ Debug . Assert ( ! ConnectProtocolEstablished ) ;
253258 Complete ( ) ;
254259 return ;
255260 }
@@ -261,6 +266,7 @@ public async Task SendRequestBodyAsync(CancellationToken cancellationToken)
261266
262267 _requestCompletionState = StreamCompletionState . Failed ;
263268 SendReset ( ) ;
269+ Debug . Assert ( ! ConnectProtocolEstablished ) ;
264270 Complete ( ) ;
265271 }
266272
@@ -313,6 +319,7 @@ public async Task SendRequestBodyAsync(CancellationToken cancellationToken)
313319
314320 if ( complete )
315321 {
322+ Debug . Assert ( ! ConnectProtocolEstablished ) ;
316323 Complete ( ) ;
317324 }
318325 }
@@ -420,7 +427,17 @@ private void Cancel()
420427 if ( sendReset )
421428 {
422429 SendReset ( ) ;
423- Complete ( ) ;
430+
431+ // Extended CONNECT notes:
432+ //
433+ // To prevent from calling it *twice*, Extended CONNECT stream's Complete() is only
434+ // called from CloseResponseBody(), as CloseResponseBody() is *always* called
435+ // from Extended CONNECT stream's Dispose().
436+
437+ if ( ! ConnectProtocolEstablished )
438+ {
439+ Complete ( ) ;
440+ }
424441 }
425442 }
426443
@@ -810,7 +827,20 @@ public void OnHeadersComplete(bool endStream)
810827 Debug . Assert ( _responseCompletionState == StreamCompletionState . InProgress , $ "Response already completed with state={ _responseCompletionState } ") ;
811828
812829 _responseCompletionState = StreamCompletionState . Completed ;
813- if ( _requestCompletionState == StreamCompletionState . Completed )
830+
831+ // Extended CONNECT notes:
832+ //
833+ // To prevent from calling it *prematurely*, Extended CONNECT stream's Complete() is only
834+ // called from CloseResponseBody(), as CloseResponseBody() is *only* called
835+ // from Extended CONNECT stream's Dispose().
836+ //
837+ // Due to bidirectional streaming nature of the Extended CONNECT request,
838+ // the *write side* of the stream can only be completed by calling Dispose().
839+ //
840+ // The streaming in both ways happens over the single "response" stream instance, which makes
841+ // _requestCompletionState *not indicative* of the actual state of the write side of the stream.
842+
843+ if ( _requestCompletionState == StreamCompletionState . Completed && ! ConnectProtocolEstablished )
814844 {
815845 Complete ( ) ;
816846 }
@@ -871,7 +901,20 @@ public void OnResponseData(ReadOnlySpan<byte> buffer, bool endStream)
871901 Debug . Assert ( _responseCompletionState == StreamCompletionState . InProgress , $ "Response already completed with state={ _responseCompletionState } ") ;
872902
873903 _responseCompletionState = StreamCompletionState . Completed ;
874- if ( _requestCompletionState == StreamCompletionState . Completed )
904+
905+ // Extended CONNECT notes:
906+ //
907+ // To prevent from calling it *prematurely*, Extended CONNECT stream's Complete() is only
908+ // called from CloseResponseBody(), as CloseResponseBody() is *only* called
909+ // from Extended CONNECT stream's Dispose().
910+ //
911+ // Due to bidirectional streaming nature of the Extended CONNECT request,
912+ // the *write side* of the stream can only be completed by calling Dispose().
913+ //
914+ // The streaming in both ways happens over the single "response" stream instance, which makes
915+ // _requestCompletionState *not indicative* of the actual state of the write side of the stream.
916+
917+ if ( _requestCompletionState == StreamCompletionState . Completed && ! ConnectProtocolEstablished )
875918 {
876919 Complete ( ) ;
877920 }
@@ -1036,17 +1079,17 @@ public async Task ReadResponseHeadersAsync(CancellationToken cancellationToken)
10361079 Debug . Assert ( _response != null && _response . Content != null ) ;
10371080 // Start to process the response body.
10381081 var responseContent = ( HttpConnectionResponseContent ) _response . Content ;
1039- if ( emptyResponse )
1082+ if ( ConnectProtocolEstablished )
1083+ {
1084+ responseContent . SetStream ( new Http2ReadWriteStream ( this , closeResponseBodyOnDispose : true ) ) ;
1085+ }
1086+ else if ( emptyResponse )
10401087 {
10411088 // If there are any trailers, copy them over to the response. Normally this would be handled by
10421089 // the response stream hitting EOF, but if there is no response body, we do it here.
10431090 MoveTrailersToResponseMessage ( _response ) ;
10441091 responseContent . SetStream ( EmptyReadStream . Instance ) ;
10451092 }
1046- else if ( ConnectProtocolEstablished )
1047- {
1048- responseContent . SetStream ( new Http2ReadWriteStream ( this ) ) ;
1049- }
10501093 else
10511094 {
10521095 responseContent . SetStream ( new Http2ReadStream ( this ) ) ;
@@ -1309,8 +1352,25 @@ private async ValueTask SendDataAsync(ReadOnlyMemory<byte> buffer, CancellationT
13091352 }
13101353 }
13111354
1355+ // This method should only be called from Http2ReadWriteStream.Dispose()
13121356 private void CloseResponseBody ( )
13131357 {
1358+ // Extended CONNECT notes:
1359+ //
1360+ // Due to bidirectional streaming nature of the Extended CONNECT request,
1361+ // the *write side* of the stream can only be completed by calling Dispose()
1362+ // (which, for Extended CONNECT case, will in turn call CloseResponseBody())
1363+ //
1364+ // Similarly to QuicStream, disposal *gracefully* closes the write side of the stream
1365+ // (unless we've received RST_STREAM before) and *abortively* closes the read side
1366+ // of the stream (unless we've received EOS before).
1367+
1368+ if ( ConnectProtocolEstablished && _resetException is null )
1369+ {
1370+ // Gracefully close the write side of the Extended CONNECT stream
1371+ _connection . LogExceptions ( _connection . SendEndStreamAsync ( StreamId ) ) ;
1372+ }
1373+
13141374 // Check if the response body has been fully consumed.
13151375 bool fullyConsumed = false ;
13161376 Debug . Assert ( ! Monitor . IsEntered ( SyncObject ) ) ;
@@ -1323,6 +1383,7 @@ private void CloseResponseBody()
13231383 }
13241384
13251385 // If the response body isn't completed, cancel it now.
1386+ // This includes aborting the read side of the Extended CONNECT stream.
13261387 if ( ! fullyConsumed )
13271388 {
13281389 Cancel ( ) ;
@@ -1337,6 +1398,12 @@ private void CloseResponseBody()
13371398
13381399 lock ( SyncObject )
13391400 {
1401+ if ( ConnectProtocolEstablished )
1402+ {
1403+ // This should be the only place where Extended Connect stream is completed
1404+ Complete ( ) ;
1405+ }
1406+
13401407 _responseBuffer . Dispose ( ) ;
13411408 }
13421409 }
@@ -1430,10 +1497,7 @@ private enum StreamCompletionState : byte
14301497
14311498 private sealed class Http2ReadStream : Http2ReadWriteStream
14321499 {
1433- public Http2ReadStream ( Http2Stream http2Stream ) : base ( http2Stream )
1434- {
1435- base . CloseResponseBodyOnDispose = true ;
1436- }
1500+ public Http2ReadStream ( Http2Stream http2Stream ) : base ( http2Stream , closeResponseBodyOnDispose : true ) { }
14371501
14381502 public override bool CanWrite => false ;
14391503
@@ -1482,12 +1546,13 @@ public class Http2ReadWriteStream : HttpBaseStream
14821546 private Http2Stream ? _http2Stream ;
14831547 private readonly HttpResponseMessage _responseMessage ;
14841548
1485- public Http2ReadWriteStream ( Http2Stream http2Stream )
1549+ public Http2ReadWriteStream ( Http2Stream http2Stream , bool closeResponseBodyOnDispose = false )
14861550 {
14871551 Debug . Assert ( http2Stream != null ) ;
14881552 Debug . Assert ( http2Stream . _response != null ) ;
14891553 _http2Stream = http2Stream ;
14901554 _responseMessage = _http2Stream . _response ;
1555+ CloseResponseBodyOnDispose = closeResponseBodyOnDispose ;
14911556 }
14921557
14931558 ~ Http2ReadWriteStream ( )
@@ -1503,7 +1568,7 @@ public Http2ReadWriteStream(Http2Stream http2Stream)
15031568 }
15041569 }
15051570
1506- protected bool CloseResponseBodyOnDispose { get ; set ; }
1571+ protected bool CloseResponseBodyOnDispose { get ; private init ; }
15071572
15081573 protected override void Dispose ( bool disposing )
15091574 {
0 commit comments