Make VM event streams look like real dart streams.

runtime/observatory/tests/service/break_on_activation_test.dart

Index: runtime/observatory/tests/service/break_on_activation_test.dart
diff --git a/runtime/observatory/tests/service/break_on_activation_test.dart b/runtime/observatory/tests/service/break_on_activation_test.dart
index b26695406bf9bd99b17b1e6eb1f283aaafaecb82..d7418a50478002a26586e9e50495ea667112fc4f 100644
--- a/runtime/observatory/tests/service/break_on_activation_test.dart
+++ b/runtime/observatory/tests/service/break_on_activation_test.dart
@@ -54,14 +54,14 @@ var tests = [
 
   var breaksHit = 0;
 
-  var subscription;
-  subscription = isolate.vm.events.stream.listen((ServiceEvent event) {
-    if (event.kind == ServiceEvent.kPauseBreakpoint) {
-      print("Hit breakpoint ${event.breakpoint}");
-      breaksHit++;
-      isolate.resume();
-    }
-  });
+  var subscriptionFuture = isolate.vm.listenEventStream(
+      VM.kDebugStream, (ServiceEvent event) {
+        if (event.kind == ServiceEvent.kPauseBreakpoint) {
+          print("Hit breakpoint ${event.breakpoint}");
+          breaksHit++;
+          isolate.resume();
+        }
+      });
 
   valueOfField(String name) async {
     var field = rootLib.variables.singleWhere((v) => v.name == name);
@@ -91,7 +91,7 @@ var tests = [
   expect(res is Instance, isTrue); // Not error.
   expect(breaksHit, equals(1));
 
-  await subscription.cancel();
+  await cancelFutureSubscription(subscriptionFuture);
 },
 
 (Isolate isolate) async {
@@ -99,14 +99,14 @@ var tests = [
 
   var breaksHit = 0;
 
-  var subscription;
-  subscription = isolate.vm.events.stream.listen((ServiceEvent event) {
-    if (event.kind == ServiceEvent.kPauseBreakpoint) {
-      print("Hit breakpoint ${event.breakpoint}");
-      breaksHit++;
-      isolate.resume();
-    }
-  });
+  var subscriptionFuture = isolate.vm.listenEventStream(
+      VM.kDebugStream, (ServiceEvent event) {
+        if (event.kind == ServiceEvent.kPauseBreakpoint) {
+          print("Hit breakpoint ${event.breakpoint}");
+          breaksHit++;
+          isolate.resume();
+        }
+      });
 
   valueOfField(String name) async {
     var field = rootLib.variables.singleWhere((v) => v.name == name);
@@ -135,7 +135,7 @@ var tests = [
   expect(res is Instance, isTrue); // Not error.
   expect(breaksHit, equals(1));
 
-  await subscription.cancel();
+  await cancelFutureSubscription(subscriptionFuture);
 },
 
 (Isolate isolate) async {
@@ -143,14 +143,14 @@ var tests = [
 
   var breaksHit = 0;
 
-  var subscription;
-  subscription = isolate.vm.events.stream.listen((ServiceEvent event) {
-    if (event.kind == ServiceEvent.kPauseBreakpoint) {
-      print("Hit breakpoint ${event.breakpoint}");
-      breaksHit++;
-      isolate.resume();
-    }
-  });
+  var subscriptionFuture = isolate.vm.listenEventStream(
+      VM.kDebugStream, (ServiceEvent event) {
+        if (event.kind == ServiceEvent.kPauseBreakpoint) {
+          print("Hit breakpoint ${event.breakpoint}");
+          breaksHit++;
+          isolate.resume();
+        }
+      });
 
   valueOfField(String name) async {
     var field = rootLib.variables.singleWhere((v) => v.name == name);
@@ -200,7 +200,7 @@ var tests = [
   expect(res is Instance, isTrue); // Not error.
   expect(breaksHit, equals(4));
 
-  await subscription.cancel();
+  await cancelFutureSubscription(subscriptionFuture);
 },
 
 ];