♻️ Refactoring code.

Author: hoc081098

lib/src/listenable_to_stream.dart

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

lib/src/streamx.dart

@@ -0,0 +1,66 @@
+import 'dart:async'
+    show MultiStreamController, Stream, StreamController, StreamSubscription;
+
+/// @private
+/// Prepends a value to the source [Stream].
+extension StartWithExtension<T> on Stream<T> {
+  /// @private
+  /// Prepends a value to the source [Stream].
+  Stream<T> startWith(T Function() seededProvider) {
+    MultiStreamController<T> controller;
+    StreamSubscription<T> subscription;
+
+    final listenUpStream = () => listen(
+          controller.addSync,
+          onError: controller.addErrorSync,
+          onDone: () {
+            subscription = null;
+            controller.closeSync();
+          },
+        );
+
+    final onListen = (MultiStreamController<T> c) {
+      if (controller != null) {
+        return;
+      }
+
+      controller = c;
+      controller.addSync(seededProvider());
+
+      subscription = listenUpStream();
+      controller.onCancel = () {
+        subscription?.cancel();
+        subscription = null;
+      };
+    };
+
+    return Stream.multi(onListen, isBroadcast: false).toSingleSubscription();
+  }
+}
+
+/// @private
+/// Convert a [Stream] to Single-Subscription [Stream].
+extension ToSingleSubscriptionStreamExtension<T> on Stream<T> {
+  /// @private
+  /// Convert a [Stream] to Single-Subscription [Stream].
+  Stream<T> toSingleSubscription() {
+    StreamController<T> controller;
+    StreamSubscription<T> subscription;
+
+    controller = StreamController<T>(
+      sync: true,
+      onListen: () {
+        subscription = listen(
+          controller.add,
+          onError: controller.addError,
+          onDone: controller.close,
+        );
+      },
+      onPause: () => subscription.pause(),
+      onResume: () => subscription.resume(),
+      onCancel: () => subscription.cancel(),
+    );
+
+    return controller.stream;
+  }
+}

lib/src/value_listenable_to_value_stream.dart

@@ -1,10 +1,10 @@
-import 'dart:async'
-    show MultiStreamController, Stream, StreamController, StreamSubscription;
+import 'dart:async' show Stream, StreamSubscription;
 
 import 'package:flutter/foundation.dart' show ValueListenable;
 import 'package:rxdart/rxdart.dart' show ErrorAndStackTrace, ValueStream;
 
 import 'listenable_to_stream.dart';
+import 'streamx.dart';
 
 /// Convert this [ValueListenable] to a [ValueStream].
 /// The returned [ValueStream] is a Single-Subscription [Stream].
@@ -68,60 +68,3 @@ class ValueListenableStream<T> extends Stream<T> implements ValueStream<T> {
     );
   }
 }
-
-/// TODO
-extension StartWithExtension<T> on Stream<T> {
-  /// TODO
-  Stream<T> startWith(T Function() seededProvider) {
-    MultiStreamController<T> multiController;
-    StreamSubscription<T> upstreamSubscription;
-
-    final listenUpStream = () => listen(
-          multiController.addSync,
-          onError: multiController.addErrorSync,
-          onDone: () {
-            upstreamSubscription = null;
-            multiController.closeSync();
-          },
-        );
-
-    final onListen = (MultiStreamController<T> c) {
-      if (multiController != null) {
-        return;
-      }
-
-      multiController = c;
-      multiController.addSync(seededProvider());
-
-      upstreamSubscription = listenUpStream();
-      multiController.onCancel = () {
-        upstreamSubscription?.cancel();
-        upstreamSubscription = null;
-      };
-    };
-
-    return Stream.multi(onListen, isBroadcast: false).toSingleSubscription();
-  }
-
-  /// TODO
-  Stream<T> toSingleSubscription() {
-    StreamController<T> controller;
-    StreamSubscription<T> subscription;
-
-    controller = StreamController<T>(
-      sync: true,
-      onListen: () {
-        subscription = listen(
-          controller.add,
-          onError: controller.addError,
-          onDone: controller.close,
-        );
-      },
-      onPause: () => subscription.pause(),
-      onResume: () => subscription.resume(),
-      onCancel: () => subscription.cancel(),
-    );
-
-    return controller.stream;
-  }
-}