Use sync forwarders in `fromHandlers` (dart-lang/stream_transform#27) Towards dart-lang/stream_transform#24 - Switch to sync StreamControllers - Set onPause, onResume, and onCancel in the onListen callback - Don't wrap pause and resume callbacks - they are guaranteed to not be called outside of the subscription lifetime - Only cancel the subscription when the values stream is not done - in stream closure path the onCancel is called but does not need to return a future. - Drop the now unnecessary `await new Future(() {})` from tests.

diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
index 7f1aedf..b684adb 100644
--- a/pkgs/stream_transform/CHANGELOG.md
+++ b/pkgs/stream_transform/CHANGELOG.md

@@ -4,6 +4,7 @@
   broadcast streams. Fixed for `throttle`, `debounce`, and `audit`.
 - Bug Fix: Only call the `tap` data callback once per event rather than once per
   listener.
+- Use sync `StreamControllers` for forwarding where possible.
 
 ## 0.0.5
 

diff --git a/pkgs/stream_transform/lib/src/from_handlers.dart b/pkgs/stream_transform/lib/src/from_handlers.dart
index 315de96..1303584 100644
--- a/pkgs/stream_transform/lib/src/from_handlers.dart
+++ b/pkgs/stream_transform/lib/src/from_handlers.dart

@@ -44,35 +44,31 @@
 
   @override
   Stream<T> bind(Stream<S> values) {
-    StreamController<T> controller;
-    if (values.isBroadcast) {
-      controller = new StreamController<T>.broadcast();
-    } else {
-      controller = new StreamController<T>();
-    }
+    var controller = values.isBroadcast
+        ? new StreamController<T>.broadcast(sync: true)
+        : new StreamController<T>(sync: true);
+
     StreamSubscription<S> subscription;
     controller.onListen = () {
-      if (subscription != null) {
-        return;
-      }
+      if (subscription != null) return;
+      bool valuesDone = false;
       subscription = values.listen((value) => _handleData(value, controller),
           onError: (error, stackTrace) {
         _handleError(error, stackTrace, controller);
       }, onDone: () {
+        valuesDone = true;
         _handleDone(controller);
       });
-    };
-    if (!values.isBroadcast) {
-      controller.onPause = () => subscription?.pause();
-      controller.onResume = () => subscription?.resume();
-    }
-    controller.onCancel = () {
-      if (controller.hasListener || subscription == null) {
-        return new Future.value();
+      if (!values.isBroadcast) {
+        controller.onPause = subscription.pause;
+        controller.onResume = subscription.resume;
       }
-      var toCancel = subscription;
-      subscription = null;
-      return toCancel.cancel();
+      controller.onCancel = () {
+        var toCancel = subscription;
+        subscription = null;
+        if (!valuesDone) return toCancel.cancel();
+        return null;
+      };
     };
     return controller.stream;
   }

diff --git a/pkgs/stream_transform/test/audit_test.dart b/pkgs/stream_transform/test/audit_test.dart
index d65fd7f..a3b2fef 100644
--- a/pkgs/stream_transform/test/audit_test.dart
+++ b/pkgs/stream_transform/test/audit_test.dart

@@ -61,9 +61,9 @@
 
         test('waits for pending value to close', () async {
           values.add(1);
-          await new Future.delayed(const Duration(milliseconds: 10));
           await values.close();
-          await new Future(() {});
+          expect(isDone, false);
+          await new Future.delayed(const Duration(milliseconds: 10));
           expect(isDone, true);
         });
 
@@ -71,7 +71,6 @@
           values.add(1);
           await new Future.delayed(const Duration(milliseconds: 10));
           values.add(2);
-          await new Future(() {});
           await values.close();
           expect(isDone, false);
           await new Future.delayed(const Duration(milliseconds: 10));

diff --git a/pkgs/stream_transform/test/from_handlers_test.dart b/pkgs/stream_transform/test/from_handlers_test.dart
index 268b55e..fe9e916 100644
--- a/pkgs/stream_transform/test/from_handlers_test.dart
+++ b/pkgs/stream_transform/test/from_handlers_test.dart

@@ -55,7 +55,6 @@
 
       test('forwards done', () async {
         await values.close();
-        await new Future(() {});
         expect(isDone, true);
       });
 
@@ -103,7 +102,6 @@
 
       test('forwards done', () async {
         await values.close();
-        await new Future(() {});
         expect(isDone, true);
         expect(isDone2, true);
       });

diff --git a/pkgs/stream_transform/test/throttle_test.dart b/pkgs/stream_transform/test/throttle_test.dart
index 3647cb4..10a34fd 100644
--- a/pkgs/stream_transform/test/throttle_test.dart
+++ b/pkgs/stream_transform/test/throttle_test.dart

@@ -63,9 +63,7 @@
           values.add(1);
           await new Future.delayed(const Duration(milliseconds: 10));
           values.add(2);
-          await new Future(() {});
           await values.close();
-          await new Future(() {});
           expect(isDone, true);
         });