nnbd

Author: hoc081098

lib/src/listenable_to_stream.dart

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

lib/src/streamx.dart

@@ -7,7 +7,7 @@ extension StartWithExtension<T> on Stream<T> {
   /// Prepends a value to the source [Stream].
   Stream<T> startWith(T Function() seededProvider) {
     final controller = StreamController<T>(sync: true);
-    StreamSubscription<T> subscription;
+    late StreamSubscription<T> subscription;
 
     controller.onListen = () {
       controller.add(seededProvider());

lib/src/value_listenable_to_value_stream.dart

@@ -2,6 +2,7 @@ import 'dart:async' show Stream, StreamSubscription;
 
 import 'package:flutter/foundation.dart' show ValueListenable;
 import 'package:rxdart/rxdart.dart' show ErrorAndStackTrace, ValueStream;
+import 'package:rxdart/src/utils/value_wrapper.dart';
 
 import 'listenable_to_stream.dart';
 import 'streamx.dart';
@@ -25,46 +26,42 @@ extension ValueListenableToValueStream<T> on ValueListenable<T> {
 class ValueListenableStream<T> extends Stream<T> implements ValueStream<T> {
   final ValueListenable<T> _valueListenable;
   final bool _replayValue;
-  Stream<T> _stream;
+  Stream<T>? _stream;
+
+  T _getValue([void _]) => _valueListenable.value;
 
   /// Construct a [ValueListenableStream] from [ValueListenable].
   ValueListenableStream(this._valueListenable, this._replayValue);
 
-  @override
-  bool get isBroadcast => false;
-
-  @override
-  ErrorAndStackTrace get errorAndStackTrace => null;
-
-  @override
-  bool get hasError => false;
-
-  @override
-  bool get hasValue => true;
-
-  @override
-  T get value => _valueListenable.value;
-
   @override
   StreamSubscription<T> listen(
-    void Function(T) onData, {
-    Function onError,
-    void Function() onDone,
-    bool cancelOnError,
+    void Function(T)? onData, {
+    Function? onError,
+    void Function()? onDone,
+    bool? cancelOnError,
   }) {
-    final getValue = ([void _]) => _valueListenable.value;
-
     if (_replayValue) {
-      _stream ??= _valueListenable.toStream().map(getValue).startWith(getValue);
+      _stream ??=
+          _valueListenable.toStream().map(_getValue).startWith(_getValue);
     } else {
-      _stream ??= _valueListenable.toStream().map(getValue);
+      _stream ??= _valueListenable.toStream().map(_getValue);
     }
 
-    return _stream.listen(
+    return _stream!.listen(
       onData,
       onError: onError,
       onDone: onDone,
       cancelOnError: cancelOnError,
     );
   }
+
+  @override
+  bool get isBroadcast => false;
+
+  @override
+  Never get errorAndStackTrace =>
+      throw StateError('This Stream always has no error!');
+
+  @override
+  ValueWrapper<T> get valueWrapper => ValueWrapper(_getValue());
 }

pubspec.lock

@@ -94,7 +94,7 @@ packages:
       name: rxdart
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "0.25.0"
+    version: "0.26.0-nullsafety.1"
   sky_engine:
     dependency: transitive
     description: flutter

pubspec.yaml

@@ -7,11 +7,11 @@ repository: https://github.com/Flutter-Dart-Open-Source/listenable_stream.git
 issue_tracker: https://github.com/Flutter-Dart-Open-Source/listenable_stream/issues
 
 environment:
-  sdk: ">=2.7.0 <3.0.0"
+  sdk: ">=2.12.0-0 <3.0.0"
   flutter: ">=1.17.0"
 
 dependencies:
-  rxdart: ^0.25.0
+  rxdart: ^0.26.0-nullsafety.1
   flutter:
     sdk: flutter
 

test/listenable_stream_test.dart

@@ -1,6 +1,7 @@
 import 'package:flutter/foundation.dart';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:listenable_stream/listenable_stream.dart';
+import 'package:rxdart/rxdart.dart';
 
 void _isSingleSubscriptionStream(Stream<dynamic> stream) {
   expect(stream.isBroadcast, isFalse);
@@ -59,7 +60,7 @@ void main() {
       final valueNotifier = ValueNotifier(0);
       final stream = valueNotifier.toValueStream(replayValue: true);
 
-      expect(stream.value, 0);
+      expect(stream.requireValue, 0);
       expect(
         stream,
         emitsInOrder([0, 1, 2, 3]),