Make StreamSubscription be the active part of a stream.

sdk/lib/_internal/pub/test/error_group_test.dart

Index: sdk/lib/_internal/pub/test/error_group_test.dart
diff --git a/sdk/lib/_internal/pub/test/error_group_test.dart b/sdk/lib/_internal/pub/test/error_group_test.dart
index ff2f699dbacd75574e7b677a65c1bd600896746b..ec5e6de6d4aa529d5b466cb8c64233c014729e97 100644
--- a/sdk/lib/_internal/pub/test/error_group_test.dart
+++ b/sdk/lib/_internal/pub/test/error_group_test.dart
@@ -210,8 +210,16 @@ main() {
     });
 
     test('should pass through values from the stream', () {
-      expect(stream.elementAt(0), completion(equals(1)));
-      expect(stream.elementAt(1), completion(equals(2)));
+      StreamIterator iter = new StreamIterator(stream);
+      iter.moveNext().then((hasNext) {
+        expect(hasNext, isTrue);
+        expect(iter.current, equals(1));
+        iter.moveNext().then((hasNext) {
+          expect(hasNext, isTrue);
+          expect(iter.current, equals(2));
+          expect(iter.moveNext(), completion(isFalse));

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

I think the `completion` is called too late. The test will not see that this is
an asynchronous test before it has already shut down.

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

Then how do you write a test that expects an async result and can only continue
after that?

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

expectAsync the first asynchronous part. then the next ...

+        });
+      });
       expect(errorGroup.done, completes);
 
       controller..add(1)..add(2)..close();
@@ -232,15 +240,13 @@ main() {
     });
 
     test('should pass a signaled exception to the stream if it has a listener '
-        'and should unsubscribe that stream', () {
-      expect(stream.first, throwsFormatException);
+         'and should unsubscribe that stream', () {
       // errorGroup shouldn't top-level the exception
       errorGroup.signalError(new FormatException());
 
       expect(stream.first.catchError((_) {
         controller.add('value');
-        return stream.isEmpty;
-      }), completion(isTrue));
+      }), completes);
     });
 
     test('should notify the error group of a signaled exception even if the '
@@ -255,7 +261,6 @@ main() {
       expect(errorGroup.done, completes);
       controller.add('value');
       controller.close();
-

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

keep line.

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

ok.

       // Now that broadcast controllers have been removed a listener should
       // see the value that has been put into the controller.
       expect(errorGroup.done.then((_) => stream.toList()),
@@ -345,10 +350,12 @@ main() {
 
     test("shouldn't throw a top-level exception if a stream receives an error "
         "after the other listened stream completes", () {
-      expect(stream1.toList(), completion(equals(['value1', 'value2'])));
+      var signal = new Completer();
+      expect(stream1.toList().whenComplete(signal.complete),
+             completion(equals(['value1', 'value2'])));
       controller1..add('value1')..add('value2')..close();
 
-      expect(stream1.toList().then((_) {
+      expect(signal.future.then((_) {
         // shouldn't cause a top-level exception
         controller2.addError(new FormatException());
       }), completes);
@@ -356,10 +363,12 @@ main() {
 
     test("shouldn't throw a top-level exception if an error is signaled after "
         "one listened stream completes", () {
-      expect(stream1.toList(), completion(equals(['value1', 'value2'])));
+      var signal = new Completer();
+      expect(stream1.toList().whenComplete(signal.complete),
+             completion(equals(['value1', 'value2'])));
       controller1..add('value1')..add('value2')..close();
 
-      expect(stream1.toList().then((_) {
+      expect(signal.future.then((_) {
         // shouldn't cause a top-level exception
         errorGroup.signalError(new FormatException());
       }), completes);
@@ -419,10 +428,12 @@ main() {
 
     test("shouldn't throw a top-level exception if the future receives an "
         "error after the listened stream completes", () {
-      expect(stream.toList(), completion(equals(['value1', 'value2'])));
+      var signal = new Completer();
+      expect(stream.toList().whenComplete(signal.complete),
+             completion(equals(['value1', 'value2'])));
       controller..add('value1')..add('value2')..close();
 
-      expect(stream.toList().then((_) {
+      expect(signal.future.then((_) {
         // shouldn't cause a top-level exception
         completer.completeError(new FormatException());
       }), completes);