Make StreamSubscription be the active part of a stream.

pkg/scheduled_test/lib/src/utils.dart

Index: pkg/scheduled_test/lib/src/utils.dart
diff --git a/pkg/scheduled_test/lib/src/utils.dart b/pkg/scheduled_test/lib/src/utils.dart
index d686ef99022b511ccc980c5c9ff718e63471b8d7..62461a6f02422bd7cf40f371548fbacb0aa42a0c 100644
--- a/pkg/scheduled_test/lib/src/utils.dart
+++ b/pkg/scheduled_test/lib/src/utils.dart
@@ -94,27 +94,38 @@ Stream futureStream(Future<Stream> future) {
   return controller.stream;
 }
 
-// TODO(nweiz): remove this when issue 7964 is fixed.
-/// Returns a [Future] that will complete to the first element of [stream].
-/// Unlike [Stream.first], this is safe to use with single-subscription streams.
-Future streamFirst(Stream stream) {
-  var stackTrace;
-  try {
-    throw '';
-  } catch (_, thrownStackTrace) {
-    stackTrace = thrownStackTrace;
-  }
+/// Returns the first element of a [StreamIterator].
+///
+/// If the [StreamIterator] has no elements, the result is a state error.
+Future<String> streamIteratorFirst(StreamIterator<String> streamIterator) {
+  StackTrace stackTrace = new Trace.current();

[nweiz, 2013/05/24 20:12:31]

"var stackTrace"

+  return streamIterator.moveNext().then((hasNext) {
+    if (hasNext) {
+      return streamIterator.current;
+    } else {
+      return new Future.error(new StateError("No elements"), stackTrace);
+    }
+  });
+}
 
-  var completer = new Completer();
-  var subscription;
-  subscription = stream.listen((value) {
-    subscription.cancel();
-    completer.complete(value);
-  }, onError: (e) {
-    completer.completeError(e);
-  }, onDone: () {
-    completer.completeError(new StateError("No elements"), stackTrace);
-  }, cancelOnError: true);
+/// Collects all remaining lines from a [StreamIterator] of lines.
+///
+/// Returns the concatenation of the collected lines joined by newlines.
+Future<String> concatRest(StreamIterator<String> streamIterator) {
+  var completer = new Completer<String>();
+  var buffer = new StringBuffer();
+  void collectAll() {
+    streamIterator.moveNext().then((hasNext) {
+      if (hasNext) {
+        if (!buffer.isEmpty) buffer.write('\n');
+        buffer.write(streamIterator.current);
+        collectAll();
+      } else {
+        completer.complete(buffer.toString());
+      }
+    }, onError: completer.completeError);
+  }
+  collectAll();
   return completer.future;
 }