Add new features to package:async.

test/stream_events_test.dart

Index: test/stream_events_test.dart
diff --git a/test/stream_events_test.dart b/test/stream_events_test.dart
new file mode 100644
index 0000000000000000000000000000000000000000..b73dafec6c4d5edff973524c548eca0d73d292a2
--- /dev/null
+++ b/test/stream_events_test.dart
@@ -0,0 +1,223 @@
+// 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_events_test;
+
+import "dart:async";
+
+import "package:async/async.dart" show StreamEvents;
+import "package:test/test.dart";
+
+main() {
+  group("next", () {
+    test("simple", () async {
+      var e = new StreamEvents<int>(createStream());
+      for (int i = 1; i <= 4; i++) {
+        expect(await e.next, i);
+      }
+      expect(e.next, throws);
+    });
+
+    test("request mutiple", () async {
+      var e = new StreamEvents<int>(createStream());
+      var r = await Future.wait(new Iterable.generate(4, (n) => e.next));
+      expect(r, [1, 2, 3, 4]);
+      await e.close();
+    });
+
+    test("error", () async {
+      var e = new StreamEvents<int>(createErrorStream());
+      expect(await e.next, 1);
+      expect(await e.next, 2);
+      expect(e.next, throws);
+      expect(await e.next, 4);
+      await e.close();
+    });
+  });
+
+  group("skip", () {
+    test("simple-2", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.next, 1);
+      expect(await e.skip(2), 0);
+      expect(await e.next, 4);
+      await e.close();
+    });
+
+    test("0 throws", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(() => e.skip(0), throws);
+      await e.close();
+    });
+
+    test("too large", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.skip(6), 2);
+      await e.close();
+    });
+
+    test("too large later", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.next, 1);
+      expect(await e.next, 2);
+      expect(await e.skip(6), 4);
+      await e.close();
+    });
+
+    test("error", () async {
+      var e = new StreamEvents<int>(createErrorStream());
+      expect(e.skip(4), throws);
+      expect(await e.next, 4);
+      await e.close();
+    });
+  });
+
+  group("take", () {
+    test("simple-2", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.next, 1);
+      expect(await e.take(2), [2, 3]);
+      expect(await e.next, 4);
+      await e.close();
+    });
+
+    test("0 throws", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(() => e.take(0), throws);
+      await e.close();
+    });
+
+    test("too large", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.take(6), [1, 2, 3, 4]);
+      await e.close();
+    });
+
+    test("too large later", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.next, 1);
+      expect(await e.next, 2);
+      expect(await e.take(6), [3, 4]);
+      await e.close();
+    });
+
+    test("error", () async {
+      var e = new StreamEvents<int>(createErrorStream());
+      expect(e.take(4), throws);
+      expect(await e.next, 4);
+      await e.close();
+    });
+  });
+
+  group("rest", () {
+    test("simple", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.next, 1);
+      expect(await e.rest.toList(), [2, 3, 4]);
+    });
+
+    test("at start", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.rest.toList(), [1, 2, 3, 4]);
+    });
+
+    test("at end", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(await e.next, 1);
+      expect(await e.next, 2);
+      expect(await e.next, 3);
+      expect(await e.next, 4);
+      expect(await e.rest.toList(), []);
+    });
+
+    test("error", () async {
+      var e = new StreamEvents<int>(createErrorStream());
+      expect(await e.next, 1);
+      var r = e.rest;
+      var e2 = new StreamEvents(r);
+      expect(await e2.next, 2);
+      expect(e2.next, throws);
+      expect(await e2.next, 4);
+    });
+
+    test("closes", () async {
+      var e = new StreamEvents<int>(createStream());
+      var stream = e.rest;
+      expect(() => e.next, throws);
+      expect(() => e.skip(1), throws);
+      expect(() => e.take(1), throws);
+      expect(() => e.rest, throws);
+      expect(() => e.close(), throws);
+    });
+  });
+
+  group("close", () {
+    test("anything throws", () async {
+      var e = new StreamEvents<int>(createStream());
+      await e.close();
+      expect(() => e.next, throws);
+      expect(() => e.skip(1), throws);
+      expect(() => e.take(1), throws);
+      expect(() => e.rest, throws);
+      expect(() => e.close(), throws);
+    });
+  });
+
+  test("mixed", () async {
+    var e = new StreamEvents<int>(createLongStream());
+    nextTest(n) {
+      for (int i = 0; i < 10; i++) {
+        expect(e.next, completion(n + i));
+      }
+    }
+    skipTest(n) {
+      expect(e.skip(10), completion(0));
+    }
+    takeTest(n) {
+      expect(e.take(10), completion(new List.generate(10, (i) => n + i)));
+    }
+    var tests = [nextTest, skipTest, takeTest];
+    int counter = 0;
+    for (int i = 0; i < tests.length; i++) {
+      for (int j = 0; j < tests.length; j++) {
+        tests[i](counter);
+        tests[j](counter + 10);
+        counter += 20;
+      }
+    }
+    expect(e.rest.toList(),
+           completion(new List.generate(20, (i) => counter + i)));
+  });
+}
+
+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;
+}
+
+Stream<int> createErrorStream() {
+  StreamController controller = new StreamController<int>();
+  () async{
+    controller.add(1);
+    await sleep(20);
+    controller.add(2);
+    await sleep(10);
+    controller.addError("To err is divine!");
+    await sleep(15);
+    controller.add(4);
+  }();
+  return controller.stream;
+}
+
+Stream<int> createLongStream() async* {
+  for (int i = 0; i < 200; i++) yield i;
+}