Add a ScheduledStream class and some stream matchers.

pkg/scheduled_test/lib/scheduled_stream.dart

+// Copyright (c) 2013, 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 scheduled_test.scheduled_stream;
+
+import 'dart:async';
+import 'dart:collection';
+
+import 'package:stack_trace/stack_trace.dart';
+
+import 'scheduled_test.dart';
+import 'src/stream_matcher.dart';
+import 'src/utils.dart';
+
+export 'src/stream_matcher.dart';
+
+/// A wrapper for streams that supports a pull-based model of retrieving values
+/// as well as a set of [StreamMatcher]s for testing the values emitted by the
+/// stream.
+///
+/// The only method on [ScheduledStream] that's actually scheduled is [expect],
+/// which is the method that users testing streaming code are most likely to
+/// want to use.
+class ScheduledStream<T> {
+  /// The underlying stream.
+  final Stream<T> _stream;
+
+  /// The subscription to [_stream].
+  StreamSubscription<T> _subscription;
+
+  /// The completer for emitting a value requested by [next].
+  ///
+  /// If this is non-null, [_pendingValues] will always be empty, since any
+  /// value coming in will be passed to this completer.
+  Completer<T> _nextCompleter;
+
+  /// The completer for emitting a value requested by [hasNext].
+  Completer<bool> _hasNextCompleter;
+
+  /// The set of all streams forked from this one.
+  final _forks = new Set<ScheduledStream<T>>();
+
+  /// The queue of values emitted by [_stream] but not yet emitted through
+  /// [next].
+  final _pendingValues = new Queue<Either<T, Pair<dynamic, StackTrace>>>();

[Bob Nystrom, 2013/12/21 00:11:34]

Instead of Either<Pair>, how about a concrete type that represents a value or a
failure? That's the only thing you're using Either to represent isn't it?

[nweiz, 2014/01/07 03:16:39]

Good idea. Done.

+
+  /// All values emitted by this stream so far.

[Bob Nystrom, 2013/12/21 00:11:34]

This is confusing. "Emitted" implies a push model, but this is really about the
pull model. How about, "All values consumed by this scheduled stream so far."

Likewise the second sentence is confusing.

[nweiz, 2014/01/07 03:16:39]

"Consumed" is also weird because it sounds like it's referring to the values
consumed from the underlying stream.

I don't love "emitted", but I also don't mind it, since I think it makes a fair
amount of sense to say that a user requests a value that later gets emitted.

+  ///
+  /// This does not include values emitted by the underlying stream but not yet
+  /// emitted through [next].
+  List<T> get values => new UnmodifiableListView(_values);
+  final _values = new List<T>();
+
+  /// All values emitted by the underlying stream.
+  ///
+  /// This is intended primarily for providing debugging information.
+  List<T> get allValues {

[Bob Nystrom, 2013/12/21 00:11:34]

The distinction in names here is a bit ambiguous. How about "streamedValues" or
"emittedValues"?

[nweiz, 2014/01/07 03:16:39]

Done.

+    var list = new List<T>.from(_values);
+    list.addAll(_pendingValues.where((value) => value.isFirst)
+        .map((value) => value.first));
+    return new UnmodifiableListView(list);
+  }
+
+  /// Whether the wrapped stream has been closed.
+  bool _isDone = false;
+
+  /// Whether [next] has been called but has not yet returned.
+  bool _nextPending = false;

[Bob Nystrom, 2013/12/21 00:11:34]

How about _isNextPending? "_nextPending" sound to me like "the thing that is
next pending".

Do you need to store this at all? Is it equivalent to _nextCompleter != null? If
not, why not?

[nweiz, 2014/01/07 03:16:39]

Done.
Added a comment about this.

+
+  /// Creates a new scheduled stream wrapping [stream].
+  ScheduledStream(Stream<T> stream)
+      : _stream = stream.asBroadcastStream() {
+    _subscription = _stream.listen((value) {
+      if (_hasNextCompleter != null) {
+        _hasNextCompleter.complete(true);
+        _hasNextCompleter = null;
+      }
+
+      if (_nextCompleter != null) {
+        _nextCompleter.complete(value);
+        _values.add(value);
+        _nextCompleter = null;
+        _nextPending = false;
+      } else {
+        _pendingValues.add(new Either.withFirst(value));
+      }
+    }, onError: (error, stackTrace) {
+      if (_hasNextCompleter != null) {
+        _hasNextCompleter.completeError(error, stackTrace);
+        _hasNextCompleter = null;
+      }
+
+      if (_nextCompleter != null) {
+        _nextCompleter.completeError(error, stackTrace);
+        _nextCompleter = null;
+      } else {
+        _pendingValues.add(new Either.withSecond(new Pair(error, stackTrace)));
+      }
+    }, onDone: _onDone);
+  }
+
+  /// Enqueue an expectation that [streamMatcher] will match the value(s)
+  /// emitted by the stream at this point in the schedule.
+  ///
+  /// If [streamMatcher] is a [StreamMatcher], it will match the stream as a
+  /// whole. If it's a [Matcher] or another object, it will match the next value
+  /// emitted by the stream (as though it were a [nextValue] matcher).
+  ///
+  /// This call is scheduled; the expectation won't be added until the schedule
+  /// reaches this point, and the schedule won't continue until the matcher has
+  /// matched the stream.
+  void expect(streamMatcher) {
+    streamMatcher = new StreamMatcher.wrap(streamMatcher);
+    var description = 'stream emits $streamMatcher';
+    schedule(() {
+      return streamMatcher.tryMatch(stream).then((description) {
+        if (description == null) return;
+
+        var expected = prefixLines(streamMatcher.toString(),
+            firstPrefix: 'Expected: ',
+            prefix:      '        | ');
+
+        var actual = prefixLines(stream.allValues.map((value) {
+          return prefixLines(value.toString(), firstPrefix: '* ');
+        }).join('\n'),
+            firstPrefix: ' Emitted: ',
+            prefix:      '          ');
+
+        var which = '';
+        if (description.length > 0) {
+          which = '\n' + prefixLines(description.toString(),
+              firstPrefix: '   Which: ',
+              prefix:      '        | ');
+        }
+
+        fail("$expected\n$actual$which");
+      });
+    }, description);
+  }
+
+  /// Returns a Future that completes to the next value emitted by this stream.
+  ///
+  /// It's a [StateError] to call [next] when another call's Future has not yet
+  /// completed, or when the stream has no more values. The latter can be
+  /// checked using [hasNext].
+  Future<T> next() {
+    if (_nextPending) {
+      return new Future.error(
+          new StateError("There's already a pending call to "
+              "ScheduledStream.next."),
+          new Chain.current());
+    }
+
+    if (_pendingValues.isNotEmpty) {
+      _nextPending = true;
+
+      return _pendingValues.removeFirst().match((value) {
+        _values.add(value);
+        return new Future.value(value);
+      }, (pair) => Chain.track(new Future.error(pair.first, pair.last)))
+          .whenComplete(() {

[Bob Nystrom, 2013/12/21 00:11:34]

It took me a long time to parse this and realize that .whenComplete is applied
to the result of .match(), not Chain.track(). How about doing:

var future = _pendingValues.removeFirst().match(...)
return future.whenComplete(...);

Also, the future will already be complete in either case, won't it? Is there a
reason you're using whenComplete() here instead of just leaving _nextPending
false?

[nweiz, 2014/01/07 03:16:39]

I changed this as part of replacing Either with FallibleValue.
There's a period of time between when [Future.value] and [Future.error] are
initialized and when they fire. Other events can happen during this time.

+        _nextPending = false;
+      });
+    } else if (_isDone) {
+      return new Future.error(
+          new StateError("ScheduledStream has no more elements."),
+          new Chain.current());
+    }
+
+    _nextPending = true;
+    _nextCompleter = new Completer();
+    return _nextCompleter.future;
+  }
+
+  /// Returns a Future that completes to a boolean indicating whether the stream
+  /// has additional values or not.
+  Future<bool> get hasNext {
+    if (_hasNextCompleter != null) return _hasNextCompleter.future;
+
+    if (_pendingValues.isNotEmpty) {
+      return _pendingValues.first.match(
+          (value) => new Future.value(true),
+          (pair) => Chain.track(new Future.error(pair.first, pair.last)));
+    } else if (_isDone) {
+      return new Future.value(false);
+    }
+
+    _hasNextCompleter = new Completer();
+    return _hasNextCompleter.future;
+  }
+
+  /// Returns a fork of this stream.
+  ///
+  /// The fork begins at the same point [this] is at. Values can be read from it
+  /// without consuming values in [this]. If [this] is closed, the fork will be
+  /// closed at whatever point it's currently at.
+  ScheduledStream<T> fork() {
+    var controller = new StreamController<T>();
+    for (var value in _pendingValues) {
+      value.match(controller.add,
+          (pair) => controller.addError(pair.first, pair.last));
+    }
+
+    if (_isDone) {
+      controller.close();
+    } else {
+      _stream.pipe(controller);
+    }
+
+    var fork = new ScheduledStream<T>(controller.stream);
+    _forks.add(fork);
+    return fork;
+  }
+
+  /// Closes this stream.
+  ///
+  /// This cancels the subscription to the underlying stream and acts as though
+  /// [this] was closed immediately after the current position, regardless of
+  /// whether the underlying stream has emitted additional events.
+  void close() {
+    _subscription.cancel();
+    _pendingValues.clear();
+
+    for (var fork in _forks) {
+      fork.close();
+    }

[Bob Nystrom, 2013/12/21 00:11:34]

Just to be nice to the GC, may as well clear _forks too.

[nweiz, 2014/01/07 03:16:39]

Done.

+
+    if (!_isDone) _onDone();
+  }
+
+  /// Handles a "done" event from the underlying stream, as well as [this] being
+  /// closed.
+  void _onDone() {
+    if (_hasNextCompleter != null) {
+      _hasNextCompleter.complete(false);
+      _hasNextCompleter = null;
+    }
+
+    if (_nextCompleter != null) {
+      _nextCompleter.completeError(
+          new StateError("ScheduledStream has no more elements."),
+          new Chain.current());
+      _nextCompleter = null;
+    }

[Bob Nystrom, 2013/12/21 00:11:34]

Probably want to clear _nextPending here too.

[nweiz, 2014/01/07 03:16:39]

Done.

+
+    _isDone = true;
+  }
+}