fix(dio): propagate response-stream backpressure to the source

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
 

dio/lib/src/response/response_stream_handler.dart

@@ -21,8 +21,11 @@ Stream<Uint8List> handleResponseStream(
   @visibleForTesting void Function()? onReceiveTimeoutWatchCancelled,
 }) {
   final source = response.stream;
-  final responseSink = StreamController<Uint8List>();
   late StreamSubscription<List<int>> responseSubscription;
+  final responseSink = StreamController<Uint8List>(
+    onPause: () => responseSubscription.pause(),
+    onResume: () => responseSubscription.resume(),
+  );
 
   late int totalLength;
   int receivedLength = 0;

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<Uint8List>(
+        onPause: () => upstreamPaused = true,
+        onResume: () => upstreamResumed = true,
+      );
+
+      final stream = handleResponseStream(
+        RequestOptions(),
+        ResponseBody(observableSource.stream, 200),
+      );
+
+      final received = <int>[];
+      late StreamSubscription<List<int>> 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<Uint8List>();
+      final stream = handleResponseStream(
+        RequestOptions(),
+        ResponseBody(source.stream, 200),
+      );
+
+      late StreamSubscription<List<int>> 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();
+    });
   });
 }