Add a ScheduledStream class and some stream matchers.

pkg/scheduled_test/lib/src/stream_matcher.dart

Index: pkg/scheduled_test/lib/src/stream_matcher.dart
diff --git a/pkg/scheduled_test/lib/src/stream_matcher.dart b/pkg/scheduled_test/lib/src/stream_matcher.dart
new file mode 100644
index 0000000000000000000000000000000000000000..58cac83485df6ef06f20a285cfe8c817bd43cd64
--- /dev/null
+++ b/pkg/scheduled_test/lib/src/stream_matcher.dart
@@ -0,0 +1,297 @@
+// 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.stream_matcher;
+
+import 'dart:async';
+import 'dart:collection';
+
+import '../scheduled_stream.dart';
+import '../scheduled_test.dart';
+import 'utils.dart';
+
+/// An abstract superclass for matchers that validate and consume zero or more
+/// values emitted by a [ScheduledStream].
+///
+/// [StreamMatcher]s are most commonly used by passing them to
+/// [ScheduledStream.expect].
+abstract class StreamMatcher {
+  /// Wrap a [Matcher], [StreamMatcher] or [Object] in a [StreamMatcher].
+  ///
+  /// If this isn't a [StreamMatcher], a [nextValue] matcher is used.
+  factory StreamMatcher.wrap(matcher) =>
+      matcher is StreamMatcher ? matcher : nextValue(matcher);
+
+  /// Tries to match [this] against [stream].
+  ///
+  /// If the match succeeds, this returns `null`. If it fails, this returns a
+  /// [Description] describing the failure.
+  Future<Description> tryMatch(ScheduledStream stream);
+
+  String toString();
+}
+
+/// A matcher that consumes and matches a single value.
+///
+/// [matcher] can be a [Matcher] or an [Object], but not a [StreamMatcher].
+StreamMatcher nextValue(matcher) => new _NextValueMatcher(matcher);
+
+/// A matcher that consumes [n] values and matches a list containing those
+/// objects against [matcher].
+///
+/// [matcher] can be a [Matcher] or an [Object], but not a [StreamMatcher].
+StreamMatcher nextValues(int n, matcher) => new _NextValuesMatcher(n, matcher);
+
+/// A matcher that matches several sub-matchers in sequence.
+///
+/// Each element of [streamMatchers] can be a [StreamMatcher], a [Matcher], or
+/// an [Object].
+StreamMatcher inOrder(Iterable streamMatchers) {
+  streamMatchers = streamMatchers.toList();
+  if (streamMatchers.length == 1) {
+    return new StreamMatcher.wrap(streamMatchers.first);
+  } else {
+    return new _InOrderMatcher(streamMatchers);
+  }
+}
+
+/// A matcher that consumes values emitted by a stream until one matching
+/// [matcher] is emitted.
+///
+/// This will fail if the stream never emits [matcher]

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

"[matcher]" -> "a value that matches [matcher]."

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

Done.

+///
+/// [matcher] can be a [Matcher] or an [Object], but not a [StreamMatcher].
+StreamMatcher consumeThrough(matcher) => new _ConsumeThroughMatcher(matcher);

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

How about "consumeUntil"?

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

That's what this was originally, but I wanted it to be clear that it consumed
the value that matched [matcher] as well.

+
+/// A matcher that matches either [streamMatcher1], [streamMatcher2], or both.
+///
+/// If both matchers match the stream, the one that consumed more values will be
+/// used.
+///
+/// Both [streamMatcher1] and [streamMatcher2] can be a [StreamMatcher], a
+/// [Matcher], or an [Object].
+StreamMatcher either(streamMatcher1, streamMatcher2) =>

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

My intuition is that "either" would not match both. How about "anyOf" or "any"
and allow more than two?

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

I think it makes sense for "either" to match both branches. I'd expect something
matching one and only one to be called "xor" or "eitherButNotBoth". I could
change this to "eitherOrBoth" if you prefer.

I think extending this to match any number of matchers would make the code
considerably more complicated, and worse would make the string descriptions more
difficult to read. I also don't think it would provide much benefit.

+  new _EitherMatcher(streamMatcher1, streamMatcher2);
+
+/// A matcher that consumes [streamMatcher] if it matches, or nothing otherwise.
+///
+/// This matcher will always match a stream. It exists to consume values that
+/// may or may not be emitted by a stream.
+///
+/// [streamMatcher] can be a [StreamMatcher], a [Matcher], or an [Object].
+StreamMatcher allow(streamMatcher) => new _AllowMatcher(streamMatcher);

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

"skipWhile"?

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

That implies that it will skip more than one value, which is incorrect.

+
+/// A matcher that matches a stream that emits no more values.
+StreamMatcher get isDone => new _IsDoneMatcher();
+
+/// See [nextValue].
+class _NextValueMatcher extends StreamMatcher {
+  final Matcher _matcher;
+
+  _NextValueMatcher(matcher)
+      : _matcher = wrapMatcher(matcher);
+
+  Future<Description> tryMatch(ScheduledStream stream) {
+    return stream.hasNext.then((hasNext) {
+      if (!hasNext) {
+        return new StringDescription("unexpected end of stream");

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

Describe the matcher here too.

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

This description will be provided elsewhere in the expectation message.

+      }
+      return stream.next().then((value) {
+        var matchState = {};
+        if (_matcher.matches(value, matchState)) return null;
+        return _matcher.describeMismatch(value, new StringDescription(),
+            matchState, false);
+      });
+    });
+  }
+
+  String toString() => _matcher.describe(new StringDescription()).toString();
+}
+
+/// See [nextValues].
+class _NextValuesMatcher extends StreamMatcher {
+  final int _n;

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

Doc.

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

Usually we don't document private instance variables that come directly from
functions that have their own documentation. In this case, [_n] is the same as
[n] in [nextValues].

+  final Matcher _matcher;
+
+  _NextValuesMatcher(this._n, matcher)
+      : _matcher = wrapMatcher(matcher);
+
+  Future<Description> tryMatch(ScheduledStream stream) {
+    var collectedValues = [];
+    collectValues(count) {
+      if (count == 0) return null;
+
+      return stream.hasNext.then((hasNext) {
+        if (!hasNext) return new StringDescription('unexpected end of stream');

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

Describe the expected values here.

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

See above.

+
+        return stream.next().then((value) {
+          collectedValues.add(value);
+          return collectValues(count - 1);
+        });
+      });
+    }
+
+    return collectValues(_n).then((description) {
+      if (description != null) return description;
+      var matchState = {};
+      if (_matcher.matches(collectedValues, matchState)) return null;
+      return _matcher.describeMismatch(collectedValues, new StringDescription(),
+          matchState, false);
+    });
+  }
+
+  String toString() {
+    return new StringDescription('$_n values that ')
+        .addDescriptionOf(_matcher)
+        .toString();
+  }
+}
+
+/// See [inOrder].
+class _InOrderMatcher extends StreamMatcher {
+  final List<StreamMatcher> _matchers;
+
+  _InOrderMatcher(Iterable streamMatchers)
+      : _matchers = streamMatchers.map((matcher) =>
+          new StreamMatcher.wrap(matcher)).toList();
+
+  Future<Description> tryMatch(ScheduledStream stream) {
+    var matchers = new Queue.from(_matchers);
+
+    matchNext() {
+      if (matchers.isEmpty) return new Future.value();
+      var matcher = matchers.removeFirst();
+      return matcher.tryMatch(stream).then((description) {
+        if (description == null) return matchNext();
+        var newDescription = new StringDescription(
+              'matcher #${_matchers.length - matchers.length} failed');
+        if (description.length != 0) newDescription.add(':\n$description');
+        return newDescription;
+      });
+    }
+
+    return matchNext();
+  }
+
+  String toString() => _matchers
+      .map((matcher) => prefixLines(matcher.toString(), firstPrefix: '* '))
+      .join('\n');
+}
+
+/// See [consumeThrough].
+class _ConsumeThroughMatcher extends StreamMatcher {
+  final Matcher _matcher;
+
+  _ConsumeThroughMatcher(matcher)
+      : _matcher = wrapMatcher(matcher);
+
+  Future<Description> tryMatch(ScheduledStream stream) {
+    consumeNext() {
+      return stream.hasNext.then((hasNext) {
+        if (!hasNext) return new StringDescription("unexpected end of stream");

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

Ditto.

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

See above.

+
+        return stream.next().then((value) {
+          if (_matcher.matches(value, {})) return null;
+          return consumeNext();
+        });
+      });
+    }
+
+    return consumeNext();
+  }
+
+  String toString() {
+    return new StringDescription('values followed by ')
+      .addDescriptionOf(_matcher).toString();
+  }
+}
+
+/// See [either].
+class _EitherMatcher extends StreamMatcher {
+  final StreamMatcher _matcher1;
+  final StreamMatcher _matcher2;
+
+  _EitherMatcher(streamMatcher1, streamMatcher2)
+      : _matcher1 = new StreamMatcher.wrap(streamMatcher1),
+        _matcher2 = new StreamMatcher.wrap(streamMatcher2);
+
+  Future<Description> tryMatch(ScheduledStream stream) {
+    var stream1 = stream.fork();
+    var stream2 = stream.fork();
+
+    return Future.wait([
+      _matcher1.tryMatch(stream1),
+      _matcher2.tryMatch(stream2)
+    ]).whenComplete(() {
+      stream1.close();
+      stream2.close();

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

Should these be closed only when *both* futures have completed, or does it make
more sense to add a whenComplete() after each tryMatch()?

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

Done.

+    }).then((descriptions) {
+      var description1 = descriptions.first;
+      var description2 = descriptions.last;
+
+      // If both matchers matched, use the one that consumed more of the stream.
+      if (description1 == null && description2 == null) {
+        if (stream1.values.length >= stream2.values.length) {
+          return _matcher1.match(stream);
+        } else {
+          return _matcher2.match(stream);
+        }
+      } else if (description1 == null) {
+        return _matcher1.match(stream);
+      } else if (description2 == null) {
+        return _matcher2.match(stream);
+      } else {
+        return new StringDescription('both\n')
+            .add(prefixLines(description1.toString(), prefix: '  '))
+            .add('\nand\n')
+            .add(prefixLines(description2.toString(), prefix: '  '))
+            .toString();
+      }
+    });
+  }
+
+  String toString() {
+    return new StringDescription('either\n')
+        .add(prefixLines(_matcher1.toString(), prefix: '  '))
+        .add('\nor\n')
+        .add(prefixLines(_matcher2.toString(), prefix: '  '))
+        .toString();
+  }
+}
+
+/// See [allow].
+class _AllowMatcher extends StreamMatcher {
+  final StreamMatcher _matcher;
+
+  _AllowMatcher(streamMatcher)
+      : _matcher = new StreamMatcher.wrap(streamMatcher);
+
+  Future<Description> tryMatch(ScheduledStream stream) {
+    var fork = stream.fork();
+    return _matcher.tryMatch(fork).whenComplete(fork.close).then((description) {
+      if (description != null) return null;
+      return _matcher.match(stream);
+    });
+  }
+
+  String toString() {
+    return new StringDescription('allow\n')
+        .add(prefixLines(_matcher.toString()))
+        .toString();
+  }
+}
+
+/// See [isDone].
+class _IsDoneMatcher extends StreamMatcher {
+  _IsDoneMatcher();
+
+  Future<Description> tryMatch(ScheduledStream stream) {
+    return stream.hasNext.then((hasNext) {
+      if (!hasNext) return null;
+      return new StringDescription("stream wasn't finished");
+    });
+  }
+
+  String toString() => 'is done';
+}