OOB messages and general message refactor.

runtime/vm/dart_api_impl.cc

Index: runtime/vm/dart_api_impl.cc
===================================================================
--- runtime/vm/dart_api_impl.cc	(revision 3210)
+++ runtime/vm/dart_api_impl.cc	(working copy)
@@ -15,6 +15,7 @@
 #include "vm/exceptions.h"
 #include "vm/growable_array.h"
 #include "vm/longjump.h"
+#include "vm/message_queue.h"
 #include "vm/native_entry.h"
 #include "vm/object.h"
 #include "vm/object_store.h"
@@ -195,11 +196,6 @@
 }
 
 
-Dart_Message Api::CastMessage(uint8_t* message) {
-  return reinterpret_cast<Dart_Message>(message);
-}
-
-
 Dart_Handle Api::Success() {
   Isolate* isolate = Isolate::Current();
   ASSERT(isolate != NULL);
@@ -667,15 +663,11 @@
 // --- Messages and Ports ---
 
 
-DART_EXPORT void Dart_SetMessageCallbacks(
-    Dart_PostMessageCallback post_message_callback,
-    Dart_ClosePortCallback close_port_callback) {
+DART_EXPORT void Dart_SetMessageNotifyCallback(
+    Dart_MessageNotifyCallback message_notify_callback) {
   Isolate* isolate = Isolate::Current();
   CHECK_ISOLATE(isolate);
-  ASSERT(post_message_callback != NULL);
-  ASSERT(close_port_callback != NULL);
-  isolate->set_post_message_callback(post_message_callback);
-  isolate->set_close_port_callback(close_port_callback);
+  isolate->set_message_notify_callback(message_notify_callback);
 }
 
 
@@ -716,20 +708,32 @@
 }
 
 
-DART_EXPORT Dart_Handle Dart_HandleMessage(Dart_Port dest_port_id,
-                                           Dart_Port reply_port_id,
-                                           Dart_Message dart_message) {
-  DARTSCOPE(Isolate::Current());
-  const Instance& msg = Instance::Handle(DeserializeMessage(dart_message));
-  // TODO(turnidge): Should this call be wrapped in a longjmp?
-  const Object& result =
-      Object::Handle(DartLibraryCalls::HandleMessage(dest_port_id,
-                                                     reply_port_id,
-                                                     msg));
-  if (result.IsError()) {
-    return Api::NewLocalHandle(result);
-  }
-  ASSERT(result.IsNull());
+DART_EXPORT Dart_Handle Dart_HandleMessage() {
+  Isolate* isolate = Isolate::Current();
+  // Process all OOB messages and at most one normal message.
+  Message* message = NULL;
+  Message::Priority priority = Message::kNormalPriority;
+  do {
+    DARTSCOPE(isolate);
+    message = isolate->message_queue()->DequeueNoWait();
+    if (message == NULL) {
+      break;
+    }
+    priority = message->priority();
+    if (priority == Message::kOOBPriority) {
+      UNIMPLEMENTED();

[Ivan Posva, 2012/01/23 22:00:43]

Can you briefly describe in a comment what should happen in this case?

[turnidge, 2012/01/23 22:32:52]

Done.

+    }
+    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.IsError()) {
+      return Api::NewLocalHandle(result);
+    }
+    ASSERT(result.IsNull());
+  } while (priority == Message::kOOBPriority);

[Ivan Posva, 2012/01/23 22:00:43]

How about:
while (priority >= Message::kOOBPriority)

This would break if we ever start using a priority higher than out-of-band.

[turnidge, 2012/01/23 22:32:52]

Anton didn't like the >= because it is unclear which way priorities are ordered,
so I decided to switch this.  But I am now feeling like switching it back.

Done.

On 2012/01/23 22:00:43, Ivan Posva wrote:

   return Api::Success();
 }
 
@@ -756,7 +760,8 @@
   writer.WriteMessage(len, data);
 
   // Post the message at the given port.
-  return PortMap::PostMessage(port_id, kNoReplyPort, Api::CastMessage(buffer));
+  return PortMap::PostMessage(new Message(
+      port_id, Message::kIllegalPort, buffer, Message::kNormalPriority));
 }
 
 
@@ -769,7 +774,8 @@
   SnapshotWriter writer(Snapshot::kMessage, &data, &allocator);
   writer.WriteObject(object.raw());
   writer.FinalizeBuffer();
-  return PortMap::PostMessage(port_id, kNoReplyPort, Api::CastMessage(data));
+  return PortMap::PostMessage(new Message(
+      port_id, Message::kIllegalPort, data, Message::kNormalPriority));
 }