fix(range): make sure RangeStream can only be listened to once (#690)

* fix(range): ensure RangeStream is a single-subscription Stream * fix(range): ensure RangeStream is a single-subscription Stream
ReactiveX · Aug 2, 2022 · 58c82c0 · 58c82c0
1 parent e13dad0
commit 58c82c0
Showing 1 changed file with 25 additions and 5 deletions.
diff --git a/lib/src/streams/range.dart b/lib/src/streams/range.dart
@@ -23,11 +23,31 @@ class RangeStream extends Stream<int> {
           onError: onError, onDone: onDone, cancelOnError: cancelOnError);
 
   static Stream<int> _buildStream(int startInclusive, int endInclusive) {
-    final length = (endInclusive - startInclusive).abs() + 1;
-    int nextValue(int index) => startInclusive > endInclusive
-        ? startInclusive - index
-        : startInclusive + index;
+    final controller = StreamController<int>(sync: true);
+    StreamSubscription<int>? subscription;
 
-    return Stream.fromIterable(Iterable.generate(length, nextValue));
+    controller.onListen = () {
+      final length = (endInclusive - startInclusive).abs() + 1;
+      int nextValue(int index) => startInclusive > endInclusive
+          ? startInclusive - index
+          : startInclusive + index;
+
+      subscription =
+          Stream.fromIterable(Iterable.generate(length, nextValue)).listen(
+        controller.add,
+        onError: controller.addError,
+        onDone: controller.close,
+      );
+
+      controller.onPause = subscription!.pause;
+      controller.onResume = subscription!.resume;
+    };
+    controller.onCancel = () {
+      final future = subscription?.cancel();
+      subscription = null;
+      return future;
+    };
+
+    return controller.stream;
   }
 }