2nd attempt at adding streamListen/streamCancel to the service protocol.

runtime/vm/service.cc

 // 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.
 
 #include "vm/service.h"
 
 #include "include/dart_api.h"
 #include "platform/globals.h"
 
 #include "vm/compiler.h"
@@ skipping 56 matching lines @@
   return zone->PrintToString("objects/%" Pd "", id);
 }
 
 
 // TODO(johnmccutchan): Unify embedder service handler lists and their APIs.
 EmbedderServiceHandler* Service::isolate_service_handler_head_ = NULL;
 EmbedderServiceHandler* Service::root_service_handler_head_ = NULL;
 struct ServiceMethodDescriptor;
 ServiceMethodDescriptor* FindMethod(const char* method_name);
 
+// TODO(turnidge): Build a general framework later.  For now, we have
+// a small set of well-known streams.
+bool Service::needs_isolate_events_ = false;
+bool Service::needs_debug_events_ = false;
+bool Service::needs_gc_events_ = false;
+bool Service::needs_echo_events_ = false;
+bool Service::needs_graph_events_ = false;
+
+void Service::ListenStream(const char* stream_id) {
+  if (FLAG_trace_service) {
+    OS::Print("vm-service: starting stream '%s'\n",
+              stream_id);
+  }
+  if (strcmp(stream_id, "Isolate") == 0) {
+    needs_isolate_events_ = true;
+  } else if (strcmp(stream_id, "Debug") == 0) {
+    needs_debug_events_ = true;
+  } else if (strcmp(stream_id, "GC") == 0) {
+    needs_gc_events_ = true;
+  } else if (strcmp(stream_id, "_Echo") == 0) {
+    needs_echo_events_ = true;
+  } else if (strcmp(stream_id, "_Graph") == 0) {
+    needs_graph_events_ = true;
+  } else {
+    UNREACHABLE();
+  }
+}
+
+void Service::CancelStream(const char* stream_id) {
+  if (FLAG_trace_service) {
+    OS::Print("vm-service: stopping stream '%s'\n",
+              stream_id);
+  }
+  if (strcmp(stream_id, "Isolate") == 0) {
+    needs_isolate_events_ = false;
+  } else if (strcmp(stream_id, "Debug") == 0) {
+    needs_debug_events_ = false;
+  } else if (strcmp(stream_id, "GC") == 0) {
+    needs_gc_events_ = false;
+  } else if (strcmp(stream_id, "_Echo") == 0) {
+    needs_echo_events_ = false;
+  } else if (strcmp(stream_id, "_Graph") == 0) {
+    needs_graph_events_ = false;
+  } else {
+    UNREACHABLE();
+  }
+}
+
 static uint8_t* allocator(uint8_t* ptr, intptr_t old_size, intptr_t new_size) {
   void* new_ptr = realloc(reinterpret_cast<void*>(ptr), new_size);
   return reinterpret_cast<uint8_t*>(new_ptr);
 }
 
 static void PrintMissingParamError(JSONStream* js,
                                    const char* param) {
   js->PrintError(kInvalidParams,
                  "%s expects the '%s' parameter", js->method(), param);
 }
@@ skipping 403 matching lines @@
   InvokeMethod(isolate, msg_instance);
 }
 
 
 void Service::HandleIsolateMessage(Isolate* isolate, const Array& msg) {
   ASSERT(isolate != NULL);
   InvokeMethod(isolate, msg);
 }
 
 
-bool Service::NeedsEvents() {
-  return ServiceIsolate::IsRunning();
-}
-
-
-void Service::SendEvent(intptr_t eventType,
-                        const Object& eventMessage) {
+void Service::SendEvent(const char* stream_id,
+                        const char* event_type,
+                        const Object& event_message) {
   ASSERT(!ServiceIsolate::IsServiceIsolateDescendant(Isolate::Current()));
   if (!ServiceIsolate::IsRunning()) {
     return;
   }
   Isolate* isolate = Isolate::Current();
   ASSERT(isolate != NULL);
   HANDLESCOPE(isolate);
 
+  const Array& list = Array::Handle(Array::New(2));
+  ASSERT(!list.IsNull());
+  const String& stream_id_str = String::Handle(String::New(stream_id));
+  list.SetAt(0, stream_id_str);
+  list.SetAt(1, event_message);
+
   // Push the event to port_.
   uint8_t* data = NULL;
   MessageWriter writer(&data, &allocator, false);
-  writer.WriteMessage(eventMessage);
+  writer.WriteMessage(list);
   intptr_t len = writer.BytesWritten();
   if (FLAG_trace_service) {
-    OS::Print("vm-service: Pushing event of type %" Pd ", len %" Pd "\n",
-              eventType, len);
+    OS::Print(
+        "vm-service: Pushing event of type %s to stream %s, len %" Pd "\n",
+        event_type, stream_id, len);
   }
   // TODO(turnidge): For now we ignore failure to send an event.  Revisit?
   PortMap::PostMessage(
       new Message(ServiceIsolate::Port(), data, len, Message::kNormalPriority));
 }
 
 
-void Service::SendEvent(const String& meta,
-                        const uint8_t* data,
-                        intptr_t size) {
+void Service::SendEventWithData(const char* stream_id,
+                                const char* event_type,
+                                const String& meta,
+                                const uint8_t* data,
+                                intptr_t size) {
   // Bitstream: [meta data size (big-endian 64 bit)] [meta data (UTF-8)] [data]
   const intptr_t meta_bytes = Utf8::Length(meta);
   const intptr_t total_bytes = sizeof(uint64_t) + meta_bytes + size;
   const TypedData& message = TypedData::Handle(
       TypedData::New(kTypedDataUint8ArrayCid, total_bytes));
   intptr_t offset = 0;
   // TODO(koda): Rename these methods SetHostUint64, etc.
   message.SetUint64(0, Utils::HostToBigEndian64(meta_bytes));
   offset += sizeof(uint64_t);
   {
     NoSafepointScope no_safepoint;
     meta.ToUTF8(static_cast<uint8_t*>(message.DataAddr(offset)), meta_bytes);
     offset += meta_bytes;
   }
   // TODO(koda): It would be nice to avoid this copy (requires changes to
   // MessageWriter code).
   {
     NoSafepointScope no_safepoint;
     memmove(message.DataAddr(offset), data, size);
     offset += size;
   }
   ASSERT(offset == total_bytes);
-  // TODO(turnidge): Pass the real eventType here.
-  SendEvent(0, message);
+  SendEvent(stream_id, event_type, message);
 }
 
 
 void Service::HandleEvent(ServiceEvent* event) {
   if (ServiceIsolate::IsServiceIsolateDescendant(event->isolate())) {
     return;
   }
   JSONStream js;
+  const char* stream_id = event->stream_id();
+  ASSERT(stream_id != NULL);
   {
     JSONObject jsobj(&js);
     jsobj.AddProperty("event", event);
+    jsobj.AddProperty("streamId", stream_id);
   }
   const String& message = String::Handle(String::New(js.ToCString()));
-  SendEvent(event->type(), message);
+  SendEvent(stream_id, ServiceEvent::EventTypeToCString(event->type()),
+            message);
 }
 
 
 class EmbedderServiceHandler {
  public:
   explicit EmbedderServiceHandler(const char* name) : name_(NULL),
                                                       callback_(NULL),
                                                       user_data_(NULL),
                                                       next_(NULL) {
     ASSERT(name != NULL);
@@ skipping 28 matching lines @@
   EmbedderServiceHandler* next_;
 };
 
 
 void Service::EmbedderHandleMessage(EmbedderServiceHandler* handler,
                                     JSONStream* js) {
   ASSERT(handler != NULL);
   Dart_ServiceRequestCallback callback = handler->callback();
   ASSERT(callback != NULL);
   const char* r = NULL;
-  const char* name = js->method();
+  const char* method = js->method();
   const char** keys = js->param_keys();
   const char** values = js->param_values();
-  r = callback(name, keys, values, js->num_params(), handler->user_data());
+  r = callback(method, keys, values, js->num_params(), handler->user_data());
   ASSERT(r != NULL);
   // TODO(johnmccutchan): Allow for NULL returns?
   TextBuffer* buffer = js->buffer();
   buffer->AddString(r);
   free(const_cast<char*>(r));
 }
 
 
 void Service::RegisterIsolateEmbedderCallback(
     const char* name,
@@ skipping 134 matching lines @@
     JSONObject jsobj(&js);
     {
       JSONObject event(&jsobj, "event");
       event.AddProperty("type", "Event");
       event.AddProperty("kind", "_Echo");
       event.AddProperty("isolate", isolate);
       if (text != NULL) {
         event.AddProperty("text", text);
       }
     }
+    jsobj.AddProperty("streamId", "_Echo");
   }
   const String& message = String::Handle(String::New(js.ToCString()));
   uint8_t data[] = {0, 128, 255};
-  SendEvent(message, data, sizeof(data));
+  SendEventWithData("_Echo", "_Echo", message, data, sizeof(data));
 }
 
 
 static bool TriggerEchoEvent(Isolate* isolate, JSONStream* js) {
-  Service::SendEchoEvent(isolate, js->LookupParam("text"));
+  if (Service::NeedsEchoEvents()) {
+    Service::SendEchoEvent(isolate, js->LookupParam("text"));
+  }
   JSONObject jsobj(js);
   return HandleCommonEcho(&jsobj, js);
 }
 
 
 static bool DumpIdZone(Isolate* isolate, JSONStream* js) {
   // TODO(johnmccutchan): Respect _idZone parameter passed to RPC. For now,
   // always send the ObjectIdRing.
   //
   ObjectIdRing* ring = isolate->object_id_ring();
@@ skipping 1288 matching lines @@
 static const MethodParameter* resume_params[] = {
   ISOLATE_PARAMETER,
   NULL,
 };
 
 
 static bool Resume(Isolate* isolate, JSONStream* js) {
   const char* step_param = js->LookupParam("step");
   if (isolate->message_handler()->paused_on_start()) {
     isolate->message_handler()->set_pause_on_start(false);
-    {
+    if (Service::NeedsDebugEvents()) {
       ServiceEvent event(isolate, ServiceEvent::kResume);
       Service::HandleEvent(&event);
     }
     PrintSuccess(js);
     return true;
   }
   if (isolate->message_handler()->paused_on_exit()) {
     isolate->message_handler()->set_pause_on_exit(false);
     // We don't send a resume event because we will be exiting.
     PrintSuccess(js);
@@ skipping 148 matching lines @@
 }
 
 
 static const MethodParameter* request_heap_snapshot_params[] = {
   ISOLATE_PARAMETER,
   NULL,
 };
 
 
 static bool RequestHeapSnapshot(Isolate* isolate, JSONStream* js) {
-  Service::SendGraphEvent(isolate);
+  if (Service::NeedsGraphEvents()) {
+    Service::SendGraphEvent(isolate);
+  }
   // TODO(koda): Provide some id that ties this request to async response(s).
   JSONObject jsobj(js);
   jsobj.AddProperty("type", "OK");
   return true;
 }
 
 
 void Service::SendGraphEvent(Isolate* isolate) {
   uint8_t* buffer = NULL;
   WriteStream stream(&buffer, &allocator, 1 * MB);
@@ skipping 22 matching lines @@
       jsobj.AddProperty("streamId", "_Graph");
     }
 
     const String& message = String::Handle(String::New(js.ToCString()));
 
     uint8_t* chunk_start = buffer + (i * kChunkSize);
     intptr_t chunk_size = (i + 1 == num_chunks)
         ? stream.bytes_written() - (i * kChunkSize)
         : kChunkSize;
 
-    SendEvent(message, chunk_start, chunk_size);
+    SendEventWithData("_Graph", "_Graph", message, chunk_start, chunk_size);
   }
 }
 
 
 void Service::SendInspectEvent(Isolate* isolate, const Object& inspectee) {
+  if (!Service::NeedsDebugEvents()) {
+    return;
+  }
   ServiceEvent event(isolate, ServiceEvent::kInspect);
   event.set_inspectee(&inspectee);
   Service::HandleEvent(&event);
 }
 
 
 class ContainsAddressVisitor : public FindObjectVisitor {
  public:
   ContainsAddressVisitor(Isolate* isolate, uword addr)
       : FindObjectVisitor(isolate), addr_(addr) { }
@@ skipping 304 matching lines @@
 
 static const MethodParameter* set_name_params[] = {
   ISOLATE_PARAMETER,
   new MethodParameter("name", true),
   NULL,
 };
 
 
 static bool SetName(Isolate* isolate, JSONStream* js) {
   isolate->set_debugger_name(js->LookupParam("name"));
-  {
+  if (Service::NeedsIsolateEvents()) {
     ServiceEvent event(isolate, ServiceEvent::kIsolateUpdate);
     Service::HandleEvent(&event);
   }
   PrintSuccess(js);
   return true;
 }
 
 
 static ServiceMethodDescriptor service_methods_[] = {
   { "_dumpIdZone", DumpIdZone, NULL },
@@ skipping 87 matching lines @@
     ServiceMethodDescriptor& method = service_methods_[i];
     if (strcmp(method_name, method.name) == 0) {
       return &method;
     }
   }
   return NULL;
 }
 
 
 }  // namespace dart