OOB messages and general message refactor.

runtime/vm/isolate.cc

Index: runtime/vm/isolate.cc
===================================================================
--- runtime/vm/isolate.cc	(revision 3210)
+++ runtime/vm/isolate.cc	(working copy)
@@ -36,8 +36,7 @@
 Isolate::Isolate()
     : store_buffer_(),
       message_queue_(NULL),
-      post_message_callback_(NULL),
-      close_port_callback_(NULL),
+      message_notify_callback_(NULL),
       num_ports_(0),
       live_ports_(0),
       main_port_(0),
@@ -83,31 +82,31 @@
 }
 
 
-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) {
+  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() {
   Isolate* result = new Isolate();
   ASSERT(result != NULL);
@@ -120,8 +119,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();
@@ -304,25 +301,28 @@
 
 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) {
+        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());
     }
   }