Make StreamSubscription be the active part of a stream.

sdk/lib/_internal/pub/test/error_group_test.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 error_group_test;
 
 import 'dart:async';
 
 import 'package:unittest/unittest.dart';
 
@@ skipping 192 matching lines @@
     StreamController controller;
     Stream stream;
 
     setUp(() {
       errorGroup = new ErrorGroup();
       controller = new StreamController();
       stream = errorGroup.registerStream(controller.stream.asBroadcastStream());
     });
 
     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();
     });
 
     test('should pass through an error from the stream if it has a '
         'listener', () {
       expect(stream.first, throwsFormatException);
       // errorGroup shouldn't top-level the exception
       controller.addError(new FormatException());
     });
 
     test('should notify the error group of an exception from the stream even '
         'if it has a listener', () {
       expect(stream.first, throwsFormatException);
       expect(errorGroup.done, throwsFormatException);
       controller.addError(new FormatException());
     });
 
     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 '
         'stream has a listener', () {
       expect(stream.first, throwsFormatException);
       expect(errorGroup.done, throwsFormatException);
       errorGroup.signalError(new FormatException());
     });
 
     test("should see one value and complete .done when the stream is done even "
          "if the stream doesn't have a listener", () {
       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()),
           completion(equals(['value'])));
     });
 
   });
 
   group('with a single single-subscription stream', () {
     StreamController controller;
@@ skipping 69 matching lines @@
       expect(stream1.toList(), completion(equals(['value1.1', 'value1.2'])));
       expect(stream2.toList(), completion(equals(['value2.1', 'value2.2'])));
       expect(errorGroup.done, completes);
 
       controller1..add('value1.1')..add('value1.2')..close();
       controller2..add('value2.1')..add('value2.2')..close();
     });
 
     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);
     });
 
     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);
     });
   });
 
   group('with a stream and a future', () {
     StreamController controller;
     Stream stream;
     Completer completer;
@@ skipping 39 matching lines @@
       completer.complete('value');
 
       expect(future.then((_) {
         // shouldn't cause a top-level exception
         controller.addError(new FormatException());
       }), completes);
     });
 
     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);
     });
   });
 }