Add reactive-inspired stream transformers:

test/debounce_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.

[nweiz, 2016/03/01 02:10:04]

Nit: I don't think this is necessary.

[Lasse Reichstein Nielsen, 2016/03/01 16:51:17]

True, that's a left-over from the original file.

+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 debounced = stream.transform(
+        new Debounce<int>(ms * 100, createStopwatch: createStopwatch));
+    expect(debounced.toList(), completion([0, 280, 380, 580]));
+  });
+
+  test("empty source", () async {
+    var stream = new Stream<int>.empty();
+    var debounced = stream.transform(
+        new Debounce<int>(ms * 100, createStopwatch: createStopwatch));
+    expect(debounced.toList(), completion([]));
+  });
+
+  test("omit last event", () async {
+    var stream =
+        createTimerStream([0, 30, 60, 90, 180, 270]);
+    var debounced = stream.transform(
+        new Debounce<int>(ms * 100, createStopwatch: createStopwatch));
+    expect(debounced.toList(), completion([0]));
+  });
+
+  test("stops at first error", () async {
+    var stream =
+        createTimerStream([0, 30, 60, 180, 280], errors: [180]);
+    var debounced = stream.transform(
+        new Debounce<int>(ms * 100, createStopwatch: createStopwatch));
+    // Stops at error.
+    Result r(v) => new Result.value(v);

[nweiz, 2016/03/01 02:10:04]

Defining an anonymous function for this only to use it once seems like overkill.

+    expect(Result.captureStream(debounced).toList(), completion([r(0), error]));
+  });
+}