Add new features to package:async.

test/stream_completer_test.dart

+// Copyright (c) 2015, 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.
+
+library async.stream_completer_test;
+
+import "dart:async";
+
+import "package:async/async.dart" show StreamCompleter;
+import "package:test/test.dart";
+
+main() {
+  test("early link", () async {
+    var c = new StreamCompleter();
+    var s = c.stream;
+    c.setSourceStream(createStream());
+    var r = [];
+    await s.listen(r.add).asFuture();
+    expect(r, [1, 2, 3, 4]);
+  });
+
+  test("later link", () async {
+    var c = new StreamCompleter();
+    var s = c.stream;
+    var r = [];
+    Future done = s.listen(r.add).asFuture();
+    await sleep(250);
+    c.setSourceStream(createStream());
+    await done;
+    expect(r, [1, 2, 3, 4]);
+  });
+
+  test("early cancel", () async {
+    var c = new StreamCompleter();
+    var s = c.stream;
+    var sub = s.listen(null);
+    sub.pause();  // Should be ignored.
+    sub.cancel();
+    c.setSourceStream(new UnusableStream());  // Doesn't throw.
+  });
+
+  test("early pause", () async {
+    var c = new StreamCompleter();
+    var s = c.stream;
+    var r = [];
+    var sub = s.listen(r.add);
+    Future done = sub.asFuture();
+    sub.pause();
+    c.setSourceStream(createStream());
+    await sleep(250);
+    expect(r, []);
+    sub.resume();
+    await done;
+    expect(r, [1, 2, 3, 4]);
+  });
+
+  test("multi-pause", () async {
+    var c = new StreamCompleter();
+    var s = c.stream;
+    var r = [];
+    var sub = s.listen(r.add);
+    Future done = sub.asFuture();
+    sub.pause();
+    sub.pause();
+    sub.pause();
+    c.setSourceStream(createStream());
+    for (int i = 0; i < 3; i++) {
+      await sleep(50);
+      expect(r, []);
+      sub.resume();
+    }
+    await done;
+    expect(r, [1, 2, 3, 4]);
+  });
+}
+
+const MS = const Duration(milliseconds: 1);
+Future sleep(int n) => new Future.delayed(MS * n);
+
+Stream<int> createStream() async* {
+  yield 1;
+  await sleep(20);
+  yield 2;
+  await sleep(10);
+  yield 3;
+  await sleep(15);
+  yield 4;
+}
+
+class UnusableStream extends Stream {
+  listen(onData, {onError, onDone, cancelOnError}) {
+    throw new UnimplementedError("Gotcha!");
+  }
+}