Make StreamSubscription be the active part of a stream.

tests/lib/async/stream_state_test.dart

Index: tests/lib/async/stream_state_test.dart
diff --git a/tests/lib/async/stream_state_test.dart b/tests/lib/async/stream_state_test.dart
index d6d873f028a513a22b9f7d7bb496d80e4540092b..08cb7a164a460bbac2fb73402538325fb57b232e 100644
--- a/tests/lib/async/stream_state_test.dart
+++ b/tests/lib/async/stream_state_test.dart
@@ -20,10 +20,10 @@ mainTest(bool broadcast) {
   var p = broadcast ? "BC" : "SC";
   test("$p-sub-data-done", () {
     var t = new StreamProtocolTest(broadcast);
-    t..expectSubscription(true, false)
+    t..expectSubscription()
      ..expectData(42)
      ..expectDone()
-     ..expectSubscription(false, false);
+     ..expectCancel();
     t..subscribe()..add(42)..close();
   });
 
@@ -32,33 +32,33 @@ mainTest(bool broadcast) {
     if (broadcast) {
       t..expectDone();
     } else {
-      t..expectSubscription(true, false)
+      t..expectSubscription()
        ..expectData(42)
        ..expectDone()
-       ..expectSubscription(false, false);
+       ..expectCancel();
     }
     t..add(42)..close()..subscribe();
   });
 
   test("$p-sub-data/pause+resume-done", () {
     var t = new StreamProtocolTest(broadcast);
-    t..expectSubscription(true, false)
+    t..expectSubscription()
      ..expectData(42, () {
          t.pause();
          t.resume();
          t.close();
        })
      ..expectDone()
-     ..expectSubscription(false, false);
+     ..expectCancel();
     t..subscribe()..add(42);
   });
 
   test("$p-sub-data-unsubonerror", () {
     var t = new StreamProtocolTest(broadcast);
-    t..expectSubscription(true, false)
+    t..expectSubscription()
      ..expectData(42)
      ..expectError("bad")
-     ..expectSubscription(false, !broadcast);
+     ..expectCancel();
     t..subscribe(cancelOnError: true)
      ..add(42)
      ..error("bad")
@@ -68,12 +68,12 @@ mainTest(bool broadcast) {
 
   test("$p-sub-data-no-unsubonerror", () {
     var t = new StreamProtocolTest(broadcast);
-    t..expectSubscription(true, false)
+    t..expectSubscription()
      ..expectData(42)
      ..expectError("bad")
      ..expectData(43)
      ..expectDone()
-     ..expectSubscription(false, false);
+     ..expectCancel();
     t..subscribe(cancelOnError: false)
      ..add(42)
      ..error("bad")
@@ -83,62 +83,15 @@ mainTest(bool broadcast) {
 
   test("$p-pause-resume-during-event", () {
     var t = new StreamProtocolTest(broadcast);
-    t..expectSubscription(true, false)
+    t..expectSubscription()
      ..expectData(42, () {
        t.pause();
        t.resume();
      })
      ..expectDone()
-     ..expectSubscription(false, false);
+     ..expectCancel();
     t..subscribe()
      ..add(42)
      ..close();
   });
-
-  test("$p-cancel-sub-during-event", () {
-    var t = new StreamProtocolTest(broadcast);
-    t..expectSubscription(true, false)
-     ..expectData(42, () {
-       t.cancel();
-       t.subscribe();
-     })
-     ..expectData(43)
-     ..expectDone()
-     ..expectSubscription(false, false);
-    t..subscribe()
-     ..add(42)
-     ..add(43)
-     ..close();
-  });
-
-  test("$p-cancel-sub-during-callback", () {
-    var t = new StreamProtocolTest(broadcast);
-    t..expectSubscription(true, false)
-     ..expectData(42, () {
-       t.pause();
-     })
-     ..expectPause(true, () {
-       t.cancel();  // Cancels pause
-       t.subscribe();
-     })
-     ..expectPause(false)
-     ..expectData(43)
-     ..expectDone()
-     ..expectSubscription(false, false);
-    t..subscribe()
-     ..add(42)
-     ..add(43)
-     ..close();
-  });
-
-  test("$p-sub-after-done-is-done", () {
-    var t = new StreamProtocolTest(broadcast);
-    t..expectSubscription(true, false)
-     ..expectDone()
-     ..expectSubscription(false, false)
-     ..expectDone();
-    t..subscribe()
-     ..close()
-     ..subscribe();  // Subscribe after done does not cause callbacks at all.
-  });
 }