fix(http2_adapter): apply receiveTimeout to response body streaming

plugins/http2_adapter/CHANGELOG.md

@@ -5,7 +5,9 @@ See the [Migration Guide][] for the complete breaking changes list.**
 
 ## Unreleased
 
-*None.*
+- Fix `receiveTimeout` so it also covers the response body: the inactivity
+  timer is now reset on every headers and data message, and a stalled body
+  stream now throws `DioException.receiveTimeout` instead of hanging.
 
 ## 2.9.0
 

plugins/http2_adapter/lib/src/http2_adapter.dart

@@ -215,10 +215,41 @@ class Http2Adapter implements HttpClientAdapter {
 
     final receiveTimeout = options.receiveTimeout ?? Duration.zero;
 
+    // Inactivity timer for the response body. The existing
+    // receiveTimeout only wraps the headers future, so without this a
+    // server that sends headers and then stalls mid-body would hang
+    // forever. This timer is reset on every headers and data message
+    // and cancelled when the body stream ends, errors, or is closed.
+    Timer? chunkTimer;
+    void cancelChunkTimer() {
+      chunkTimer?.cancel();
+      chunkTimer = null;
+    }
+
+    void resetChunkTimer() {
+      cancelChunkTimer();
+      if (receiveTimeout > Duration.zero) {
+        chunkTimer = Timer(receiveTimeout, () {
+          responseSubscription.cancel().catchError((_) {}).whenComplete(() {
+            if (!responseSink.isClosed) {
+              responseSink.addError(
+                DioException.receiveTimeout(
+                  timeout: receiveTimeout,
+                  requestOptions: options,
+                ),
+              );
+              responseSink.close();
+            }
+          });
+        });
+      }
+    }
+
     late int statusCode;
     responseSubscription = stream.incomingMessages.listen(
       (StreamMessage message) async {
         if (message is HeadersStreamMessage) {
+          resetChunkTimer();
           for (final header in message.headers) {
             final name = utf8.decode(header.name);
             final value = utf8.decode(header.value);
@@ -237,6 +268,7 @@ class Http2Adapter implements HttpClientAdapter {
           }
         } else if (message is DataStreamMessage) {
           if (needResponse) {
+            resetChunkTimer();
             responseSink.add(
               message.bytes is Uint8List
                   ? message.bytes as Uint8List
@@ -250,6 +282,7 @@ class Http2Adapter implements HttpClientAdapter {
         }
       },
       onError: (Object error, StackTrace stackTrace) {
+        cancelChunkTimer();
         // If connection is being forcefully terminated, remove the connection.
         if (error is TransportConnectionException) {
           connectionManager.removeConnection(transport);
@@ -263,6 +296,7 @@ class Http2Adapter implements HttpClientAdapter {
         responseSink.close();
       },
       onDone: () {
+        cancelChunkTimer();
         responseSubscription.cancel();
         responseSink.close();
       },
@@ -289,6 +323,7 @@ class Http2Adapter implements HttpClientAdapter {
 
     // Handle redirection.
     if (needRedirect) {
+      cancelChunkTimer();
       if (responseHeaders['location'] == null) {
         // Redirect without location is illegal.
         throw DioException.connectionError(
@@ -324,6 +359,7 @@ class Http2Adapter implements HttpClientAdapter {
       redirects: redirects,
       isRedirect: redirects.isNotEmpty,
       onClose: () {
+        cancelChunkTimer();
         responseSubscription.cancel();
         responseSink.close();
         streamWR.target?.outgoingMessages.close();

plugins/http2_adapter/test/http2_test.dart

@@ -229,6 +229,45 @@ void main() {
       expect(response.statusCode, 200);
       expect(requestController.hasListener, isFalse);
     });
+
+    test('throws receiveTimeout when the body stalls after headers', () async {
+      serverSocket.listen((socket) {
+        final connection = ServerTransportConnection.viaSocket(socket);
+        serverConnections.add(connection);
+        connection.incomingStreams.listen((stream) {
+          stream.sendHeaders(
+            [Header.ascii(':status', '200')],
+            endStream: false,
+          );
+          stream.sendData([1, 2, 3], endStream: false);
+          // Stall here: never send more data and never close the stream.
+        });
+      });
+
+      final response = await adapter
+          .fetch(
+            RequestOptions(
+              path: '/',
+              method: 'GET',
+              baseUrl: 'http://127.0.0.1:${serverSocket.port}',
+              receiveTimeout: const Duration(milliseconds: 200),
+            ),
+            null,
+            null,
+          )
+          .timeout(const Duration(seconds: 2));
+      expect(response.statusCode, 200);
+      await expectLater(
+        response.stream.toList().timeout(const Duration(seconds: 2)),
+        throwsA(
+          isA<DioException>().having(
+            (e) => e.type,
+            'type',
+            DioExceptionType.receiveTimeout,
+          ),
+        ),
+      );
+    });
   });
 
   group(ConnectionManager, () {