Add IsolateRunner as a helper around Isolate.

lib/ports.dart

+// Copyright (c) 2015, 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.
+
+/**
+ * Utility functions for setting up ports and sending data.
+ *
+ * This library contains a number of functions that handle the
+ * boiler-plate of setting up a receive port and receiving a
+ * single message on the port.
+ *
+ * There are different functions that offer different ways to
+ * handle the incoming message.
+ *
+ * The simplest function, [singleCallbackPort], takes a callback
+ * and returns a port, and then calls the callback for the first
+ * message sent on the port.
+ *
+ * Other functions intercept the returned value and either
+ * does something with it, or puts it into a [Future] or [Completer].
+ */
+library dart.pkg.isolate.ports;
+
+import "dart:isolate";
+import "dart:async";
+import "src/lists.dart";
+
+/**
+ * Create a [SendPort] that accepts only one message.
+ *
+ * The [callback] function is called once, with the first message
+ * received by the receive port.
+ * All futher messages are ignored.
+ *
+ * If [timeout] is supplied, it is used as a limit on how
+ * long it can take before the message is received. If a
+ * message isn't received in time, the `callback` function
+ * is called once with the [timeoutValue] instead.
+ *
+ * Returns the `SendPort` expecting the single message.
+ *
+ * Equivalent to:
+ *
+ *     (new ReceivePort()
+ *         ..first.timeout(duration, () => timeoutValue).then(callback))
+ *         .sendPort
+ */
+SendPort singleCallbackPort(void callback(response),
+                            {Duration timeout,
+                             var timeoutValue}) {
+  RawReceivePort responsePort = new RawReceivePort();
+  Zone zone = Zone.current;
+  callback = zone.registerUnaryCallback(callback);
+  var timer;
+  responsePort.handler = (response) {
+    responsePort.close();
+    if (timer != null) timer.cancel();
+    zone.runUnary(callback, response);
+  };
+  if (timeout != null) {
+    timer = new Timer(timeout, () {
+      responsePort.close();
+      callback(timeoutValue);
+    });
+  }
+  return responsePort.sendPort;
+}
+
+/**
+ * Create a [SendPort] that accepts only one message.
+ *
+ * When the first message is received, the [callback] function is
+ * called with the message as argument,
+ * and the [completer] is completed with the result of that call.
+ * All further messages are ignored.
+ *
+ * If `callback` is omitted, it defaults to an identity function.
+ * The `callback` call may return a future, and the completer will
+ * wait for that future to complete.
+ *
+ * If [timeout] is supplied, it is used as a limit on how
+ * long it can take before the message is received. If a
+ * message isn't received in time, the [onTimeout] is called,
+ * and `completer` is completed with the result of that call
+ * instead.
+ * The [callback] function will not be interrupted by the time-out,
+ * as long as the initial message is received in time.
+ * If `onTimeout` is omitted, it defaults to completing the `completer` with
+ * a [TimeoutException].
+ *
+ * The [completer] may be a synchronous completer. It is only
+ * completed in response to another event, either a port message or a timer.
+ *
+ * Returns the `SendPort` expecting the single message.
+ */
+SendPort singleCompletePort(Completer completer,
+                            {callback(message),
+                             Duration timeout,
+                             onTimeout()}) {
+  if (callback == null && timeout == null) {
+    return singleCallbackPort(completer.complete);
+  }
+  RawReceivePort responsePort = new RawReceivePort();
+  var timer;
+  if (callback == null) {
+    responsePort.handler = (response) {
+      responsePort.close();
+      if (timer != null) timer.cancel();
+      completer.complete(response);
+    };
+  } else {
+    Zone zone = Zone.current;
+    Function action = zone.registerUnaryCallback((response) {
+      completer.complete(new Future.sync(() => callback(response)));
+    });
+    responsePort.handler = (response) {
+      responsePort.close();
+      if (timer != null) timer.cancel();
+      zone.runUnary(action, response);
+    };
+  }
+  if (timeout != null) {
+    timer = new Timer(timeout, () {
+      responsePort.close();
+      if (onTimeout != null) {
+        completer.complete(new Future.sync(onTimeout));
+      } else {
+        completer.completeError(new TimeoutException("Future not completed",
+                                                     timeout));
+      }
+    });
+  }
+  return responsePort.sendPort;
+}
+
+/**
+ * Creates a [Future], and a [SendPort] that can be used to complete that
+ * future.
+ *
+ * Calls [action] with the response `SendPort`, then waits for someone
+ * to send a value on that port
+ * The returned `Future` is completed with the value sent on the port.
+ *
+ * If [action] throws, which it shouldn't,
+ * the returned future is completed with that error.
+ * Any return value of `action` is ignored, and if it is asynchronous,
+ * it should handle its own errors.
+ *
+ * If [timeout] is supplied, it is used as a limit on how
+ * long it can take before the message is received. If a
+ * message isn't received in time, the [timeoutValue] used
+ * as the returned future's value instead.
+ *
+ * If you want a timeout on the returned future, it's recommended to
+ * use the [timeout] parameter, and not [Future.timeout] on the result.
+ * The `Future` method won't be able to close the underlying [ReceivePort].
+ */
+Future singleResponseFuture(void action(SendPort responsePort),
+                            {Duration timeout,
+                             var timeoutValue}) {
+  Completer completer = new Completer.sync();
+  RawReceivePort responsePort = new RawReceivePort();
+  Timer timer;
+  Zone zone = Zone.current;
+  responsePort.handler = (v) {
+    responsePort.close();
+    if (timer != null) timer.cancel();
+    zone.run(() {
+      completer.complete(v);
+    });
+  };
+  if (timeout != null) {
+    timer = new Timer(timeout, () {
+      responsePort.close();
+      completer.complete(timeoutValue);
+    });
+  }
+  try {
+    action(responsePort.sendPort);
+  } catch (e, s) {
+    responsePort.close();
+    if (timer != null) timer.cancel();
+    // Delay completion because completer is sync.
+    scheduleMicrotask(() { completer.completeError(e, s); });
+  }
+  return completer.future;
+}
+
+
+/**
+ * Send the result of a future, either value or error, as a message.
+ *
+ * The result of [future] is sent on [resultPort] in a form expected by
+ * either [receiveFutureResult], [completeFutureResult], or
+ * by the port of [singleResultFuture].
+ */
+void sendFutureResult(Future future, SendPort resultPort) {
+  future.then((v) { resultPort.send(list1(v)); },
+              onError: (e, s) { resultPort.send(list2("$e", "$s")); });
+}
+
+
+/**
+ * Creates a [Future], and a [SendPort] that can be used to complete that
+ * future.
+ *
+ * Calls [action] with the response `SendPort`, then waits for someone
+ * to send a future result on that port using [sendFutureResult].
+ * The returned `Future` is completed with the future result sent on the port.
+ *
+ * If [action] throws, which it shouldn't,
+ * the returned future is completed with that error,
+ * unless someone manages to send a message on the port before `action` throws.
+ *
+ * If [timeout] is supplied, it is used as a limit on how
+ * long it can take before the message is received. If a
+ * message isn't received in time, the [onTimeout] is called,
+ * and the future is completed with the result of that call
+ * instead.
+ * If `onTimeout` is omitted, it defaults to throwing
+ * a [TimeoutException].
+ */
+Future singleResultFuture(void action(SendPort responsePort),
+                          {Duration timeout,
+                           onTimeout()}) {
+  Completer completer = new Completer.sync();
+  SendPort port = singleCompletePort(completer,
+                                     callback: receiveFutureResult,
+                                     timeout: timeout,
+                                     onTimeout: onTimeout);
+  try {
+    action(port);
+  } catch (e, s) {
+    // This should not happen.
+    sendFutureResult(new Future.error(e, s), port);
+  }
+  return completer.future;
+}
+
+/**
+ * Completes a completer with a message created by [sendFutureResult]
+ *
+ * The [response] must be a message on the format sent by [sendFutureResult].
+ */
+void completeFutureResult(var response, Completer completer) {
+  if (response.length == 2) {
+    var error = new RemoteError(response[0], response[1]);
+    completer.completeError(error, error.stackTrace);
+  } else {
+    var result = response[0];
+    completer.complete(result);
+  }
+}
+
+
+/**
+ * Convertes a received message created by [sendFutureResult] to a future
+ * result.
+ *
+ * The [response] must be a message on the format sent by [sendFutureResult].
+ */
+Future receiveFutureResult(var response) {
+  if (response.length == 2) {
+    var error = new RemoteError(response[0], response[1]);
+    return new Future.error(error, error.stackTrace);
+  }
+  var result = response[0];
+  return new Future.value(result);
+}