Extract receiveTimeout handling from adapters to response stream handler

cfug · Dec 19, 2023 · f5cebf8 · f5cebf8
1 parent 6a62c8a
commit f5cebf8
Show file tree

Hide file tree

Showing 6 changed files with 98 additions and 155 deletions.
diff --git a/dio/lib/src/adapters/io_adapter.dart b/dio/lib/src/adapters/io_adapter.dart
@@ -198,69 +198,12 @@ class IOHttpClientAdapter implements HttpClientAdapter {
       }
     }
 
-    // Use a StreamController to explicitly handle receive timeouts.
-    final responseSink = StreamController<Uint8List>();
-    late StreamSubscription<List<int>> responseSubscription;
-
-    final receiveStopwatch = Stopwatch();
-    Timer? receiveTimer;
-
-    void stopWatchReceiveTimeout() {
-      receiveTimer?.cancel();
-      receiveTimer = null;
-      receiveStopwatch.stop();
-    }
-
-    void watchReceiveTimeout() {
-      if (receiveTimeout <= Duration.zero) {
-        return;
-      }
-      receiveStopwatch.reset();
-      if (!receiveStopwatch.isRunning) {
-        receiveStopwatch.start();
-      }
-      receiveTimer?.cancel();
-      receiveTimer = Timer(receiveTimeout, () {
-        responseSink.addError(
-          DioException.receiveTimeout(
-            timeout: receiveTimeout,
-            requestOptions: options,
-          ),
-        );
-        responseSink.close();
-        responseSubscription.cancel();
-        responseStream.detachSocket().then((socket) => socket.destroy());
-        stopWatchReceiveTimeout();
-      });
-    }
-
-    responseSubscription = responseStream.cast<Uint8List>().listen(
-      (data) {
-        watchReceiveTimeout();
-        // Always true if the receive timeout was not set.
-        if (receiveStopwatch.elapsed <= receiveTimeout) {
-          responseSink.add(data);
-        }
-      },
-      onError: (error, stackTrace) {
-        stopWatchReceiveTimeout();
-        responseSink.addError(error, stackTrace);
-        responseSink.close();
-      },
-      onDone: () {
-        stopWatchReceiveTimeout();
-        responseSubscription.cancel();
-        responseSink.close();
-      },
-      cancelOnError: true,
-    );
-
     final headers = <String, List<String>>{};
     responseStream.headers.forEach((key, values) {
       headers[key] = values;
     });
     return ResponseBody(
-      responseSink.stream,
+      responseStream.cast(),
       responseStream.statusCode,
       headers: headers,
       isRedirect:
@@ -270,10 +213,7 @@ class IOHttpClientAdapter implements HttpClientAdapter {
           .toList(),
       statusMessage: responseStream.reasonPhrase,
       onClose: () {
-        responseSink.close();
-        responseSubscription.cancel();
         responseStream.detachSocket().then((socket) => socket.destroy());
-        stopWatchReceiveTimeout();
       },
     );
   }

diff --git a/dio/lib/src/dio_mixin.dart b/dio/lib/src/dio_mixin.dart
@@ -536,6 +536,8 @@ abstract class DioMixin implements Dio {
       );
       final statusOk = reqOpt.validateStatus(responseBody.statusCode);
       if (statusOk || reqOpt.receiveDataWhenStatusError == true) {
+        responseBody.stream = handleResponseStream(reqOpt, responseBody);
+
         Object? data = await transformer.transformResponse(
           reqOpt,
           responseBody,
@@ -547,13 +549,10 @@ abstract class DioMixin implements Dio {
             T != String &&
             reqOpt.responseType == ResponseType.json) {
           data = null;
-        } else if (reqOpt.responseType == ResponseType.stream &&
-            data is ResponseBody) {
-          data.stream = handleResponseStream(reqOpt, data);
         }
         ret.data = data;
       } else {
-        await responseBody.stream.listen(null).cancel();
+        responseBody.close();
       }
       checkCancelled(cancelToken);
       if (statusOk) {

diff --git a/dio/lib/src/response/response_stream_handler.dart b/dio/lib/src/response/response_stream_handler.dart
@@ -3,37 +3,80 @@ import 'dart:typed_data';
 
 import 'package:dio/dio.dart';
 
-/// An internal helper function to handle things around
-/// streamed responses.
+/// An internal helper which handles functionality
+/// common to all adapters. This function ensures that
+/// all resources are closed when the request is finished
+/// or cancelled.
+///
+/// - [options.receiveTimeout] between received chunks
+/// - [options.onReceiveProgress] progress for received chunks
+/// - [options.cancelToken] for cancellation while receiving
 Stream<Uint8List> handleResponseStream(
   RequestOptions options,
   ResponseBody response,
 ) {
   final source = response.stream;
-
-  // Use a StreamController to explicitly handle receive timeouts.
   final responseSink = StreamController<Uint8List>();
   late StreamSubscription<List<int>> responseSubscription;
 
   late int totalLength;
+  int receivedLength = 0;
   if (options.onReceiveProgress != null) {
     totalLength = response.contentLength;
   }
 
-  int dataLength = 0;
+  final receiveTimeout = options.receiveTimeout ?? Duration.zero;
+  final receiveStopwatch = Stopwatch();
+  Timer? receiveTimer;
+
+  void stopWatchReceiveTimeout() {
+    receiveTimer?.cancel();
+    receiveTimer = null;
+    receiveStopwatch.stop();
+  }
+
+  void watchReceiveTimeout() {
+    if (receiveTimeout <= Duration.zero) {
+      return;
+    }
+    receiveStopwatch.reset();
+    if (!receiveStopwatch.isRunning) {
+      receiveStopwatch.start();
+    }
+    receiveTimer?.cancel();
+    receiveTimer = Timer(receiveTimeout, () {
+      responseSink.addError(
+        DioException.receiveTimeout(
+          timeout: receiveTimeout,
+          requestOptions: options,
+        ),
+      );
+      response.close();
+      responseSink.close();
+      responseSubscription.cancel();
+      stopWatchReceiveTimeout();
+    });
+  }
+
   responseSubscription = source.listen(
     (data) {
-      responseSink.add(data);
-      options.onReceiveProgress?.call(
-        dataLength += data.length,
-        totalLength,
-      );
+      watchReceiveTimeout();
+      // Always true if the receive timeout was not set.
+      if (receiveStopwatch.elapsed <= receiveTimeout) {
+        responseSink.add(data);
+        options.onReceiveProgress?.call(
+          receivedLength += data.length,
+          totalLength,
+        );
+      }
     },
     onError: (error, stackTrace) {
+      stopWatchReceiveTimeout();
       responseSink.addError(error, stackTrace);
       responseSink.close();
     },
     onDone: () {
+      stopWatchReceiveTimeout();
       responseSubscription.cancel();
       responseSink.close();
     },

diff --git a/dio/lib/src/transformers/sync_transformer.dart b/dio/lib/src/transformers/sync_transformer.dart
@@ -62,43 +62,8 @@ class SyncTransformer extends Transformer {
       return responseBody;
     }
 
-    final int totalLength;
-    if (options.onReceiveProgress != null) {
-      totalLength = responseBody.contentLength;
-    } else {
-      totalLength = 0;
-    }
-
-    final streamCompleter = Completer<void>();
-    int finalLength = 0;
-    // Keep references to the data chunks and concatenate them later.
-    final chunks = <Uint8List>[];
-    final subscription = responseBody.stream.listen(
-      (Uint8List chunk) {
-        finalLength += chunk.length;
-        chunks.add(chunk);
-        options.onReceiveProgress?.call(finalLength, totalLength);
-      },
-      onError: (Object error, StackTrace stackTrace) {
-        streamCompleter.completeError(error, stackTrace);
-      },
-      onDone: () {
-        streamCompleter.complete();
-      },
-      cancelOnError: true,
-    );
-    options.cancelToken?.whenCancel.then((_) {
-      return subscription.cancel();
-    });
-    await streamCompleter.future;
-
-    // Copy all chunks into the final bytes.
-    final responseBytes = Uint8List(finalLength);
-    int chunkOffset = 0;
-    for (final chunk in chunks) {
-      responseBytes.setAll(chunkOffset, chunk);
-      chunkOffset += chunk.length;
-    }
+    final chunks = await responseBody.stream.toList();
+    final responseBytes = Uint8List.fromList(chunks.expand((c) => c).toList());
 
     // Return the finalized bytes if the response type is bytes.
     if (responseType == ResponseType.bytes) {

diff --git a/dio/test/response/response_stream_test.dart b/dio/test/response/response_stream_test.dart
@@ -17,9 +17,7 @@ void main() {
 
     test('completes', () async {
       final stream = handleResponseStream(
-        RequestOptions(
-          cancelToken: CancelToken(),
-        ),
+        RequestOptions(),
         ResponseBody(
           source.stream,
           200,
@@ -80,7 +78,6 @@ void main() {
 
       final stream = handleResponseStream(
         RequestOptions(
-          cancelToken: CancelToken(),
           onReceiveProgress: (c, t) {
             count = c;
             total = t;
@@ -132,7 +129,6 @@ void main() {
 
       final stream = handleResponseStream(
         RequestOptions(
-          cancelToken: CancelToken(),
           onReceiveProgress: (c, t) {
             count = c;
             total = t;
@@ -177,9 +173,7 @@ void main() {
 
     test('emits error on source error', () async {
       final stream = handleResponseStream(
-        RequestOptions(
-          cancelToken: CancelToken(),
-        ),
+        RequestOptions(),
         ResponseBody(
           source.stream,
           200,
@@ -203,5 +197,42 @@ void main() {
         expect(source.hasListener, isFalse);
       });
     });
+
+    test('emits error on receiveTimeout', () async {
+      final stream = handleResponseStream(
+        RequestOptions(
+          receiveTimeout: Duration(milliseconds: 100),
+        ),
+        ResponseBody(
+          source.stream,
+          200,
+        ),
+      );
+
+      expectLater(
+        stream,
+        emitsInOrder([
+          Uint8List.fromList([0]),
+          Uint8List.fromList([1]),
+          emitsError(matchesDioException(
+            DioExceptionType.receiveTimeout,
+            stackTraceContains: 'test/response/response_stream_test.dart',
+          )),
+          emitsDone,
+        ]),
+      );
+
+      source.add(Uint8List.fromList([0]));
+      await Future.delayed(Duration(milliseconds: 90), () {
+        source.add(Uint8List.fromList([1]));
+      });
+      await Future.delayed(Duration(milliseconds: 110), () {
+        source.add(Uint8List.fromList([2]));
+      });
+
+      await Future.delayed(Duration(milliseconds: 100), () {
+        expect(source.hasListener, isFalse);
+      });
+    });
   });
 }
diff --git a/plugins/http2_adapter/lib/src/http2_adapter.dart b/plugins/http2_adapter/lib/src/http2_adapter.dart
@@ -121,37 +121,6 @@ class Http2Adapter implements HttpClientAdapter {
     bool needResponse = false;
 
     final receiveTimeout = options.receiveTimeout ?? Duration.zero;
-    final receiveStopwatch = Stopwatch();
-    Timer? receiveTimer;
-
-    void stopWatchReceiveTimeout() {
-      receiveTimer?.cancel();
-      receiveTimer = null;
-      receiveStopwatch.stop();
-    }
-
-    void watchReceiveTimeout() {
-      if (receiveTimeout <= Duration.zero) {
-        return;
-      }
-      receiveStopwatch.reset();
-      if (!receiveStopwatch.isRunning) {
-        receiveStopwatch.start();
-      }
-      receiveTimer?.cancel();
-      receiveTimer = Timer(receiveTimeout, () {
-        responseSink.addError(
-          DioException.receiveTimeout(
-            timeout: receiveTimeout,
-            requestOptions: options,
-          ),
-        );
-        responseSink.close();
-        responseSubscription.cancel();
-        stream.terminate();
-        stopWatchReceiveTimeout();
-      });
-    }
 
     late int statusCode;
     responseSubscription = stream.incomingMessages.listen(
@@ -175,14 +144,12 @@ class Http2Adapter implements HttpClientAdapter {
           }
         } else if (message is DataStreamMessage) {
           if (needResponse) {
-            watchReceiveTimeout();
             responseSink.add(
               message.bytes is Uint8List
                   ? message.bytes as Uint8List
                   : Uint8List.fromList(message.bytes),
             );
           } else {
-            stopWatchReceiveTimeout();
             responseSubscription.cancel().whenComplete(() {
               stream.terminate();
               responseSink.close();
@@ -200,12 +167,10 @@ class Http2Adapter implements HttpClientAdapter {
         } else {
           responseSink.addError(error, stackTrace);
         }
-        stopWatchReceiveTimeout();
         responseSubscription.cancel();
         responseSink.close();
       },
       onDone: () {
-        stopWatchReceiveTimeout();
         responseSubscription.cancel();
         responseSink.close();
       },