isolates in frog: playing with API improvements

frog/lib/isolate_serialization.dart

-// Copyright (c) 2011, the Dart project authors.  Please see the AUTHORS file
+// Copyright (c) 2012, 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.
 
 /**
  * Abstract visitor for dart objects that can be passed as messages between any
  * isolates.
  */
 class MessageTraverser {
   static bool isPrimitive(x) {
     return (x === null) || (x is String) || (x is num) || (x is bool);
@@ skipping 33 matching lines @@
   _getInfo(var o) {
     return _getAttachedInfo(o);
   }
 
   _dispatch(var x) {
     if (isPrimitive(x)) return visitPrimitive(x);
     if (x is List) return visitList(x);
     if (x is Map) return visitMap(x);
     if (x is NativeJsSendPort) return visitNativeJsSendPort(x);
     if (x is WorkerSendPort) return visitWorkerSendPort(x);
+    if (x is BufferingSendPort) return visitBufferingSendPort(x);
     if (x is ReceivePortImpl) return visitReceivePort(x);
     if (x is ReceivePortSingleShotImpl) return visitReceivePortSingleShot(x);
     // TODO(floitsch): make this a real exception. (which one)?
     throw "Message serialization: Illegal value $x passed";
   }
 
   abstract visitPrimitive(x);
   abstract visitList(List x);
   abstract visitMap(Map x);
   abstract visitNativeJsSendPort(NativeJsSendPort x);
   abstract visitWorkerSendPort(WorkerSendPort x);
+  abstract visitBufferingSendPort(BufferingSendPort x);
   abstract visitReceivePort(ReceivePortImpl x);
   abstract visitReceivePortSingleShot(ReceivePortSingleShotImpl x);
 
   List _taggedObjects;
 
   _clearAttachedInfo(var o) native
       "o['__MessageTraverser__attached_info__'] = (void 0);";
 
   _setAttachedInfo(var o, var info) native
       "o['__MessageTraverser__attached_info__'] = info;";
 
   _getAttachedInfo(var o) native
       "return o['__MessageTraverser__attached_info__'];";
+
+  _visitNativeOrWorkerPort(SendPort p) {
+    if (p is NativeJsSendPort) return visitNativeJsSendPort(p);
+    if (p is WorkerSendPort) return visitWorkerSendPort(p);
+    throw "Illegal underlying port $p";
+  }
 }
 
 /** A visitor that recursively copies a message. */
 class Copier extends MessageTraverser {
   Copier() : super();
 
   visitPrimitive(x) => x;
 
   List visitList(List list) {
     List copy = _getInfo(list);
@@ skipping 25 matching lines @@
 
   SendPort visitNativeJsSendPort(NativeJsSendPort port) {
     return new NativeJsSendPort(port._receivePort, port._isolateId);
   }
 
   SendPort visitWorkerSendPort(WorkerSendPort port) {
     return new WorkerSendPort(
         port._workerId, port._isolateId, port._receivePortId);
   }
 
+  SendPort visitBufferingSendPort(BufferingSendPort port) {
+    if (port._port != null) {
+      return _visitNativeOrWorkerPort(port._port);
+    } else {
+      // TODO(floitsch): Use real exception (which one?).
+      throw "internal error: must call _waitForPendingPorts to ensure all"
+          + " ports are resolved at this point.";
+    }
+  }
+
   SendPort visitReceivePort(ReceivePortImpl port) {
     return port.toSendPort();
   }
 
   SendPort visitReceivePortSingleShot(ReceivePortSingleShotImpl port) {
     return port.toSendPort();
   }
 }
 
 /** Visitor that serializes a message as a JSON array. */
@@ skipping 27 matching lines @@
 
   visitNativeJsSendPort(NativeJsSendPort port) {
     return ['sendport', _globalState.currentWorkerId,
         port._isolateId, port._receivePort._id];
   }
 
   visitWorkerSendPort(WorkerSendPort port) {
     return ['sendport', port._workerId, port._isolateId, port._receivePortId];
   }
 
+  SendPort visitBufferingSendPort(BufferingSendPort port) {
+    if (port._port != null) {
+      return _visitNativeOrWorkerPort(port._port);
+    } else {
+      // TODO(floitsch): Use real exception (which one?).
+      throw "internal error: must call _waitForPendingPorts to ensure all"
+          + " ports are resolved at this point.";
+    }
+  }
+
   visitReceivePort(ReceivePortImpl port) {
     return visitNativeJsSendPort(port.toSendPort());;
   }
 
   visitReceivePortSingleShot(ReceivePortSingleShotImpl port) {
     return visitNativeJsSendPort(port.toSendPort());
   }
 
   _serializeList(List list) {
     int len = list.length;
     var result = new List(len);
     for (int i = 0; i < len; i++) {
       result[i] = _dispatch(list[i]);
     }
     return result;
   }
 
   int _nextFreeRefId = 0;
 }
 
+/** Visitor that finds all unresolved [SendPort]s in a message. */
+class PendingSendPortFinder extends MessageTraverser {
+  List<Future<SendPort>> ports;
+  PendingSendPortFinder() : super(), ports = [];
+
+  visitPrimitive(x) {}
+  visitNativeJsSendPort(NativeJsSendPort port) {}
+  visitWorkerSendPort(WorkerSendPort port) {}
+  visitReceivePort(ReceivePortImpl port) {}
+  visitReceivePortSingleShot(ReceivePortSingleShotImpl port) {}
+
+  visitList(List list) {
+    final visited = _getInfo(list);
+    if (visited !== null) return;
+    _attachInfo(list, true);
+    list.forEach(_dispatch);
+  }
+
+  visitMap(Map map) {
+    final visited = _getInfo(map);
+    if (visited !== null) return;
+
+    _attachInfo(map, true);
+    map.getValues().forEach(_dispatch);
+  }
+
+  visitBufferingSendPort(BufferingSendPort port) {
+    if (port._port == null) {
+      ports.add(port._futurePort);
+    }
+  }
+}
+
+
 /** Deserializes arrays created with [Serializer]. */
 class Deserializer {
   Deserializer();
 
   static bool isPrimitive(x) {
     return (x === null) || (x is String) || (x is num) || (x is bool);
   }
 
   deserialize(x) {
     if (isPrimitive(x)) return x;
@@ skipping 61 matching lines @@
       if (isolate == null) return null; // Isolate has been closed.
       var receivePort = isolate.lookup(receivePortId);
       return new NativeJsSendPort(receivePort, isolateId);
     } else {
       return new WorkerSendPort(workerId, isolateId, receivePortId);
     }
   }
 
   Map<int, Dynamic> _deserialized;
 }