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

lib/src/stream_queue.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.
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'package:collection/collection.dart';
 
 import "cancelable_operation.dart";
@@ skipping 113 matching lines @@
   /// one events.
   Future<bool> get hasNext {
     if (!_isClosed) {
       var hasNextRequest = new _HasNextRequest();
       _addRequest(hasNextRequest);
       return hasNextRequest.future;
     }
     throw _failClosed();
   }
 
+
+  /// Returns the next [count] data events without consuming them.
+  ///
+  /// Works like [take] except that the events are left in the queue.
+  /// If one of the next [count] events is an error, the returned future
+  /// completes with this error, and the error is still left in the queue.
+  Future<List<T>> lookAhead(int count) {
+    RangeError.checkNotNegative(count, "count");
+    if (!_isClosed) {
+      var request = new _LookAheadRequest<T>(count);
+      _addRequest(request);
+      return request.future;
+    }
+    throw _failClosed();
+  }
+
   /// Requests the next (yet unrequested) event from the stream.
   ///
   /// When the requested event arrives, the returned future is completed with
   /// the event.
   /// If the event is a data event, the returned future completes
   /// with its value.
   /// If the event is an error event, the returned future completes with
   /// its error and stack trace.
   /// If the stream closes before an event arrives, the returned future
   /// completes with a [StateError].
   ///
   /// It's possible to have several pending [next] calls (or other requests),
   /// and they will be completed in the order they were requested, by the
   /// first events that were not consumed by previous requeusts.
   Future<T> get next {
     if (!_isClosed) {
       var nextRequest = new _NextRequest<T>();
       _addRequest(nextRequest);
       return nextRequest.future;
     }
     throw _failClosed();
   }
 
+  /// Returns the next (yet unrequested) event from the stream.
+  ///
+  /// Like [next] except that the event is not consumed.
+  /// If the next event is an error event, it also stays in the queue.
+  Future<T> get peek {
+    if (!_isClosed) {
+      var nextRequest = new _PeekRequest<T>();
+      _addRequest(nextRequest);
+      return nextRequest.future;
+    }
+    throw _failClosed();
+  }
+
   /// Returns a stream of all the remaning events of the source stream.
   ///
   /// All requested [next], [skip] or [take] operations are completed
   /// first, and then any remaining events are provided as events of
   /// the returned stream.
   ///
   /// Using `rest` closes this stream queue. After getting the
   /// `rest` the caller may no longer request other events, like
   /// after calling [cancel].
   Stream<T> get rest {
@@ skipping 15 matching lines @@
   ///
   /// If an error occurs before `count` data events have been skipped,
   /// the returned future completes with that error instead.
   ///
   /// If the stream closes before `count` data events,
   /// the remaining unskipped event count is returned.
   /// If the returned future completes with the integer `0`,
   /// then all events were succssfully skipped. If the value
   /// is greater than zero then the stream ended early.
   Future<int> skip(int count) {
-    if (count < 0) throw new RangeError.range(count, 0, null, "count");
+    RangeError.checkNotNegative(count, "count");
     if (!_isClosed) {
       var request = new _SkipRequest(count);
       _addRequest(request);
       return request.future;
     }
     throw _failClosed();
   }
 
   /// Requests the next [count] data events as a list.
   ///
   /// The [count] must be non-negative.
   ///
   /// Equivalent to calling [next] `count` times and
   /// storing the data values in a list.
   ///
   /// If an error occurs before `count` data events has
   /// been collected, the returned future completes with
   /// that error instead.
   ///
   /// If the stream closes before `count` data events,
   /// the returned future completes with the list
   /// of data collected so far. That is, the returned
   /// list may have fewer than [count] elements.
   Future<List<T>> take(int count) {
-    if (count < 0) throw new RangeError.range(count, 0, null, "count");
+    RangeError.checkNotNegative(count, "count");
     if (!_isClosed) {
       var request = new _TakeRequest<T>(count);
       _addRequest(request);
       return request.future;
     }
     throw _failClosed();
   }
 
   /// Requests a transaction that can conditionally consume events.
   ///
@@ skipping 117 matching lines @@
   /// all previously requested events have been processed, then it cancels the
   /// subscription providing the events.
   ///
   /// If [immediate] is `true`, the source is instead canceled
   /// immediately. Any pending events are completed as though the underlying
   /// stream had closed.
   ///
   /// The returned future completes with the result of calling
   /// `cancel`.
   ///
-  /// After calling `cancel`, no further events can be requested.
-  /// None of [next], [rest], [skip], [take] or [cancel] may be
-  /// called again.
+  /// After calling `cancel`, no further events can be requested,
+  /// so methods like [next] or [peek] may not be called again.
   Future cancel({bool immediate: false}) {
     if (_isClosed) throw _failClosed();
     _isClosed = true;
 
     if (!immediate) {
       var request = new _CancelRequest(this);
       _addRequest(request);
       return request.future;
     }
 
@@ skipping 317 matching lines @@
   _NextRequest();
 
   Future<T> get future => _completer.future;
 
   bool update(QueueList<Result<T>> events, bool isDone) {
     if (events.isNotEmpty) {
       events.removeFirst().complete(_completer);
       return true;
     }
     if (isDone) {
-      var errorFuture =
-          new Future.sync(() => throw new StateError("No elements"));
-      _completer.complete(errorFuture);
+      _completer.completeError(new StateError("No elements"),
+                               StackTrace.current);
       return true;
     }
     return false;
   }
 }
 
+
+/// Request for a [StreamQueue.peek] call.
+///
+/// Completes the returned future when receiving the first event,
+/// and is then complete, but doesn't consume the event.
+class _PeekRequest<T> implements _EventRequest<T> {
+  /// Completer for the future returned by [StreamQueue.next].
+  final _completer = new Completer<T>();
+
+  _PeekRequest();
+
+  Future<T> get future => _completer.future;
+
+  bool update(QueueList<Result<T>> events, bool isDone) {
+    if (events.isNotEmpty) {
+      events.first.complete(_completer);
+      return true;
+    }
+    if (isDone) {
+      _completer.completeError(new StateError("No elements"),
+                               StackTrace.current);
+      return true;
+    }
+    return false;
+  }
+}
+
+
 /// Request for a [StreamQueue.skip] call.
 class _SkipRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by the skip call.
   final _completer = new Completer<int>();
 
   /// Number of remaining events to skip.
   ///
   /// The request [isComplete] when the values reaches zero.
   ///
   /// Decremented when an event is seen.
@@ skipping 17 matching lines @@
       if (event.isError) {
         _completer.completeError(event.asError.error, event.asError.stackTrace);
         return true;
       }
     }
     _completer.complete(_eventsToSkip);
     return true;
   }
 }
 
-/// Request for a [StreamQueue.take] call.
-class _TakeRequest<T> implements _EventRequest<T> {
+/// Common superclass for [_TakeRequest] and [_LookAheadRequest].
+abstract class _ListRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by the take call.
   final _completer = new Completer<List<T>>();
 
   /// List collecting events until enough have been seen.
   final _list = <T>[];
 
   /// Number of events to capture.
   ///
   /// The request [isComplete] when the length of [_list] reaches
   /// this value.
   final int _eventsToTake;
 
-  _TakeRequest(this._eventsToTake);
+  _ListRequest(this._eventsToTake);
 
   /// The future completed when the correct number of events have been captured.
   Future<List<T>> get future => _completer.future;
+}
+
+
+/// Request for a [StreamQueue.take] call.
+class _TakeRequest<T> extends _ListRequest<T> {
+  _TakeRequest(int eventsToTake) : super(eventsToTake);
 
   bool update(QueueList<Result<T>> events, bool isDone) {
     while (_list.length < _eventsToTake) {
       if (events.isEmpty) {
         if (isDone) break;
         return false;
       }
 
       var event = events.removeFirst();
       if (event.isError) {
-        _completer.completeError(event.asError.error, event.asError.stackTrace);
+        event.asError.complete(_completer);
         return true;
       }
       _list.add(event.asValue.value);
     }
     _completer.complete(_list);
     return true;
   }
 }
 
+
+/// Request for a [StreamQueue.lookAhead] call.
+class _LookAheadRequest<T> extends _ListRequest<T> {
+  _LookAheadRequest(int eventsToTake) : super(eventsToTake);
+
+  bool update(QueueList<Result<T>> events, bool isDone) {
+    while (_list.length < _eventsToTake) {
+      if (events.length == _list.length) {
+        if (isDone) break;
+        return false;
+      }
+      var event = events.elementAt(_list.length);
+      if (event.isError) {
+        event.asError.complete(_completer);
+        return true;
+      }
+      _list.add(event.asValue.value);
+    }
+    _completer.complete(_list);
+    return true;
+  }
+}
+
+
 /// Request for a [StreamQueue.cancel] call.
 ///
 /// The request needs no events, it just waits in the request queue
 /// until all previous events are fulfilled, then it cancels the stream queue
 /// source subscription.
 class _CancelRequest<T> implements _EventRequest<T> {
   /// Completer for the future returned by the `cancel` call.
   final _completer = new Completer();
   ///
   /// When the event is completed, it needs to cancel the active subscription
@@ skipping 104 matching lines @@
   }
 
   bool update(QueueList<Result<T>> events, bool isDone) {
     while (_eventsSent < events.length) {
       events[_eventsSent++].addTo(_controller);
     }
     if (isDone && !_controller.isClosed) _controller.close();
     return false;
   }
 }