@@ -14,6 +14,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
14
14
/// </summary>
15
15
internal sealed class Http1UpgradeMessageBody : Http1MessageBody
16
16
{
17
+ private int _userCanceled ;
18
+
17
19
public Http1UpgradeMessageBody ( Http1Connection context , bool keepAlive )
18
20
: base ( context , keepAlive )
19
21
{
@@ -26,13 +28,13 @@ public Http1UpgradeMessageBody(Http1Connection context, bool keepAlive)
26
28
public override ValueTask < ReadResult > ReadAsync ( CancellationToken cancellationToken = default )
27
29
{
28
30
ThrowIfCompleted ( ) ;
29
- return _context . Input . ReadAsync ( cancellationToken ) ;
31
+ return ReadAsyncInternal ( cancellationToken ) ;
30
32
}
31
33
32
34
public override bool TryRead ( out ReadResult result )
33
35
{
34
36
ThrowIfCompleted ( ) ;
35
- return _context . Input . TryRead ( out result ) ;
37
+ return TryReadInternal ( out result ) ;
36
38
}
37
39
38
40
public override void AdvanceTo ( SequencePosition consumed )
@@ -54,6 +56,7 @@ public override void Complete(Exception exception)
54
56
55
57
public override void CancelPendingRead ( )
56
58
{
59
+ Interlocked . Exchange ( ref _userCanceled , 1 ) ;
57
60
_context . Input . CancelPendingRead ( ) ;
58
61
}
59
62
@@ -69,12 +72,49 @@ public override Task StopAsync()
69
72
70
73
public override bool TryReadInternal ( out ReadResult readResult )
71
74
{
72
- return _context . Input . TryRead ( out readResult ) ;
75
+ // Ignore the canceled readResult unless it was canceled by the user.
76
+ do
77
+ {
78
+ if ( ! _context . Input . TryRead ( out readResult ) )
79
+ {
80
+ return false ;
81
+ }
82
+ } while ( readResult . IsCanceled && Interlocked . Exchange ( ref _userCanceled , 0 ) == 0 ) ;
83
+
84
+ return true ;
73
85
}
74
86
75
87
public override ValueTask < ReadResult > ReadAsyncInternal ( CancellationToken cancellationToken = default )
76
88
{
77
- return _context . Input . ReadAsync ( cancellationToken ) ;
89
+ ReadResult readResult ;
90
+
91
+ // Ignore the canceled readResult unless it was canceled by the user.
92
+ do
93
+ {
94
+ var readTask = _context . Input . ReadAsync ( cancellationToken ) ;
95
+
96
+ if ( ! readTask . IsCompletedSuccessfully )
97
+ {
98
+ return ReadAsyncInternalAwaited ( readTask , cancellationToken ) ;
99
+ }
100
+
101
+ readResult = readTask . GetAwaiter ( ) . GetResult ( ) ;
102
+ } while ( readResult . IsCanceled && Interlocked . Exchange ( ref _userCanceled , 0 ) == 0 ) ;
103
+
104
+ return new ValueTask < ReadResult > ( readResult ) ;
105
+ }
106
+
107
+ private async ValueTask < ReadResult > ReadAsyncInternalAwaited ( ValueTask < ReadResult > readTask , CancellationToken cancellationToken = default )
108
+ {
109
+ var readResult = await readTask ;
110
+
111
+ // Ignore the canceled readResult unless it was canceled by the user.
112
+ while ( readResult . IsCanceled && Interlocked . Exchange ( ref _userCanceled , 0 ) == 0 )
113
+ {
114
+ readResult = await _context . Input . ReadAsync ( cancellationToken ) ;
115
+ }
116
+
117
+ return readResult ;
78
118
}
79
119
}
80
120
}
0 commit comments