Revert "Add the streamListen and streamCancel rpcs to the vm service."

runtime/vm/service/vmservice.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 vmservice;
 
 import 'dart:async';
 import 'dart:convert';
 import 'dart:isolate';
 import 'dart:typed_data';
@@ skipping 18 matching lines @@
 
   /// Collection of currently running isolates.
   RunningIsolates runningIsolates = new RunningIsolates();
 
   /// A port used to receive events from the VM.
   final RawReceivePort eventPort;
 
   ShutdownCallback onShutdown;
 
   void _addClient(Client client) {
-    assert(client.streams.isEmpty);
     clients.add(client);
   }
 
   void _removeClient(Client client) {
     clients.remove(client);
-    for (var streamId in client.streams) {
-      if (!_isAnyClientSubscribed(streamId)) {
-        _vmCancelStream(streamId);
+  }
+
+  void _eventMessageHandler(dynamic eventMessage) {
+    for (var client in clients) {
+      if (client.sendEvents) {
+        client.post(eventMessage);
       }
     }
   }
-
-  void _eventMessageHandler(List eventMessage) {
-    var streamId = eventMessage[0];
-    var event = eventMessage[1];
-    for (var client in clients) {
-      if (client.sendEvents && client.streams.contains(streamId)) {
-        client.post(event);
-      }
-    }
-  }
 
   void _controlMessageHandler(int code,
                               int portId,
                               SendPort sp,
                               String name) {
     switch (code) {
       case Constants.ISOLATE_STARTUP_MESSAGE_ID:
         runningIsolates.isolateStartup(portId, sp, name);
       break;
       case Constants.ISOLATE_SHUTDOWN_MESSAGE_ID:
@@ skipping 11 matching lines @@
       client.close();
     }
     // Call embedder shutdown hook after the internal shutdown.
     if (onShutdown != null) {
       onShutdown();
     }
     _onExit();
   }
 
   void messageHandler(message) {
+    if (message is String) {
+      // This is an event intended for all clients.
+      _eventMessageHandler(message);
+      return;
+    }
+    if (message is Uint8List) {
+      // This is "raw" data intended for a specific client.
+      //
+      // TODO(turnidge): Do not broadcast this data to all clients.
+      _eventMessageHandler(message);
+      return;
+    }
     if (message is List) {
-      if (message.length == 2) {
-        // This is an event.
-        assert(message[0] is String);
-        assert(message[1] is String || message[1] is Uint8List);
-        _eventMessageHandler(message);
-        return;
-      }
+      // This is an internal vm service event.
       if (message.length == 1) {
         // This is a control message directing the vm service to exit.
         assert(message[0] == Constants.SERVICE_EXIT_MESSAGE_ID);
         _exit();
         return;
       }
       if (message.length == 4) {
         // This is a message informing us of the birth or death of an
         // isolate.
         _controlMessageHandler(message[0], message[1], message[2], message[3]);
         return;
       }
     }
+
     Logger.root.severe(
         'Internal vm-service error: ignoring illegal message: $message');
   }
 
   void _notSupported(_) {
     throw new UnimplementedError('Service script loading not supported.');
   }
 
   VMService._internal()
       : eventPort = isolateLifecyclePort {
@@ skipping 13 matching lines @@
     var members = [];
     var result = {};
     clients.forEach((client) {
       members.add(client.toJson());
     });
     result['type'] = 'ClientList';
     result['members'] = members;
     message.setResponse(JSON.encode(result));
   }
 
-  // These must be kept in sync with the declarations in vm/json_stream.h.
-  static const _kInvalidStream = 100;
-  static const _kStreamAlreadySubscribed = 101;
-  static const _kStreamNotSubscribed = 102;
-
-  var _errorMessages = {
-    _kInvalidStream: 'Invalid stream',
-    _kStreamAlreadySubscribed: 'Stream already subscribed',
-    _kStreamNotSubscribed: 'Stream not subscribed',
-  };
-
-  String _encodeError(Message message, int code) {
-    var response = {
-      'id' : message.serial,
-      'error' : {
-        'code': code,
-        'message': _errorMessages[code],
-      },
-    };
-    return JSON.encode(response);
-  }
-
-  String _encodeResult(Message message, Map result) {
-    var response = {
-      'id' : message.serial,
-      'result' : result,
-    };
-    return JSON.encode(response);
-  }
-
-  bool _isValidStream(String streamId) {
-    var validStreams = [ 'Isolate', 'Debug', 'GC', '_Echo', '_Graph' ];
-    return validStreams.contains(streamId);
-  }
-
-  bool _isAnyClientSubscribed(String streamId) {
-    for (var client in clients) {
-      if (client.streams.contains(streamId)) {
-        return true;
-      }
-    }
-    return false;
-  }
-
-  Future<String> _streamListen(Message message) async {
-    var client = message.client;
-    var streamId = message.params['streamId'];
-
-    if (!_isValidStream(streamId)) {
-      return _encodeError(message, _kInvalidStream);
-    }
-    if (client.streams.contains(streamId)) {
-      return _encodeError(message, _kStreamAlreadySubscribed);
-    }
-    if (!_isAnyClientSubscribed(streamId)) {
-      _vmListenStream(streamId);
-    }
-    client.streams.add(streamId);
-
-    var result = { 'type' : 'Success' };
-    return _encodeResult(message, result);
-  }
-
-  Future<String> _streamCancel(Message message) async {
-    var client = message.client;
-    var streamId = message.params['streamId'];
-
-    if (!_isValidStream(streamId)) {
-      return _encodeError(message, _kInvalidStream);
-    }
-    if (!client.streams.contains(streamId)) {
-      return _encodeError(message, _kStreamNotSubscribed);
-    }
-    client.streams.remove(streamId);
-    if (!_isAnyClientSubscribed(streamId)) {
-      _vmCancelStream(streamId);
-    }
-
-    var result = { 'type' : 'Success' };
-    return _encodeResult(message, result);
-  }
-
   // TODO(johnmccutchan): Turn this into a command line tool that uses the
   // service library.
-  Future<String> _getCrashDump(Message message) async {
-    var client = message.client;
+  Future<String> _getCrashDump() async {
     final perIsolateRequests = [
         // ?isolateId=<isolate id> will be appended to each of these requests.
         // Isolate information.
         Uri.parse('getIsolate'),
         // State of heap.
         Uri.parse('_getAllocationProfile'),
         // Call stack + local variables.
         Uri.parse('getStack?_full=true'),
     ];
 
     // Snapshot of running isolates.
     var isolates = runningIsolates.isolates.values.toList();
 
     // Collect the mapping from request uris to responses.
     var responses = {
     };
 
     // Request VM.
     var getVM = Uri.parse('getVM');
     var getVmResponse = JSON.decode(
-        await new Message.fromUri(client, getVM).sendToVM());
+        await new Message.fromUri(getVM).sendToVM());
     responses[getVM.toString()] = getVmResponse['result'];
 
     // Request command line flags.
     var getFlagList = Uri.parse('getFlagList');
     var getFlagListResponse = JSON.decode(
-        await new Message.fromUri(client, getFlagList).sendToVM());
+        await new Message.fromUri(getFlagList).sendToVM());
     responses[getFlagList.toString()] = getFlagListResponse['result'];
 
     // Make requests to each isolate.
     for (var isolate in isolates) {
       for (var request in perIsolateRequests) {
         var message = new Message.forIsolate(request, isolate);
         // Decode the JSON and and insert it into the map. The map key
         // is the request Uri.
         var response = JSON.decode(await isolate.route(message));
         responses[message.toUri().toString()] = response['result'];
       }
       // Dump the object id ring requests.
       var message =
-          new Message.forIsolate(client, Uri.parse('_dumpIdZone'), isolate);
+          new Message.forIsolate(Uri.parse('_dumpIdZone'), isolate);
       var response = JSON.decode(await isolate.route(message));
       // Insert getObject requests into responses map.
       for (var object in response['result']['objects']) {
         final requestUri =
             'getObject&isolateId=${isolate.serviceId}?objectId=${object["id"]}';
         responses[requestUri] = object;
       }
     }
 
     // Encode the entire crash dump.
-    return _encodeResult(message, responses);
+    return JSON.encode({
+      'id' : null,
+      'result' : responses,
+    });
   }
 
   Future<String> route(Message message) {
     if (message.completed) {
       return message.response;
     }
     // TODO(turnidge): Update to json rpc.  BEFORE SUBMIT.
     if ((message.path.length == 1) && (message.path[0] == 'clients')) {
       _clientCollection(message);
       return message.response;
     }
     if (message.method == '_getCrashDump') {
-      return _getCrashDump(message);
-    }
-    if (message.method == 'streamListen') {
-      return _streamListen(message);
-    }
-    if (message.method == 'streamCancel') {
-      return _streamCancel(message);
+      return _getCrashDump();
     }
     if (message.params['isolateId'] != null) {
       return runningIsolates.route(message);
     }
     return message.sendToVM();
   }
 }
 
 RawReceivePort boot() {
   // Return the port we expect isolate startup and shutdown messages on.
   return isolateLifecyclePort;
 }
 
 void _registerIsolate(int port_id, SendPort sp, String name) {
   var service = new VMService();
   service.runningIsolates.isolateStartup(port_id, sp, name);
 }
 
 void _onStart() native "VMService_OnStart";
 
 void _onExit() native "VMService_OnExit";
-
-void _vmListenStream(String streamId) native "VMService_ListenStream";
-
-void _vmCancelStream(String streamId) native "VMService_CancelStream";