Stream blocks of timeline events over the service protocol

runtime/observatory/tests/service/service_test_common.dart

 // Copyright (c) 2016, 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 service_test_common;
 
 import 'dart:async';
 import 'package:observatory/service_common.dart';
 import 'package:unittest/unittest.dart';
 
 typedef Future IsolateTest(Isolate isolate);
 typedef Future VMTest(VM vm);
 
+Map<String, StreamSubscription> streamSubscriptions = {};
+
+Future subscribeToStream(VM vm, String streamName, onEvent) async {
+  assert(streamSubscriptions[streamName] == null);
+
+  Stream stream = await vm.getEventStream(streamName);
+  StreamSubscription subscription = stream.listen(onEvent);
+  streamSubscriptions[streamName] = subscription;
+}
+
+Future cancelStreamSubscription(String streamName) async {
+  StreamSubscription subscription = streamSubscriptions[streamName];
+  subscription.cancel();
+  streamSubscriptions.remove(streamName);
+}
+
 Future asyncStepOver(Isolate isolate) async {
   final Completer pausedAtSyntheticBreakpoint = new Completer();
   StreamSubscription subscription;
 
   // Cancel the subscription.
   cancelSubscription() {
     if (subscription != null) {
       subscription.cancel();
       subscription = null;
     }
@@ skipping 203 matching lines @@
 
 Future<Instance> rootLibraryFieldValue(Isolate isolate,
                                        String fieldName) async {
   Library rootLib = await isolate.rootLibrary.load();
   Field field = rootLib.variables.singleWhere((v) => v.name == fieldName);
   await field.load();
   Instance value = field.staticValue;
   await value.load();
   return value;
 }