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);
}