Merge branch 'master' into nnbd

Author: hoc081098

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.0.2 - Jan 12, 2021
+
+-   Internal refactor
+
 ## 0.0.1 - Dec 15, 2020
 
 -   Initial version.

README.md

@@ -7,6 +7,9 @@
 [![codecov](https://codecov.io/gh/Flutter-Dart-Open-Source/listenable_stream/branch/master/graph/badge.svg?token=6eORcR6Web)](https://codecov.io/gh/Flutter-Dart-Open-Source/listenable_stream)
 [![Flutter Tests](https://github.com/Flutter-Dart-Open-Source/listenable_stream/workflows/Flutter%20Tests/badge.svg)](https://github.com/Flutter-Dart-Open-Source/listenable_stream.git)
 
+- [x] `Listenable` ▶ `Stream<Listenable>`
+- [x] `ValueListenable<T>` ▶ `ValueStream<T>`
+
 ## Listenable.toStream()
 ```dart
 final ChangeNotifier changeNotifier = ChangeNotifier();

lib/src/common.dart

@@ -0,0 +1,46 @@
+import 'dart:async';
+
+import 'package:flutter/foundation.dart' show Listenable, VoidCallback;
+
+/// @private
+/// Convert this [Listenable] to a [Stream].
+Stream<R> toStreamWithTransform<T extends Listenable, R>(
+  T listenable,
+  R Function(T) transform,
+) {
+  late StreamController<R> controller;
+  VoidCallback? listener;
+
+  final onListenOrOnResume = () {
+    assert(listener == null);
+    try {
+      listenable
+          .addListener(listener = () => controller.add(transform(listenable)));
+    } catch (_ /*Ignore*/) {
+      controller.close();
+    }
+  };
+
+  final createOnPauseOrOnCancel = ([bool closeOnError = false]) {
+    return () {
+      assert(listener != null);
+      try {
+        listenable.removeListener(listener!);
+        listener = null;
+      } catch (_ /*Ignore*/) {
+        if (identical(closeOnError, true)) {
+          controller.close();
+        }
+      }
+    };
+  };
+
+  controller = StreamController<R>(
+    onListen: onListenOrOnResume,
+    onPause: createOnPauseOrOnCancel(true),
+    onResume: onListenOrOnResume,
+    onCancel: createOnPauseOrOnCancel(),
+  );
+
+  return controller.stream;
+}

lib/src/listenable_to_stream.dart

@@ -1,25 +1,11 @@
 import 'dart:async';
 
-import 'package:flutter/foundation.dart' show Listenable, VoidCallback;
+import 'package:flutter/foundation.dart' show Listenable;
+
+import 'common.dart';
 
 /// Convert this [Listenable] to a [Stream].
 extension ListenableToStream<T extends Listenable> on T {
   /// Convert this [Listenable] to a [Stream].
-  Stream<T> toStream() {
-    late StreamController<T> controller;
-    VoidCallback? listener;
-
-    controller = StreamController<T>(
-      sync: true,
-      onListen: () => addListener(listener = () => controller.add(this)),
-      onCancel: () {
-        try {
-          removeListener(listener!);
-          listener = null;
-        } catch (_ /*Ignore*/) {}
-      },
-    );
-
-    return controller.stream;
-  }
+  Stream<T> toStream() => toStreamWithTransform<T, T>(this, (t) => t);
 }

lib/src/value_listenable_to_value_stream.dart

@@ -1,12 +1,17 @@
 import 'dart:async' show Stream, StreamSubscription;
 
 import 'package:flutter/foundation.dart' show ValueListenable;
-import 'package:rxdart/rxdart.dart' show ErrorAndStackTrace, ValueStream;
+import 'package:rxdart/rxdart.dart' show ValueStream;
 import 'package:rxdart/src/utils/value_wrapper.dart';
 
-import 'listenable_to_stream.dart';
+import 'common.dart';
 import 'streamx.dart';
 
+extension _ValueListenableToStreamExtension<T> on ValueListenable<T> {
+  Stream<T> toStream() =>
+      toStreamWithTransform<ValueListenable<T>, T>(this, (l) => l.value);
+}
+
 /// Convert this [ValueListenable] to a [ValueStream].
 /// The returned [ValueStream] is a Single-Subscription [Stream].
 ///
@@ -28,8 +33,6 @@ class ValueListenableStream<T> extends Stream<T> implements ValueStream<T> {
   final bool _replayValue;
   Stream<T>? _stream;
 
-  T _getValue([void _]) => _valueListenable.value;
-
   /// Construct a [ValueListenableStream] from [ValueListenable].
   ValueListenableStream(this._valueListenable, this._replayValue);
 
@@ -42,9 +45,9 @@ class ValueListenableStream<T> extends Stream<T> implements ValueStream<T> {
   }) {
     if (_replayValue) {
       _stream ??=
-          _valueListenable.toStream().map(_getValue).startWith(_getValue);
+          _valueListenable.toStream().startWith(() => _valueListenable.value);
     } else {
-      _stream ??= _valueListenable.toStream().map(_getValue);
+      _stream ??= _valueListenable.toStream();
     }
 
     return _stream!.listen(
@@ -63,5 +66,5 @@ class ValueListenableStream<T> extends Stream<T> implements ValueStream<T> {
       throw StateError('This Stream always has no error!');
 
   @override
-  ValueWrapper<T> get valueWrapper => ValueWrapper(_getValue());
+  ValueWrapper<T> get valueWrapper => ValueWrapper(_valueListenable.value);
 }

pubspec.yaml

@@ -1,6 +1,6 @@
 name: listenable_stream
 description: Convert Listenable (eg. ChangeNotifier), ValueListenable(eg. ValueNotifier) to Stream / ValueStream.
-version: 0.0.1
+version: 1.0.0-nullsafety
 author: Petrus Nguyễn Thái Học <hoc081098@gmail.com>
 homepage: https://github.com/Flutter-Dart-Open-Source/listenable_stream.git
 repository: https://github.com/Flutter-Dart-Open-Source/listenable_stream.git

test/listenable_stream_test.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:flutter/foundation.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:listenable_stream/listenable_stream.dart';
@@ -37,6 +39,50 @@ void main() {
       final stream = ChangeNotifier().toStream();
       _isSingleSubscriptionStream(stream);
     });
+
+    test('Cancel', () async {
+      final changeNotifier = ChangeNotifier();
+      final stream = changeNotifier.toStream();
+
+      final subscription = stream.listen(
+        expectAsync1(
+          (e) => expect(e, changeNotifier),
+          count: 3,
+        ),
+      );
+
+      changeNotifier.notifyListeners();
+      changeNotifier.notifyListeners();
+      changeNotifier.notifyListeners();
+
+      await pumpEventQueue();
+      await subscription.cancel();
+
+      changeNotifier.notifyListeners();
+      changeNotifier.notifyListeners();
+    });
+
+    test('Pause resume', () async {
+      final changeNotifier = ChangeNotifier();
+      final stream = changeNotifier.toStream();
+
+      final subscription = stream.listen(
+        expectAsync1(
+          (v) => expect(v, changeNotifier),
+          count: 1,
+        ),
+      )..pause();
+
+      // no effect
+      changeNotifier.notifyListeners();
+      changeNotifier.notifyListeners();
+      changeNotifier.notifyListeners();
+
+      await Future<void>.delayed(const Duration(milliseconds: 50));
+      subscription.resume();
+
+      changeNotifier.notifyListeners();
+    });
   });
 
   group('ValueListenableToStream', () {
@@ -82,5 +128,132 @@ void main() {
         _isSingleSubscriptionStream(stream);
       }
     });
+
+    test('Cancel', () async {
+      {
+        final valueNotifier = ValueNotifier(0);
+        final stream = valueNotifier.toValueStream();
+
+        var i = 1;
+        final subscription = stream.listen(
+          expectAsync1(
+            (e) => expect(e, i++),
+            count: 3,
+          ),
+        );
+
+        valueNotifier.value = 1;
+        valueNotifier.value = 2;
+        valueNotifier.value = 3;
+
+        await pumpEventQueue();
+        await subscription.cancel();
+
+        valueNotifier.value = 4;
+        valueNotifier.value = 5;
+      }
+
+      {
+        final valueNotifier = ValueNotifier(0);
+        final stream = valueNotifier.toValueStream(replayValue: true);
+
+        var i = 0;
+        final subscription = stream.listen(
+          expectAsync1(
+            (e) => expect(e, i++),
+            count: 4,
+          ),
+        );
+
+        valueNotifier.value = 1;
+        valueNotifier.value = 2;
+        valueNotifier.value = 3;
+
+        await pumpEventQueue();
+        await subscription.cancel();
+
+        valueNotifier.value = 4;
+        valueNotifier.value = 5;
+      }
+    });
+
+    group('Pause resume', () {
+      test('not replay', () async {
+        final valueNotifier = ValueNotifier(0);
+        final stream = valueNotifier.toValueStream();
+        final expected = 4;
+
+        final subscription = stream.listen(
+          expectAsync1(
+            (v) => expect(v, expected),
+            count: 1,
+          ),
+        )..pause();
+
+        // no effect
+        valueNotifier.value = 1;
+        valueNotifier.value = 2;
+        valueNotifier.value = 3;
+
+        await Future<void>.delayed(const Duration(milliseconds: 50));
+        subscription.resume();
+
+        valueNotifier.value = expected;
+      });
+
+      test('replay + pause immediately', () async {
+        final valueNotifier = ValueNotifier(0);
+        final stream = valueNotifier.toValueStream(replayValue: true);
+        final expected = [0, 4, 5];
+
+        var i = 0;
+        final subscription = stream.listen(
+          expectAsync1(
+            (v) => expect(v, expected[i++]),
+            count: expected.length,
+            max: expected.length,
+          ),
+        )..pause();
+
+        // no effect
+        valueNotifier.value = 1;
+        valueNotifier.value = 2;
+        valueNotifier.value = 3;
+
+        subscription.resume();
+
+        await pumpEventQueue();
+        valueNotifier.value = 4;
+        valueNotifier.value = 5;
+      });
+
+      test('replay + pause after events queue.', () async {
+        final valueNotifier = ValueNotifier(0);
+        final stream = valueNotifier.toValueStream(replayValue: true);
+        final expected = [0, 4, 5];
+
+        var i = 0;
+        final subscription = stream.listen(
+          expectAsync1(
+            (v) => expect(v, expected[i++]),
+            count: expected.length,
+            max: expected.length,
+          ),
+        );
+
+        await pumpEventQueue();
+        subscription.pause();
+
+        // no effect
+        valueNotifier.value = 1;
+        valueNotifier.value = 2;
+        valueNotifier.value = 3;
+
+        subscription.resume();
+
+        valueNotifier.value = 4;
+        valueNotifier.value = 5;
+      });
+    });
   });
 }