Add reactive-inspired stream transformers:

test/throttle_test.dart

+// Copyright (c) 2016, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+// Test stream transformers.
+import "dart:async";
+
+import "package:async/async.dart";
+import "package:test/test.dart";
+import "package:clock/clock.dart";
+
+const ms = const Duration(milliseconds: 1);
+
+void main() {
+  var currentTime;
+  var createStopwatch;
+  var error = new Result.error("BAD", StackTrace.current);
+
+  setUp(() {
+    currentTime = Duration.ZERO;
+    createStopwatch = new Clock(elapsed: () => currentTime).getStopwatch;
+  });
+
+  /// Stream with events at the specified times.
+  Stream createTimerStream(List<int> eventTimes,
+                           {List errors: const[]}) async* {
+    for (var time in eventTimes) {
+      currentTime = ms * time;
+      if (errors.contains(time)) {
+        yield* error.asFuture.asStream();
+      } else {
+        yield time;
+      }
+      // Wait for the event to propagate before changing timer again.
+      await new Future.delayed(Duration.ZERO);
+    }
+  }
+
+  test("expected behavior", () async {
+    var stream =
+        createTimerStream([0, 30, 60, 90, 180, 280, 380, 479, 480, 580]);
+    var throttled = stream.transform(
+        new Throttle<int>(ms * 100, createStopwatch: createStopwatch));
+    expect(throttled.toList(), completion([0, 180, 280, 380, 480, 580]));
+  });
+
+  test("empty source", () async {
+    var stream = new Stream<int>.empty();
+    var throttled = stream.transform(
+        new Throttle<int>(ms * 100, createStopwatch: createStopwatch));
+    expect(throttled.toList(), completion([]));
+  });
+
+  test("omit last event", () async {
+    var stream =
+        createTimerStream([0, 30, 60, 90, 180, 270]);
+    var throttled = stream.transform(
+        new Throttle<int>(ms * 100, createStopwatch: createStopwatch));
+    expect(throttled.toList(), completion([0, 180]));
+  });
+
+  test("stops at first error", () async {
+    var stream =
+        createTimerStream([0, 30, 60, 120, 180, 280], errors: [180]);
+    var throttled = stream.transform(
+        new Throttle<int>(ms * 100, createStopwatch: createStopwatch));
+    // Stops at error.
+    Result r(v) => new Result.value(v);
+    expect(Result.captureStream(throttled).toList(),
+           completion([r(0), r(120), error]));
+  });
+}