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) {

[floitsch, 2013/05/24 15:53:17]

iter.moveNext().then(expectAsync1((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));
+        });
+      });
       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', () {
+         'and should unsubscribe that stream', () {
+      // errorGroup shouldn't top-level the exception
       expect(stream.first, throwsFormatException);
-      // errorGroup shouldn't top-level the exception
       errorGroup.signalError(new FormatException());
 
-      expect(stream.first.catchError((_) {
+      expect(new Future(() {

[floitsch, 2013/05/24 15:53:17]

Does this part of the test still make sense now?

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

It sort of does, but it has a race condition. The [controller.add] call is now
not guaranteed to run after [stream.first] completes. I think the previous
version of the test is still correct.

         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 "
@@ skipping 84 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);
     });
   });
 }