Make StreamSubscription be the active part of a stream.

pkg/scheduled_test/lib/src/utils.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 utils;
 
 import 'dart:async';
 
 import 'package:stack_trace/stack_trace.dart';
 
@@ skipping 76 matching lines @@
         controller.add,
         onError: (error) => controller.addError(error),
         onDone: controller.close);
   }).catchError((e) {
     controller.addError(e);
     controller.close();
   });
   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();
+  return streamIterator.moveNext().then((hasNext) {
+    if (hasNext) {
+      return streamIterator.current;
+    } else {
+      return new Future.error(new StateError("No elements"), stackTrace);
+    }
+  });
+}
+
+/// 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) {

[floitsch, 2013/05/23 18:38:23]

We could consider having this on the iterator. It seems like a common operation.

[Lasse Reichstein Nielsen, 2013/05/24 11:13:30]

What would the operation be?
 Future StreamIterator.moveNext(action(bool hasNext)); ?

[floitsch, 2013/05/24 13:53:41]

I meant the "collectAll".

+      if (hasNext) {
+        if (!buffer.isEmpty) buffer.write('\n');
+        buffer.write(streamIterator.current);
+        collectAll();
+      } else {
+        completer.complete(buffer.toString());
+      }
+    }, onError: completer.completeError);
   }
-
-  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);
+  collectAll();
   return completer.future;
 }
 
 /// A function that can be called to cancel a [Stream] and send a done message.
 typedef void StreamCanceller();
 
 // TODO(nweiz): use a StreamSubscription when issue 9026 is fixed.
 /// Returns a wrapped version of [stream] along with a function that will cancel
 /// the wrapped stream. Unlike [StreamSubscription], this canceller will send a
 /// "done" message to the wrapped stream.
@@ skipping 63 matching lines @@
 }
 
 /// Returns a string representation of [trace] that has the core and test frames
 /// folded together.
 String terseTraceString(StackTrace trace) {
   return new Trace.from(trace).terse.foldFrames((frame) {
     return frame.package == 'scheduled_test' || frame.package == 'unittest' ||
         frame.isCore;
   }).toString().trim();
 }