Redo StreamTransformers so they work with Stack traces.

tests/lib/async/stream_event_transformed_test.dart

Index: tests/lib/async/stream_event_transformed_test.dart
diff --git a/tests/lib/async/stream_event_transformed_test.dart b/tests/lib/async/stream_event_transformed_test.dart
new file mode 100644
index 0000000000000000000000000000000000000000..1978bc383bd51a3089e86fe0262d830f944c9d8e
--- /dev/null
+++ b/tests/lib/async/stream_event_transformed_test.dart
@@ -0,0 +1,313 @@
+// 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.
+
+import 'package:expect/expect.dart';
+import 'package:async_helper/async_helper.dart';
+import 'dart:async';
+import 'event_helper.dart';
+
+
+class DecrementingTransformerSink implements EventSink {
+  final outSink;
+  DecrementingTransformerSink(this.outSink);
+
+  void add(int i) => outSink.add(i - 1);
+  void addError(int e, st) => outSink.addError(e - 1, st);
+  void close() => outSink.close();
+}
+
+class FutureWaitingTransformerSink implements EventSink {
+  final outSink;
+  final closeFuture;
+  FutureWaitingTransformerSink(this.outSink, this.closeFuture);
+
+  void add(Future future) { future.then(outSink.add); }
+  void addError(Future e, st) { e.then((val) { outSink.addError(val, st); }); }
+  void close() { closeFuture.whenComplete(outSink.close); }
+}
+
+class ZoneTransformerSink implements EventSink {
+  final outSink;
+  ZoneTransformerSink(this.outSink);
+
+  void add(_) { outSink.add(Zone.current); }
+  void addError(_, st) { outSink.add(Zone.current); }
+  void close() {
+    outSink.add(Zone.current);
+    outSink.close();
+  }
+}
+
+class TypeChangingSink implements EventSink<int, String> {
+  final Sink<String> outSink;
+  TypeChangingSink(this.outSink);
+
+  void add(int data) { outSink.add(data.toString()); }
+  void addError(error, st) { outSink.addError(error, st); }
+  void close() { outSink.close(); }
+}
+
+class SinkTransformer<S, T> implements StreamTransformer<S, T> {
+  final Function sinkMapper;
+  SinkTransformer(this.sinkMapper);
+
+  Stream<T> bind(Stream<S> stream) {
+    return new Stream<T>.eventTransformed(stream, sinkMapper);
+  }
+}
+
+get currentStackTrace {
+  try {
+    throw 0;
+  } catch (e, st) {
+    return st;
+  }
+}
+
+// In most cases the callback will be 'asyncEnd'. Errors are reported
+// asynchronously. We want to give them time to surface before reporting
+// asynchronous tests as done.
+void delayCycles(callback, int nbCycles) {
+  if (nbCycles == 0) {
+    callback();
+    return;
+  }
+  Timer.run(() {
+    delayCycles(callback, nbCycles - 1);
+  });
+}
+
+main() {
+  {
+    // Simple test: use the SinkTransformer (using the Stream.eventTransformed
+    // constructor) to transform a sequence of numbers. This is basically
+    // similar to a map.
+    asyncStart();
+    new Stream.fromIterable([1, 2, 3])
+      .transform(
+          new SinkTransformer((sink) => new DecrementingTransformerSink(sink)))
+      .toList()
+      .then((list) {
+        Expect.listEquals([0, 1, 2], list);
+        asyncEnd();
+      });
+  }
+
+  {
+    // Similar test as above: but this time also transform errors. Also
+    // checks that the stack trace is correctly passed through.
+    asyncStart();
+    var controller;
+    var events = [];
+    var stackTrace = currentStackTrace;
+    controller = new StreamController(onListen: () {
+      controller.add(499);
+      controller.addError(42, stackTrace);
+      controller.close();
+    });
+    controller.stream
+      .transform(
+          new SinkTransformer((sink) => new DecrementingTransformerSink(sink)))
+      .listen((data) {
+        events.add(data);
+      }, onError: (e, st) {
+        events.add(e);
+        events.add(st);
+      }, onDone: () {
+        Expect.listEquals([498, 41, stackTrace], events);
+        asyncEnd();
+      });
+  }
+
+  {
+    // Test that the output sink of the transformer can be used asynchronously.
+    asyncStart();
+    var controller;
+    var events = [];
+    var stackTrace = currentStackTrace;
+    var completer1 = new Completer();
+    var completer2 = new Completer();
+    var completer3 = new Completer();
+    var closeCompleter = new Completer();
+    controller = new StreamController(onListen: () {
+      controller.add(completer1.future);
+      controller.addError(completer2.future, stackTrace);
+      controller.add(completer3.future);
+      controller.close();
+    });
+    controller.stream
+      .transform(
+        new SinkTransformer((sink) =>
+            new FutureWaitingTransformerSink(sink, closeCompleter.future)))
+      .listen((data) {
+        events.add(data);
+      }, onError: (e, st) {
+        events.add(e);
+        events.add(st);
+      }, onDone: () {
+        Expect.listEquals(["error2", stackTrace, "future3", "future1"], events);
+        asyncEnd();
+      });
+    Timer.run(() {
+      completer2.complete("error2");
+      Timer.run(() {
+        completer3.complete("future3");
+        Timer.run(() {
+          completer1.complete("future1");
+          runAsync(closeCompleter.complete);
+        });
+      });
+    });
+  }
+
+  {
+    // Test that the output sink of the transformer can be used asynchronously
+    // and that events are paused if necessary.
+    asyncStart();
+    var controller;
+    var events = [];
+    var stackTrace = currentStackTrace;
+    var completer1 = new Completer.sync();
+    var completer2 = new Completer.sync();
+    var completer3 = new Completer.sync();
+    var closeCompleter = new Completer();
+    controller = new StreamController(onListen: () {
+      controller.add(completer1.future);
+      controller.addError(completer2.future, stackTrace);
+      controller.add(completer3.future);
+      controller.close();
+    });
+    var subscription;
+    completer1.future.then((_) { Expect.isTrue(subscription.isPaused); });
+    completer2.future.then((_) { Expect.isTrue(subscription.isPaused); });
+    completer3.future.then((_) { Expect.isTrue(subscription.isPaused); });
+    subscription = controller.stream
+      .transform(
+        new SinkTransformer((sink) =>
+            new FutureWaitingTransformerSink(sink, closeCompleter.future)))
+      .listen((data) {
+        Expect.isFalse(subscription.isPaused);
+        events.add(data);
+      }, onError: (e, st) {
+        events.add(e);
+        events.add(st);
+      }, onDone: () {
+        Expect.listEquals(["error2", stackTrace, "future3", "future1"], events);
+        asyncEnd();
+      });
+    Timer.run(() {
+      subscription.pause();
+      completer2.complete("error2");
+      Timer.run(() {
+        subscription.resume();
+        Timer.run(() {
+          Expect.listEquals(["error2", stackTrace], events);
+          subscription.pause();
+          completer3.complete("future3");
+          Timer.run(() {
+            subscription.resume();
+            Timer.run(() {
+              Expect.listEquals(["error2", stackTrace, "future3"], events);
+              subscription.pause();
+              completer1.complete("future1");
+              subscription.resume();
+              runAsync(closeCompleter.complete);
+            });
+          });
+        });
+      });
+    });
+  }
+
+  {
+    // Test that the output sink of the transformer reports errors when the
+    // stream is already closed.
+    asyncStart();
+    var controller;
+    var events = [];
+    var stackTrace = currentStackTrace;
+    var completer1 = new Completer();
+    var completer2 = new Completer();
+    var completer3 = new Completer();
+    var closeCompleter = new Completer();
+    controller = new StreamController(onListen: () {
+      controller.add(completer1.future);
+      controller.addError(completer2.future, stackTrace);
+      controller.add(completer3.future);
+      controller.close();
+    });
+
+    bool streamIsDone = false;
+    int errorCount = 0;
+    runZoned(() {
+      controller.stream
+        .transform(
+          new SinkTransformer((sink) =>
+              new FutureWaitingTransformerSink(sink, closeCompleter.future)))
+        .listen((data) {
+          events.add(data);
+        }, onError: (e, st) {
+          events.add(e);
+          events.add(st);
+        }, onDone: () {
+          Expect.listEquals([], events);
+          streamIsDone = true;
+        });
+    }, onError: (e) {
+      Expect.isTrue(e is StateError);
+      errorCount++;
+    });
+    closeCompleter.complete();
+    Timer.run(() {
+      Expect.isTrue(streamIsDone);
+      // Each of the delayed completions should trigger an unhandled error
+      // in the zone the stream was listened to.
+      Timer.run(() { completer1.complete(499); });
+      Timer.run(() { completer2.complete(42); });
+      Timer.run(() { completer3.complete(99); });
+      delayCycles(() {
+        Expect.equals(3, errorCount);
+        asyncEnd();
+      }, 5);
+    });
+  }
+
+  {
+    // Test that the transformer is executed in the zone it was listened to.
+    asyncStart();
+    var stackTrace = currentStackTrace;
+    var events = [];
+    var controller;
+    controller = new StreamController(onListen: () {
+      // Events are added outside the zone.
+      controller.add(499);
+      controller.addError(42, stackTrace);
+      controller.close();
+    });
+    Zone zone = Zone.current.fork();
+    var stream = controller.stream.transform(
+        new SinkTransformer((sink) => new ZoneTransformerSink(sink)));
+    zone.run(() {
+        stream.listen((data) {
+          events.add(data);
+        }, onDone: () {
+          Expect.listEquals([zone, zone, zone], events);
+          delayCycles(asyncEnd, 3);
+        });
+    });
+  }
+
+  {
+    // Just make sure that the generic types are correct everywhere.
+    asyncStart();
+    new Stream.fromIterable([1, 2, 3])
+      .transform(new SinkTransformer<int, String>(
+          (sink) => new TypeChangingSink(sink)))
+      .toList()
+      .then((list) {
+        Expect.listEquals(["1", "2", "3"], list);
+        asyncEnd();
+      });
+  }
+}