OOB messages and general message refactor.

runtime/vm/isolate.cc

Index: runtime/vm/isolate.cc
===================================================================
--- runtime/vm/isolate.cc	(revision 3557)
+++ runtime/vm/isolate.cc	(working copy)
@@ -37,8 +37,7 @@
 Isolate::Isolate()
     : store_buffer_(),
       message_queue_(NULL),
-      post_message_callback_(NULL),
-      close_port_callback_(NULL),
+      message_notify_callback_(NULL),
       name_(NULL),
       num_ports_(0),
       live_ports_(0),
@@ -87,31 +86,43 @@
 }
 
 
-static bool StandardPostMessageCallback(Dart_Isolate dart_isolate,
-                                        Dart_Port dest_port,
-                                        Dart_Port reply_port,
-                                        Dart_Message dart_message) {
-  Isolate* isolate = reinterpret_cast<Isolate*>(dart_isolate);
-  ASSERT(isolate != NULL);
-  PortMessage* message = new PortMessage(dest_port, reply_port, dart_message);
-  isolate->message_queue()->Enqueue(message);
-  return true;
+void Isolate::PostMessage(Message* message) {
+  if (FLAG_trace_isolates) {
+    const char* source_name = "<native code>";
+    Isolate* source_isolate = Isolate::Current();
+    if (source_isolate) {
+      source_name = source_isolate->name();
+    }
+    OS::Print("[>] Posting message:\n"
+              "\tsource:     %s\n"
+              "\treply_port: %lld\n"
+              "\tdest:       %s\n"
+              "\tdest_port:  %lld\n",
+              source_name, message->reply_port(), name(), message->dest_port());
+  }
+  message_queue()->Enqueue(message);
+  if (message->priority() >= Message::kOOBPriority) {
+    // Handle out of band messages even if the isolate is busy.
+    ScheduleInterrupts(Isolate::kMessageInterrupt);
+  }
+  Dart_MessageNotifyCallback callback = message_notify_callback();
+  if (callback) {
+    // Allow the embedder to handle message notification.
+    (*callback)(Api::CastIsolate(this));
+  }
 }
 
 
-static void StandardClosePortCallback(Dart_Isolate dart_isolate,
-                                      Dart_Port port) {
-  // Remove the pending messages for this port.
-  Isolate* isolate = reinterpret_cast<Isolate*>(dart_isolate);
-  ASSERT(isolate != NULL);
-  if (port == kCloseAllPorts) {
-    isolate->message_queue()->FlushAll();
-  } else {
-    isolate->message_queue()->Flush(port);
-  }
+void Isolate::ClosePort(Dart_Port port) {
+  message_queue()->Flush(port);
 }
 
 
+void Isolate::CloseAllPorts() {
+  message_queue()->FlushAll();
+}
+
+
 Isolate* Isolate::Init(const char* name_prefix) {
   Isolate* result = new Isolate();
   ASSERT(result != NULL);
@@ -124,8 +135,6 @@
   MessageQueue* queue = new MessageQueue();
   ASSERT(queue != NULL);
   result->set_message_queue(queue);
-  result->set_post_message_callback(&StandardPostMessageCallback);
-  result->set_close_port_callback(&StandardClosePortCallback);
 
   // Setup the Dart API state.
   ApiState* state = new ApiState();
@@ -338,25 +347,31 @@
 
 RawObject* Isolate::StandardRunLoop() {
   ASSERT(long_jump_base() != NULL);
-  ASSERT(post_message_callback() == &StandardPostMessageCallback);
-  ASSERT(close_port_callback() == &StandardClosePortCallback);
+  ASSERT(message_notify_callback() == NULL);
 
   while (live_ports() > 0) {
     ASSERT(this == Isolate::Current());
     Zone zone(this);
     HandleScope handle_scope(this);
 
-    PortMessage* message = message_queue()->Dequeue(0);
+    Message* message = message_queue()->Dequeue(0);
     if (message != NULL) {
+      if (message->priority() >= Message::kOOBPriority) {
+        // TODO(turnidge): Out of band messages will not go through the
+        // regular message handler.  Instead they will be dispatched to
+        // special vm code.  Implement.
+        UNIMPLEMENTED();
+      }
       const Instance& msg =
           Instance::Handle(DeserializeMessage(message->data()));
       const Object& result = Object::Handle(
           DartLibraryCalls::HandleMessage(
               message->dest_port(), message->reply_port(), msg));
       delete message;
-      if (result.IsUnhandledException()) {
+      if (result.IsError()) {
         return result.raw();
       }
+      ASSERT(result.IsNull());
     }
   }