Add a ScheduledStream class and some stream matchers.

pkg/scheduled_test/lib/scheduled_stream.dart

Index: pkg/scheduled_test/lib/scheduled_stream.dart
diff --git a/pkg/scheduled_test/lib/scheduled_stream.dart b/pkg/scheduled_test/lib/scheduled_stream.dart
new file mode 100644
index 0000000000000000000000000000000000000000..dba84a1c278f9a7fe335b7cd0357b4c2fc978316
--- /dev/null
+++ b/pkg/scheduled_test/lib/scheduled_stream.dart
@@ -0,0 +1,262 @@
+// 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<FallibleValue<T>>();
+
+  /// All values emitted by this stream so far.
+  ///
+  /// This does not include values emitted by the underlying stream but not yet
+  /// emitted through [next].
+  List<T> get emittedValues => new UnmodifiableListView(_emittedValues);
+  final _emittedValues = new List<T>();
+
+  /// All values emitted by the underlying stream.
+  ///
+  /// This is intended primarily for providing debugging information.
+  List<T> get allValues {
+    var list = new List<T>.from(_emittedValues);
+    list.addAll(_pendingValues.where((value) => value.hasValue)
+        .map((value) => value.value));
+    return new UnmodifiableListView(list);
+  }
+
+  /// Whether the wrapped stream has been closed.
+  bool _isDone = false;
+
+  /// Whether [next] has been called but has not yet returned.
+  ///
+  /// This is distinct from `_nextCompleter != null` when [next] is called while
+  /// there are pending values available, until the future it returns completes.
+  bool _isNextPending = false;
+
+  /// 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);
+        _emittedValues.add(value);
+        _nextCompleter = null;
+        _isNextPending = false;
+      } else {
+        _pendingValues.add(new FallibleValue.withValue(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 FallibleValue.withError(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(this).then((description) {
+        if (description == null) return;
+
+        var expected = prefixLines(streamMatcher.toString(),
+            firstPrefix: 'Expected: ',
+            prefix:      '        | ');
+
+        var actual = prefixLines(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 (_isNextPending) {
+      return new Future.error(
+          new StateError("There's already a pending call to "
+              "ScheduledStream.next."),
+          new Chain.current());
+    }
+
+    if (_pendingValues.isNotEmpty) {
+      _isNextPending = true;
+
+      return syncFuture(() {
+        var valueOrError = _pendingValues.removeFirst();
+        if (valueOrError.hasValue) {
+          _emittedValues.add(valueOrError.value);
+          return valueOrError.value;
+        } else {
+          return new Chain.track(new Future.error(
+              valueOrError.error, valueOrError.stackTrace));
+        }
+      }).whenComplete(() {
+        _isNextPending = false;
+      });
+    } else if (_isDone) {
+      return new Future.error(
+          new StateError("ScheduledStream has no more elements."),
+          new Chain.current());
+    }
+
+    _isNextPending = 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) {
+      var valueOrError = _pendingValues.first;
+      if (valueOrError.hasValue) return new Future.value(true);
+      return new Future.error(valueOrError.error, valueOrError.stackTrace);

[Bob Nystrom, 2014/01/08 00:47:40]

Given how nicely it maps 1-1, you could just add a .toFuture() to FallibleValue.

[nweiz, 2014/01/08 22:15:09]

Done.

+    } 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 valueOrError in _pendingValues) {
+      if (valueOrError.hasValue) {
+        controller.add(valueOrError.value);
+      } else {
+        controller.addError(valueOrError.error, valueOrError.stackTrace);
+      }
+    }
+
+    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();
+    }
+    _forks.clear();
+
+    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;
+      _isNextPending = false;
+    }
+
+    _isDone = true;
+  }
+}