diff --git a/dio/CHANGELOG.md b/dio/CHANGELOG.md index 5ea9e79e3..acd9ddfba 100644 --- a/dio/CHANGELOG.md +++ b/dio/CHANGELOG.md @@ -5,7 +5,10 @@ See the [Migration Guide][] for the complete breaking changes list.** ## Unreleased -*None.* +- Fix response stream not propagating backpressure to the underlying socket. + When a consumer paused the stream, the source subscription was never paused, + so the network kept buffering response data into memory, risking OOM on + constrained platforms. ## 5.11.0 diff --git a/dio/lib/src/response/response_stream_handler.dart b/dio/lib/src/response/response_stream_handler.dart index ad6f32b4a..b37de1563 100644 --- a/dio/lib/src/response/response_stream_handler.dart +++ b/dio/lib/src/response/response_stream_handler.dart @@ -21,8 +21,11 @@ Stream handleResponseStream( @visibleForTesting void Function()? onReceiveTimeoutWatchCancelled, }) { final source = response.stream; - final responseSink = StreamController(); late StreamSubscription> responseSubscription; + final responseSink = StreamController( + onPause: () => responseSubscription.pause(), + onResume: () => responseSubscription.resume(), + ); late int totalLength; int receivedLength = 0; diff --git a/dio/test/response/response_stream_test.dart b/dio/test/response/response_stream_test.dart index bc0b8982d..8431e0db6 100644 --- a/dio/test/response/response_stream_test.dart +++ b/dio/test/response/response_stream_test.dart @@ -288,5 +288,107 @@ void main() { await Future.microtask(() {}); expect(timerCancelled, isTrue); }); + + test('propagates downstream pause/resume as backpressure to the source', + () async { + // An observable upstream: its onPause/onResume fire only when its own + // subscriber pauses — proving backpressure reaches the socket. + var upstreamPaused = false; + var upstreamResumed = false; + final observableSource = StreamController( + onPause: () => upstreamPaused = true, + onResume: () => upstreamResumed = true, + ); + + final stream = handleResponseStream( + RequestOptions(), + ResponseBody(observableSource.stream, 200), + ); + + final received = []; + late StreamSubscription> downstream; + var pausedOnce = false; + downstream = stream.listen((data) { + received.addAll(data); + // The consumer is slow: apply backpressure once, then stay drained. + if (!pausedOnce) { + pausedOnce = true; + downstream.pause(); + } + }); + + observableSource.add(Uint8List.fromList([1])); + // Let the chunk travel downstream and the pause travel upstream. + await Future.delayed(Duration.zero); + + expect( + upstreamPaused, + isTrue, + reason: 'A downstream pause must propagate to the source. ' + 'Otherwise the socket keeps draining into memory (OOM risk).', + ); + + // While paused, additional data must not reach the consumer. + observableSource.add(Uint8List.fromList([2, 3])); + await Future.delayed(Duration.zero); + expect(received, [1], reason: 'No data should arrive while paused.'); + + downstream.resume(); + // Resuming flows the buffered upstream chunk through. + await Future.delayed(Duration.zero); + expect(upstreamResumed, isTrue); + expect(received, [1, 2, 3]); + + // Tear down without depending on done delivery. + await downstream.cancel(); + await observableSource.close(); + }); + + test('does not buffer an unbounded source when the consumer pauses', + () async { + // Model a socket-like source with an explicit production loop that + // yields control between chunks and stops once the source subscription + // is paused. This is platform-agnostic (unlike an `async*` generator, + // whose pause semantics are not honored under dart2wasm) and directly + // mirrors a network socket that keeps pushing until backpressure arrives. + final source = StreamController(); + final stream = handleResponseStream( + RequestOptions(), + ResponseBody(source.stream, 200), + ); + + late StreamSubscription> downstream; + downstream = stream.listen((_) { + // Slow consumer applies backpressure after the first chunk. + downstream.pause(); + }); + + // The source keeps pushing data; a correct pipe pauses the subscription + // and the loop observes it via [StreamController.isPaused]. + var produced = 0; + for (var i = 0; i < 10000; i++) { + if (source.isPaused) { + break; + } + produced++; + source.add(Uint8List(1024)); + // Let the event loop deliver the chunk and propagate the pause. + await Future.delayed(Duration.zero); + } + + expect( + produced, + lessThan(5), + reason: 'With backpressure the source must halt after the consumer ' + 'pauses. Without it the entire response (~10MB here) would be ' + 'buffered in memory, which triggers OOM on constrained devices.', + ); + + // Clear the backpressure so the upstream subscription can drain and + // the source can close, then tear down. + downstream.resume(); + await downstream.cancel(); + await source.close(); + }); }); }