Make StreamSubscription be the active part of a stream.

pkg/scheduled_test/lib/scheduled_process.dart

Index: pkg/scheduled_test/lib/scheduled_process.dart
diff --git a/pkg/scheduled_test/lib/scheduled_process.dart b/pkg/scheduled_test/lib/scheduled_process.dart
index 23370f10942602b5ccc0e42e56dbce3c3d87c70e..049d8b8739140c37f1a6347dffa145760f8f6295 100644
--- a/pkg/scheduled_test/lib/scheduled_process.dart
+++ b/pkg/scheduled_test/lib/scheduled_process.dart
@@ -39,7 +39,7 @@ class ScheduledProcess {
   Stream<String> _stdoutLog;
 
   /// A line-by-line view of the standard output stream of the process.
-  Stream<String> _stdout;
+  StreamIterator<String> _stdout;

[nweiz, 2013/05/22 18:42:41]

I'm not sure I understand why using a StreamIterator here is better than using a
Stream. It makes the implementation of [remainingStdout] and [remainingStderr]
more complicated, and I don't see a corresponding benefit.

[floitsch, 2013/05/22 20:23:35]

The moment you cancel a stream it is dead. We have had a bug for some time where
you could still subscribe to it.
This CL fixes the bug. Unfortunately there is code that relied on the broken
behavior. The StreamIterator class provides an interface that usually allows to
simulate the old behavior.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Exactly. 
The current code subscribed to the same stream more than once (using streamFirst
one or more time). That does not work any more.

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

Okay, I suppose that makes sense. It seems a little weird to me that having an
Iterator-style API and supporting multiple subscriptions are conflated concepts,
but that's a different discussion and I don't feel too strongly about it.

If streamFirst is completely broken now, please remove it from utils.dart.

[Lasse Reichstein Nielsen, 2013/05/27 08:04:53]

Done.

 
   /// A canceller that controls both [_stdout] and [_stdoutLog].
   StreamCanceller _stdoutCanceller;
@@ -49,7 +49,7 @@ class ScheduledProcess {
   Stream<String> _stderrLog;
 
   /// A line-by-line view of the standard error stream of the process.
-  Stream<String> _stderr;
+  StreamIterator<String> _stderr;
 
   /// A canceller that controls both [_stderr] and [_stderrLog].
   StreamCanceller _stderrCanceller;
@@ -102,8 +102,8 @@ class ScheduledProcess {
     _stderrCanceller = stderrWithCanceller.last;
     _stderrLog = stderrWithCanceller.first;
 
-    _stdout = stdoutStream();
-    _stderr = stderrStream();
+    _stdout = new StreamIterator<String>(stdoutStream());
+    _stderr = new StreamIterator<String>(stderrStream());
   }
 
   /// Updates [_description] to reflect [executable] and [arguments], which are
@@ -239,14 +239,49 @@ class ScheduledProcess {
     }, "cleaning up process '$description'");
   }
 
+  // Similar to [streamFirst], but works on streamIterators.
+  // And it works at all, which streamFirst doesn't.

[floitsch, 2013/05/22 16:26:29]

Reword. This could be read like an insult.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Ack, working comment left in :)

+  Future<String> _streamIteratorFirst(StreamIterator<String> streamIterator) {

[nweiz, 2013/05/22 18:42:41]

This belongs in utils.dart.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Ok, will move it.

+    StackTrace stackTrace;
+    try {
+      throw 0;
+    } catch (_, thrownStackTrace) {
+      stackTrace = thrownStackTrace;
+    }

[Bob Nystrom, 2013/05/22 16:23:46]

Instead of doing this by hand, use new Trace.current() in the stack_trace
package.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Will do.

+    return streamIterator.moveNext().then((hasNext) {
+      if (hasNext) {
+        return streamIterator.current;
+      } else {
+        return new Future.error(new StateError("No elements"), stackTrace);
+      }
+    });
+  }
+
   /// Reads the next line of stdout from the process.
-  Future<String> nextLine() => schedule(() => streamFirst(_stdout),
+  Future<String> nextLine() => schedule(() => _streamIteratorFirst(_stdout),
       "reading the next stdout line from process '$description'");
 
   /// Reads the next line of stderr from the process.
-  Future<String> nextErrLine() => schedule(() => streamFirst(_stderr),
+  Future<String> nextErrLine() => schedule(() => _streamIteratorFirst(_stderr),
       "reading the next stderr line from process '$description'");
 
+  /// Collects all remaining lines from a [StreamIterator] of lines.
+  ///
+  /// Returns the concatenation of the collected lines joined by newlines.
+  static Future<String> _concatRest(StreamIterator<String> streamIterator) {

[nweiz, 2013/05/22 18:42:41]

This also belongs in utils.dart.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Done.

+    Future<String> collectAll(List<String> accumulator) {
+      return streamIterator.moveNext().then((bool hasNext) {

[Bob Nystrom, 2013/05/22 16:23:46]

Don't bother type annotating parameters in fn expressions.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

ok.
Grudgingly.
How else would I know the type of hasNext from looking at the code? :(

[Bob Nystrom, 2013/05/24 15:54:38]

1. You know what you're passing that callback to, so you probably know the fn
type it expects.
2. It's name starts with "has" -> probably a bool.
3. The type doesn't matter that much, its name tells you what it is.

[Lasse Reichstein Nielsen, 2013/05/27 08:04:53]

I don't believe in "knowing" being good enough as documentation.
Writing code is easy. Reading it is hard.
You can't convince me that types are irrelevant. Redundancy is *good* when it
comes to documentation. It dispels any doubt you may have over either of the
parts in isolation.

+        if (hasNext) {
+          accumulator.add(streamIterator.current);
+          return collectAll(accumulator);

[Bob Nystrom, 2013/05/22 16:23:46]

If the stream has a lot of output, this can build a very long chain of futures.
They're async, so they won't blow the stack, but I've seen long future chains
exhaust the heap or consume a lot of CPU traversing the chain.

Maybe we should look to add something reduce-like to dart:async specifically to
address this? You can handle it yourself using a completer, but it's tedious and
error-prone to do so.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

I've rewritten this to not chain the futures, and just complete a completer at
the end.

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

Working around this here before we know it's actually a problem seems like
premature optimization. So far (as far as I know) the only case where we've seen
this actually become an issue is in pathologically-exponential cases in the Pub
version solver. I don't want to add a workaround here unless we've got good
reason to believe the issue will come up in practice.

If it does come up in practice, that means that Futures are memory-hungry enough
that something like 10,000 of them can blow the heap. This seems like a serious
problem external to this code. In general, it's not unreasonable to want to
write code that iterates over a collection or stream and creates a Future for
each member. Sometimes it's unavoidable.

It's going to be untenable to try to identify every case where a Future chain
could become sufficiently long to tickle this issue, even for those of us aware
of it. For people coming to dart:async for the first time, there's no way
they'll intuit that this issue exists. We need to find a more general solution.

[Lasse Reichstein Nielsen, 2013/05/27 08:04:53]

In this case, I actually think the new version is also more readable, so it's a
pure win.

In general, I doubt memory is going to be the constraining factor, but stack
space might.

We can consider special casing future-chaining, so if a future.then returns a
new future, the original future isn't just made a listener on the new future -
instead all its listeners, current and future, are moved to the new future
directly. That should reduce the nesting.

+        } else {
+          return accumulator.join("\n");

[Bob Nystrom, 2013/05/22 16:23:46]

A StringBuffer seems like a more natural fit for this than a List.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Good point.

+        }
+      });
+    }
+    return collectAll(<String>[]);
+  }
+
   /// Reads the remaining stdout from the process. This should only be called
   /// after kill() or shouldExit().
   Future<String> remainingStdout() {
@@ -254,8 +289,8 @@ class ScheduledProcess {
       throw new StateError("remainingStdout() should only be called after "
           "kill() or shouldExit().");
     }
-
-    return schedule(() => _stdout.toList().then((lines) => lines.join("\n")),
+    List<String> accumulator = <String>[];

[Bob Nystrom, 2013/05/22 16:23:46]

Unused.

[floitsch, 2013/05/22 16:26:29]

Looks unused.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Well spotted.

+    return schedule(() => _concatRest(_stdout),
         "reading the remaining stdout from process '$description'");
   }
 
@@ -267,7 +302,7 @@ class ScheduledProcess {
           "kill() or shouldExit().");
     }
 
-    return schedule(() => _stderr.toList().then((lines) => lines.join("\n")),
+    return schedule(() => _concatRest(_stderr),
         "reading the remaining stderr from process '$description'");
   }