diff --git a/dio/CHANGELOG.md b/dio/CHANGELOG.md index 8089c2f21..39fa74449 100644 --- a/dio/CHANGELOG.md +++ b/dio/CHANGELOG.md @@ -5,7 +5,8 @@ See the [Migration Guide][] for the complete breaking changes list.** ## Unreleased -*None.* +- Release response resources and stop receive timeouts when a streamed response + consumer cancels its subscription or stops reading with `first` or `take`. ## 5.11.1 diff --git a/dio/lib/src/response/response_stream_handler.dart b/dio/lib/src/response/response_stream_handler.dart index b37de1563..989887b79 100644 --- a/dio/lib/src/response/response_stream_handler.dart +++ b/dio/lib/src/response/response_stream_handler.dart @@ -70,6 +70,15 @@ Stream handleResponseStream( }); } + responseSink.onCancel = () { + stopWatchReceiveTimeout(); + if (!responseSink.isClosed) { + response.close(); + responseSink.close(); + } + return responseSubscription.cancel(); + }; + // Also start watching before the first chunk to cover the gap between // headers and body (see onData below for per-chunk reset). watchReceiveTimeout(); @@ -99,6 +108,9 @@ Stream handleResponseStream( ); options.cancelToken?.whenCancel.whenComplete(() { + if (responseSink.isClosed) { + return; + } stopWatchReceiveTimeout(); // Close the response stream upon a cancellation. response.close(); diff --git a/dio/test/response/response_stream_test.dart b/dio/test/response/response_stream_test.dart index 8431e0db6..110eb5179 100644 --- a/dio/test/response/response_stream_test.dart +++ b/dio/test/response/response_stream_test.dart @@ -7,6 +7,179 @@ import 'package:dio_test/util.dart'; import 'package:test/test.dart'; void main() { + group('ResponseType.stream consumer cancellation', () { + test('first cancels the adapter response stream', () async { + var cancelCount = 0; + final source = StreamController( + onCancel: () => cancelCount++, + ); + final cancelToken = CancelToken(); + final dio = Dio() + ..httpClientAdapter = + _StreamingAdapter(ResponseBody(source.stream, 200)); + addTearDown(() async { + cancelToken.cancel(); + await Future.delayed(Duration.zero); + await source.close(); + dio.close(); + }); + + final response = await dio.get( + 'https://example.test/stream', + options: Options(responseType: ResponseType.stream), + cancelToken: cancelToken, + ); + final first = response.data!.stream.first; + source.add(Uint8List.fromList([1])); + + expect(await first, [1]); + expect(source.hasListener, isFalse); + expect(cancelCount, 1); + }); + + for (final pauseBeforeCancel in [false, true]) { + test( + 'cancel releases the response${pauseBeforeCancel ? ' while paused' : ''}', + () async { + final source = StreamController(); + var closeCount = 0; + final progress = []; + final cancelToken = CancelToken(); + final dio = Dio() + ..httpClientAdapter = _StreamingAdapter( + ResponseBody(source.stream, 200, onClose: () => closeCount++), + ); + addTearDown(() async { + cancelToken.cancel(); + await Future.delayed(Duration.zero); + await source.close(); + dio.close(); + }); + + final response = await dio.get( + 'https://example.test/stream', + options: Options( + responseType: ResponseType.stream, + receiveTimeout: const Duration(milliseconds: 30), + ), + cancelToken: cancelToken, + onReceiveProgress: (count, _) => progress.add(count), + ); + final subscription = response.data!.stream.listen((_) {}); + source.add(Uint8List.fromList([1])); + await Future.delayed(Duration.zero); + if (pauseBeforeCancel) { + subscription.pause(); + expect(source.isPaused, isTrue); + } + + await subscription.cancel(); + expect(source.hasListener, isFalse); + expect(closeCount, 1); + + source.add(Uint8List.fromList([2])); + await Future.delayed(const Duration(milliseconds: 60)); + expect(progress, [1]); + expect(closeCount, 1, reason: 'The receive timer must not fire.'); + + cancelToken.cancel(); + await Future.delayed(Duration.zero); + expect( + closeCount, + 1, + reason: 'Later cancellation must not close again.', + ); + }); + } + + test('cancel waits for asynchronous adapter cleanup', () async { + final cleanupStarted = Completer(); + final cleanupFinished = Completer(); + final source = StreamController( + onCancel: () { + cleanupStarted.complete(); + return cleanupFinished.future; + }, + ); + final cancelToken = CancelToken(); + final dio = Dio() + ..httpClientAdapter = + _StreamingAdapter(ResponseBody(source.stream, 200)); + addTearDown(() async { + if (!cleanupFinished.isCompleted) { + cleanupFinished.complete(); + } + cancelToken.cancel(); + await Future.delayed(Duration.zero); + await source.close(); + dio.close(); + }); + final response = await dio.get( + 'https://example.test/stream', + options: Options(responseType: ResponseType.stream), + cancelToken: cancelToken, + ); + var cancellationCompleted = false; + final cancellation = response.data!.stream.listen((_) {}).cancel().then( + (_) => cancellationCompleted = true, + ); + + await Future.delayed(Duration.zero); + expect(cleanupStarted.isCompleted, isTrue); + expect(cancellationCompleted, isFalse); + cleanupFinished.complete(); + await cancellation; + expect(cancellationCompleted, isTrue); + }); + + for (final withError in [false, true]) { + test( + 'late token cancellation after source ${withError ? 'error' : 'done'}', + () async { + var sourceCancelCount = 0; + var closeCount = 0; + final source = StreamController( + onCancel: () => sourceCancelCount++, + ); + final cancelToken = CancelToken(); + final dio = Dio() + ..httpClientAdapter = _StreamingAdapter( + ResponseBody(source.stream, 200, onClose: () => closeCount++), + ); + addTearDown(() async { + await source.close(); + dio.close(); + }); + final response = await dio.get( + 'https://example.test/stream', + options: Options(responseType: ResponseType.stream), + cancelToken: cancelToken, + ); + final received = response.data!.stream.toList(); + if (withError) { + source.addError(StateError('response failed')); + await expectLater(received, throwsA(isA())); + } else { + source.add(Uint8List.fromList([1])); + source.close(); + expect( + await received, + [ + Uint8List.fromList([1]), + ], + ); + } + + expect(sourceCancelCount, 1); + expect(closeCount, 0); + cancelToken.cancel(); + await Future.delayed(Duration.zero); + expect(sourceCancelCount, 1); + expect(closeCount, 0); + }); + } + }); + group(handleResponseStream, () { late StreamController source; @@ -392,3 +565,20 @@ void main() { }); }); } + +class _StreamingAdapter implements HttpClientAdapter { + _StreamingAdapter(this.response); + + final ResponseBody response; + + @override + Future fetch( + RequestOptions options, + Stream? requestStream, + Future? cancelFuture, + ) async => + response; + + @override + void close({bool force = false}) {} +}