Add `peek` and `lookAhead` to `StreamQueue`.

test/stream_queue_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 filevents.
 
 import "dart:async";
 
 import "package:async/async.dart";
 import "package:test/test.dart";
 
 import "utils.dart";
@@ skipping 24 matching lines @@
 
       expect(await next, 2);
       expect(controller.hasListener, isTrue);
       expect(controller.isPaused, isTrue);
 
       events.cancel();
       expect(controller.hasListener, isFalse);
     });
   });
 
+  group("lookAhead operation", () {
+    test("as simple list of events", () async {
+      var events = new StreamQueue<int>(createStream());
+      expect(await events.lookAhead(4), [1, 2, 3, 4]);
+      expect(await events.next, 1);
+      expect(await events.lookAhead(2), [2, 3]);
+      expect(await events.take(2), [2, 3]);
+      expect(await events.next, 4);
+      await events.cancel();
+    });
+
+    test("of 0 events", () async {
+      var events = new StreamQueue<int>(createStream());
+      expect(events.lookAhead(0), completion([]));
+      expect(events.next, completion(1));
+      expect(events.lookAhead(0), completion([]));
+      expect(events.next, completion(2));
+      expect(events.lookAhead(0), completion([]));
+      expect(events.next, completion(3));
+      expect(events.lookAhead(0), completion([]));
+      expect(events.next, completion(4));
+      expect(events.lookAhead(0), completion([]));
+      expect(events.lookAhead(5), completion([]));
+      expect(events.next, throwsStateError);
+      await events.cancel();
+    });
+
+    test("with bad arguments throws", () async {
+      var events = new StreamQueue<int>(createStream());
+      expect(() => events.lookAhead(-1), throwsArgumentError);
+      expect(await events.next, 1);  // Did not consume event.
+      expect(() => events.lookAhead(-1), throwsArgumentError);
+      expect(await events.next, 2);  // Did not consume event.
+      await events.cancel();
+    });
+
+    test("of too many arguments", () async {
+      var events = new StreamQueue<int>(createStream());
+      expect(await events.lookAhead(6), [1, 2, 3, 4]);
+      await events.cancel();
+    });
+
+    test("too large later", () async {
+      var events = new StreamQueue<int>(createStream());
+      expect(await events.next, 1);
+      expect(await events.next, 2);
+      expect(await events.lookAhead(6), [3, 4]);
+      await events.cancel();
+    });
+
+    test("error", () async {
+      var events = new StreamQueue<int>(createErrorStream());
+      expect(events.lookAhead(4), throwsA("To err is divine!"));
+      expect(events.take(4), throwsA("To err is divine!"));
+      expect(await events.next, 4);
+      await events.cancel();
+    });
+  });
+
   group("next operation", () {
     test("simple sequence of requests", () async {
       var events = new StreamQueue<int>(createStream());
       for (int i = 1; i <= 4; i++) {
         expect(await events.next, i);
       }
       expect(events.next, throwsStateError);
     });
 
     test("multiple requests at the same time", () async {
@@ skipping 271 matching lines @@
       await flushMicrotasks();
       expect(lastEvent, 3);
 
       var cancelFuture = subscription.cancel();
       expect(controller.hasListener, isFalse);
       cancel.complete(42);
       expect(cancelFuture, completion(42));
     });
   });
 
+  group("peek operation", () {
+    test("peeks one event", () async {
+      var events = new StreamQueue<int>(createStream());
+      expect(await events.peek, 1);
+      expect(await events.next, 1);
+      expect(await events.peek, 2);
+      expect(await events.take(2), [2, 3]);
+      expect(await events.peek, 4);
+      expect(await events.next, 4);
+      // Throws at end.
+      expect(events.peek, throws);
+      await events.cancel();
+    });
+    test("multiple requests at the same time", () async {
+      var events = new StreamQueue<int>(createStream());
+      var result = await Future.wait(
+          [events.peek, events.peek, events.next, events.peek, events.peek]);
+      expect(result, [1, 1, 1, 2, 2]);
+      await events.cancel();
+    });
+    test("sequence of requests with error", () async {
+      var events = new StreamQueue<int>(createErrorStream());
+      expect(await events.next, 1);
+      expect(await events.next, 2);
+      expect(events.peek, throwsA("To err is divine!"));
+      // Error stays in queue.
+      expect(events.peek, throwsA("To err is divine!"));
+      expect(events.next, throwsA("To err is divine!"));
+      expect(await events.next, 4);
+      await events.cancel();
+    });
+  });
+
   group("cancel operation", () {
     test("closes the events, prevents any other operation", () async {
       var events = new StreamQueue<int>(createStream());
       await events.cancel();
+      expect(() => events.lookAhead(1), throwsStateError);
       expect(() => events.next, throwsStateError);
+      expect(() => events.peek, throwsStateError);
       expect(() => events.skip(1), throwsStateError);
       expect(() => events.take(1), throwsStateError);
       expect(() => events.rest, throwsStateError);
       expect(() => events.cancel(), throwsStateError);
     });
 
     test("cancels underlying subscription when called before any event",
         () async {
       var cancelFuture = new Future.value(42);
       var controller = new StreamController<int>(onCancel: () => cancelFuture);
@@ skipping 616 matching lines @@
     controller.add(4);
     await flushMicrotasks();
     controller.close();
   }();
   return controller.stream;
 }
 
 Stream<int> createLongStream(int eventCount) async* {
   for (int i = 0; i < eventCount; i++) yield i;
 }