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 104 matching lines @@
   /// one events.
   Future<bool> get hasNext {
     if (!_isClosed) {
       var hasNextRequest = new _HasNextRequest();
       _addRequest(hasNextRequest);
       return hasNextRequest.future;
     }
     throw _failClosed();
   }
 
+
+  /// Look at the next [count] data events without consuming them.

[nweiz, 2017/01/24 22:37:59]

"Look at" -> "Returns"

[Lasse Reichstein Nielsen, 2017/01/25 07:58:03]

Done.

+  ///
+  /// 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) {
+    if (count < 0) throw new RangeError.range(count, 0, null, "count");

[nweiz, 2017/01/24 22:37:59]

`RangeError.checkNotNegative()`?

[Lasse Reichstein Nielsen, 2017/01/25 07:58:03]

Done.

+    if (!_isClosed) {
+      var request = new _LookAheadRequest<T>(count);
+      _addRequest(request);
+      return request.future;
+    }
+    throw _failClosed();
+  }

[nweiz, 2017/01/24 22:37:59]

Why not just implement these in terms of transactions and take()/next? It would
be a lot simpler.

[Lasse Reichstein Nielsen, 2017/01/25 07:58:03]

Because it would also be a lot more expensive.
I prefer not to implement less complex operations in terms of more complex ones.

[nweiz, 2017/01/25 22:41:18]

Do you have benchmarks where the extra processing produces a real-world
difference? Do you have a strong expectation that these APIs will be used in
performance-critical core loops? If not, I think we should strongly favor
cleaner, simpler implementations in terms of APIs that are already
thoroughly-tested. It may be a little slower, but it will make maintenance and
external contribution much easier.

[Lasse Reichstein Nielsen, 2017/01/26 12:13:43]

I've added a benchmark below. It shows that `return sq.peek` is 4-6 times as
fast as `var t = sq.startTransaction();return
t.newQueue().next().whenComplete(t.reject)` (which is the fastest version I
found that handles errors correctly).

My expectation of how it will be used probably isn't really useful, since
history shows that prediction is hard - this is a general purpose library, so I
cannot predict how it will be used. Libraries, platform or otherwise, should
always assume that they are being used in performance critical places, because
the users of the libraries are not in a (realistic) position to change them.

And I do believe, very strongly, that defining simple operations in terms of
more complex ones is likely to give you unexpected dependencies and make
maintenance harder in the long run (but then, that's a prediction too).

Benchmark:

import "dart:async";
import "package:async/async.dart";

bench(n, f, t) async {
  var c = 0;
  var e = 0;
  var s = new Stopwatch()..start();
  do {
    var st = new Stream.fromIterable([1,2,3,4,5,6]);
    var sq = new StreamQueue(st);
    while (await sq.hasNext) {
      var p = await f(sq);
      var a = await sq.next;
      if (!identical(p, a)) throw "wat";
    }
    e = s.elapsedMilliseconds;
    c += 6;
  } while (e < t);
  s.stop();
  return c/e;
}

main() async {
  await bench("peek", peek, 0);
  await bench("startTransaction", transact, 0);
  print("peek: ${await bench("peek", peek, 2000)} /ms");
  print("startTransaction: ${await bench("peek", transact, 2000)} /ms");
  // Do the important computation more than once, to emphasize it relative
  // to the overhead around it.
  print("peek*2: ${await bench("peek", twice(peek), 2000)} /ms");
  print("startTransaction*2: ${await bench("peek", twice(transact), 2000)}
/ms");
  print("peek*3: ${await bench("peek", thrice(peek), 2000)} /ms");
  print("startTransaction*3: ${await bench("peek", thrice(transact), 2000)}
/ms");
}

peek(sq) => sq.peek;
transact(sq) {
  var t = sq.startTransaction();
  return t.newQueue().next.whenComplete(t.reject);
}

twice(f) => (x) => f(x).then((_)=>f(x));
thrice(f) => (x) {
  var again = (_) => f(x);
  return f(x).then(again).then(again);
};

+
   /// 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();
   }
 
+  /// Looks at the next (yet unrequested) event from the stream.

[nweiz, 2017/01/24 22:37:59]

"Looks at" -> "Returns"

[Lasse Reichstein Nielsen, 2017/01/25 07:58:03]

Done.

+  ///
+  /// Like [next] except that the event is not consumed.
+  /// If the next event is an error event, it 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 179 matching lines @@
   /// 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.
+  /// None of [lookAhead], [next], [peek], [rest], [skip], [take] or [cancel]
+  /// may be called again.

[nweiz, 2017/01/24 22:37:59]

Is it really necessary to explicitly list all the requests? It seems difficult
to maintain, and we're already saying that no more requests can be added.

[Lasse Reichstein Nielsen, 2017/01/25 07:58:03]

It seemed like a great idea at the time ... but it is getting ridiculous.
I'll reword it to only use examples, not an exhausting list.

   Future cancel({bool immediate: false}) {
     if (_isClosed) throw _failClosed();
     _isClosed = true;
 
     if (!immediate) {
       var request = new _CancelRequest(this);
       _addRequest(request);
       return request.future;
     }
 
@@ skipping 337 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;
   }
 }