Fix broadcast bug for Audit and clarify behavior (dart-lang/stream_transform#22)

- Use `fromHandlers` to correctly handle broadcast Streams.
- Clarify in the Doc comment the behavior difference from `debounce`
- Add a test which makes explicit the difference from `debounce` - all
  the other tests also pass with `debounce`.

pkgs/stream_transform/CHANGELOG.md

diff --git a/pkgs/stream_transform/CHANGELOG.md b/pkgs/stream_transform/CHANGELOG.md
index 50c23d8..e595919 100644
--- a/pkgs/stream_transform/CHANGELOG.md
+++ b/pkgs/stream_transform/CHANGELOG.md
@@ -1,7 +1,7 @@
 ## 0.0.6
 
-- Bug Fix: `debounce` correctly adds data to all listeners on a broadcast
-  stream.
+- Bug Fix: Some transformers did not correctly add data to all listeners on
+  broadcast streams. Fixed for `debounce`, and `audit`.
 
 ## 0.0.5
 

pkgs/stream_transform/lib/src/audit.dart

diff --git a/pkgs/stream_transform/lib/src/audit.dart b/pkgs/stream_transform/lib/src/audit.dart
index f4cadff..12bb9d5 100644
--- a/pkgs/stream_transform/lib/src/audit.dart
+++ b/pkgs/stream_transform/lib/src/audit.dart
@@ -3,18 +3,25 @@
 // BSD-style license that can be found in the LICENSE file.
 import 'dart:async';
 
+import 'from_handlers.dart';
+
 /// Creates a StreamTransformer which only emits once per [duration], at the
 /// end of the period.
 ///
-/// Like `throttle`, except it always emits the most recently received event in
-/// a period.  Always introduces a delay of at most [duration].
+/// Always introduces a delay of at most [duration].
+///
+/// Differs from `throttle` in that it always emits the most recently received
+/// event rather than the first in the period.
+///
+/// Differs from `debounce` in that a value will always be emitted after
+/// [duration], the output will not be starved by values coming in repeatedly
+/// within [duration].
 StreamTransformer<T, T> audit<T>(Duration duration) {
   Timer timer;
   bool shouldClose = false;
   T recentData;
 
-  return new StreamTransformer.fromHandlers(
-      handleData: (T data, EventSink<T> sink) {
+  return fromHandlers(handleData: (T data, EventSink<T> sink) {
     recentData = data;
     timer ??= new Timer(duration, () {
       sink.add(recentData);

pkgs/stream_transform/test/audit_test.dart

diff --git a/pkgs/stream_transform/test/audit_test.dart b/pkgs/stream_transform/test/audit_test.dart
index b4e3151..d65fd7f 100644
--- a/pkgs/stream_transform/test/audit_test.dart
+++ b/pkgs/stream_transform/test/audit_test.dart
@@ -14,6 +14,7 @@
       bool valuesCanceled;
       bool isDone;
       List errors;
+      Stream transformed;
       StreamSubscription subscription;
 
       void setUpStreams(StreamTransformer transformer) {
@@ -25,8 +26,8 @@
         emittedValues = [];
         errors = [];
         isDone = false;
-        subscription = values.stream
-            .transform(transformer)
+        transformed = values.stream.transform(transformer);
+        subscription = transformed
             .listen(emittedValues.add, onError: errors.add, onDone: () {
           isDone = true;
         });
@@ -76,6 +77,28 @@
           await new Future.delayed(const Duration(milliseconds: 10));
           expect(isDone, true);
         });
+
+        test('does not starve output if many values come closer than duration',
+            () async {
+          values.add(1);
+          await new Future.delayed(const Duration(milliseconds: 3));
+          values.add(2);
+          await new Future.delayed(const Duration(milliseconds: 3));
+          values.add(3);
+          expect(emittedValues, [2]);
+        });
+
+        if (streamType == 'broadcast') {
+          test('multiple listeners all get values', () async {
+            var otherValues = [];
+            transformed.listen(otherValues.add);
+            values.add(1);
+            values.add(2);
+            await new Future.delayed(const Duration(milliseconds: 10));
+            expect(emittedValues, [2]);
+            expect(otherValues, [2]);
+          });
+        }
       });
     });
   }