Add new features to package:async.

test/stream_events_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_events_test;
+
+import "dart:async";
+
+import "package:async/async.dart" show StreamEvents;
+import "package:test/test.dart";
+
+main() {

[Søren Gjesse, 2015/06/11 07:57:07]

Add check of isDone.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:23]

Acknowledged.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:25]

Done.

+  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);

[nweiz, 2015/06/12 01:24:27]

throwsStateError. In general, check that the exceptions have the types you
expect throughout this file (or in the case of thrown strings, that they're the
strings you expect).

[Lasse Reichstein Nielsen, 2015/06/15 15:46:25]

Done.

+    });
+
+    test("request mutiple", () async {

[nweiz, 2015/06/12 01:24:28]

"mutiple" -> "multiple"

[Lasse Reichstein Nielsen, 2015/06/15 15:46:24]

Done.

+      var e = new StreamEvents<int>(createStream());
+      var r = await Future.wait(new Iterable.generate(4, (n) => e.next));

[Søren Gjesse, 2015/06/11 07:57:07]

n -> _

[nweiz, 2015/06/12 01:24:27]

Explicitly writing out the [e.next] calls would be more readable here, in my
opinion.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:23]

Acknowledged.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:25]

It's also shorter. Until I start using longer names for "e".

+      expect(r, [1, 2, 3, 4]);
+      await e.close();

[nweiz, 2015/06/12 01:24:27]

Is there any reason to explicitly close the stream here? It and all its
subscribers should get garbage-collected when it goes out of scope.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:25]

True. It's good style, and I'd like to encourage it, but in this case it's
probably not necessary.

+    });
+
+    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();
+    });
+  });

[nweiz, 2015/06/12 01:24:28]

Test that the underlying subscription is paused when there are no [next] calls
pending.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:25]

Done.

+
+  group("skip", () {
+    test("simple-2", () async {

[nweiz, 2015/06/12 01:24:27]

This name isn't very descriptive.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:25]

Done.

+      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("negative/bad throws", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(() => e.skip(-1), throws);
+      expect(() => e.skip("not a number"), throws);

[nweiz, 2015/06/12 01:24:27]

I don't think it's necessary to test type errors. That's the language's job to
enforce, and it would be a huge pain to test everywhere.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:24]

Done.

+      expect(await e.next, 1);  // Did not consume event.
+      expect(() => e.skip(-1), throws);
+      expect(() => e.skip("not a number"), throws);
+      expect(await e.next, 2);  // Did not consume event.
+      await e.close();
+    });
+
+    test("0 works", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(e.skip(0), completion(0));
+      expect(e.next, completion(1));
+      expect(e.skip(0), completion(0));
+      expect(e.next, completion(2));
+      expect(e.skip(0), completion(0));
+      expect(e.next, completion(3));
+      expect(e.skip(0), completion(0));
+      expect(e.next, completion(4));
+      expect(e.skip(0), completion(0));
+      expect(e.skip(5), completion(5));
+      expect(e.next, 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 works", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(e.take(0), completion([]));
+      expect(e.next, completion(1));
+      expect(e.take(0), completion([]));
+      expect(e.next, completion(2));
+      expect(e.take(0), completion([]));
+      expect(e.next, completion(3));
+      expect(e.take(0), completion([]));
+      expect(e.next, completion(4));
+      expect(e.take(0), completion([]));
+      expect(e.take(5), completion([]));
+      expect(e.next, throws);
+      await e.close();
+    });
+
+    test("negative/bad throws", () async {
+      var e = new StreamEvents<int>(createStream());
+      expect(() => e.take(-1), throws);
+      expect(() => e.take("not a number"), throws);
+      expect(await e.next, 1);  // Did not consume event.
+      expect(() => e.take(-1), throws);
+      expect(() => e.take("not a number"), throws);
+      expect(await e.next, 2);  // Did not consume event.
+      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("after 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(e.next, throws);
+      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);
+    });
+  });

[nweiz, 2015/06/12 01:24:27]

Also test forwarding pause/cancel to the underlying subscription.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:25]

Done.

+
+  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);
+    });
+  });

[nweiz, 2015/06/12 01:24:27]

Ensure that it cancels the underlying subscription.

[Lasse Reichstein Nielsen, 2015/06/15 15:46:24]

Done.

+
+  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];

[Søren Gjesse, 2015/06/11 07:57:07]

assert(tests.length * tests.length + 20 == 200) or pass 'tests.length *
tests.length + 20' as an argument to createLongStream.

+    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)));

[nweiz, 2015/06/12 01:24:27]

It's hard for me to follow what's going on here... can you add some comments?

[Lasse Reichstein Nielsen, 2015/06/15 15:46:25]

Done.

+  });
+}
+
+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{

[Søren Gjesse, 2015/06/11 07:57:07]

Space after async.

[Lasse Reichstein Nielsen, 2015/06/12 13:04:23]

Done.

+    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;
+}