Use sync forwarders for `switchLatest` (dart-lang/stream_transform#30)

Towards dart-lang/stream_transform#24

- Don't return a Future from `onCancel` when the values were already
  done.
- Use `sync: true`
- Drop unnecessary `await new Future(() {})`s
- Set onPause, onResume, and onCancel inside onListen.

pkgs/stream_transform/lib/src/switch.dart

diff --git a/pkgs/stream_transform/lib/src/switch.dart b/pkgs/stream_transform/lib/src/switch.dart
index be9eeeb..b7fc110 100644
--- a/pkgs/stream_transform/lib/src/switch.dart
+++ b/pkgs/stream_transform/lib/src/switch.dart
@@ -32,58 +32,54 @@
 
   @override
   Stream<T> bind(Stream<Stream<T>> outer) {
-    StreamController<T> controller;
-    if (outer.isBroadcast) {
-      controller = new StreamController<T>.broadcast();
-    } else {
-      controller = new StreamController<T>();
-    }
-    StreamSubscription<T> innerSubscription;
+    var controller = outer.isBroadcast
+        ? new StreamController<T>.broadcast(sync: true)
+        : new StreamController<T>(sync: true);
+
     StreamSubscription<Stream<T>> outerSubscription;
+
     controller.onListen = () {
+      if (outerSubscription != null) return;
+
+      StreamSubscription<T> innerSubscription;
       var outerStreamDone = false;
-      var innerStreamDone = false;
-      outerSubscription = outer.listen((innerStream) {
-        innerSubscription?.cancel();
-        innerSubscription = innerStream.listen(controller.add);
-        innerSubscription.onDone(() {
-          innerStreamDone = true;
-          if (outerStreamDone) {
-            controller.close();
-          }
-        });
-        innerSubscription.onError(controller.addError);
-      });
-      outerSubscription.onDone(() {
-        outerStreamDone = true;
-        if (innerStreamDone) {
-          controller.close();
-        }
-      });
-      outerSubscription.onError(controller.addError);
-    };
 
-    cancelSubscriptions() => Future.wait([
-          innerSubscription?.cancel() ?? new Future.value(),
-          outerSubscription?.cancel() ?? new Future.value()
-        ]);
-
-    if (!outer.isBroadcast) {
-      controller.onPause = () {
-        innerSubscription?.pause();
-        outerSubscription?.pause();
-      };
-      controller.onResume = () {
-        innerSubscription?.resume();
-        outerSubscription?.resume();
-      };
-      controller.onCancel = () => cancelSubscriptions();
-    } else {
+      outerSubscription = outer.listen(
+          (innerStream) {
+            innerSubscription?.cancel();
+            innerSubscription = innerStream.listen(controller.add,
+                onError: controller.addError, onDone: () {
+              innerSubscription = null;
+              if (outerStreamDone) controller.close();
+            });
+          },
+          onError: controller.addError,
+          onDone: () {
+            outerStreamDone = true;
+            if (innerSubscription == null) controller.close();
+          });
+      if (!outer.isBroadcast) {
+        controller.onPause = () {
+          innerSubscription?.pause();
+          outerSubscription.pause();
+        };
+        controller.onResume = () {
+          innerSubscription?.resume();
+          outerSubscription.resume();
+        };
+      }
       controller.onCancel = () {
-        if (controller.hasListener) return new Future.value();
-        return cancelSubscriptions();
+        var toCancel = <StreamSubscription>[];
+        if (!outerStreamDone) toCancel.add(outerSubscription);
+        if (innerSubscription != null) {
+          toCancel.add(innerSubscription);
+        }
+        outerSubscription = null;
+        innerSubscription = null;
+        if (toCancel.isEmpty) return null;
+        return Future.wait(toCancel.map((s) => s.cancel()));
       };
-    }
+    };
     return controller.stream;
   }
 }

pkgs/stream_transform/test/switch_test.dart

diff --git a/pkgs/stream_transform/test/switch_test.dart b/pkgs/stream_transform/test/switch_test.dart
index fbae7d7..91f10c8 100644
--- a/pkgs/stream_transform/test/switch_test.dart
+++ b/pkgs/stream_transform/test/switch_test.dart
@@ -82,12 +82,9 @@
           await new Future(() {});
 
           await outer.close();
-
           expect(isDone, false);
 
           await second.close();
-
-          await new Future(() {});
           expect(isDone, true);
         });
 
@@ -97,11 +94,9 @@
           outer.add(first.stream);
           await new Future(() {});
           await first.close();
-          await new Future(() {});
           expect(isDone, false);
 
           await outer.close();
-          await new Future(() {});
           expect(isDone, true);
         });