Very simple version of Isolates.

sdk/lib/isolate/isolate.dart

 // 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.
 
 /**
  * Concurrent programming using _isolates_:
  * independent workers that are similar to threads
  * but don't share memory,
  * communicating only via messages.
  *
  * See also:
  * [dart:isolate - Concurrency with Isolates](https://www.dartlang.org/docs/dart-up-and-running/contents/ch03.html#ch03-dartisolate---concurrency-with-isolates)
  * in the library tour.
  */
 library dart.isolate;
 
 import "dart:async";
 
-part "isolate_stream.dart";
-
+/**
+ * Thrown when an isolate cannot be created.
+ */
 class IsolateSpawnException implements Exception {
+  // TODO(floitsch): clean up spawn exception.
   const IsolateSpawnException(String this._s);
   String toString() => "IsolateSpawnException: '$_s'";
   final String _s;
 }
 
-/**
- * The initial ReceivePort available by default for this isolate.
- *
- * This ReceivePort is created automatically
- * and is commonly used to establish
- * the first communication between isolates.
- * (See [spawnFunction] and [spawnUri].)
- */
-ReceivePort get port => _Isolate.port;
+class Isolate {
+
+  final SendPort _controlPort;
+
+  Isolate._fromControlPort(this._controlPort);
+
+  /**
+   * Creates and spawns an isolate that shares the same code as the current
+   * isolate.
+   *
+   * The argument [entryPoint] specifies the entry point of the spawned
+   * isolate. It must be a static top-level function or a static method that
+   * takes no arguments. It is not allowed to pass a function closure.
+   *
+   * The entry-point function is invoked with the initial [message].
+   * Usually the initial [message] contains a [SendPort] so
+   * that the spawner and spawnee can communicate with each other.
+   *
+   * Returns a future that will complete with an [Isolate] instance. The
+   * isolate instance can be used to control the spawned isolate.
+   */
+  external static Future<Isolate> spawn(void entryPoint(message), var message);
+
+  /**
+   * Creates and spawns an isolate that runs the code from the library with
+   * the specified URI.
+   *
+   * The isolate starts executing the top-level `main` function of the library
+   * with the given URI.
+   *
+   * The target `main` may have one of the four following signatures:
+   *
+   * * `main({message})`
+   * * `main(args, {message})`
+   * * `main(args)`
+   * * `main()`
+   *
+   * When present, the named argument `message` is set to the initial [message].
+   * When present, the positional argument `args` is set to the empty list
+   * (`[]`).
+   *
+   * Returns a future that will complete with an [Isolate] instance. The
+   * isolate instance can be used to control the spawned isolate.
+   */
+  external static Future<Isolate> spawnUri(Uri uri, var message);
+}
 
 /**
- * Creates and spawns an isolate
- * that shares the same code as the current isolate,
- * but that starts from the specified function.
- *
- * The [topLevelFunction] argument must be
- * a static top-level function or a static method that takes no
- * arguments. It is illegal to pass a function closure.
- *
- * When any isolate starts (even the main script of the application), a default
- * [ReceivePort] is created for it. This port is available from the top-level
- * getter [port] defined in this library.
- *
- * This function returns a [SendPort] derived from
- * the child isolate's default port.
- *
- * The optional [unhandledExceptionCallback] argument is invoked whenever an
- * exception inside the isolate is unhandled. It can be seen as a big
- * `try/catch` around everything that is executed inside the isolate. The
- * callback should return `true` if it was able to handle the exception.
- */
-SendPort spawnFunction(void topLevelFunction(),
-    [bool unhandledExceptionCallback(IsolateUnhandledException e)])
-    => _Isolate.spawnFunction(topLevelFunction, unhandledExceptionCallback);
-
-/**
- * Creates and spawns an isolate that runs the code from the specified URI.
- *
- * As with [spawnFunction],
- * the child isolate has a default [ReceivePort],
- * and this function returns a [SendPort] derived from it.
- */
-SendPort spawnUri(String uri) => _Isolate.spawnUri(uri);
-
-/**
- * Together with [ReceivePort],
- * the only means of communication between isolates.
+ * Sends messages to its [ReceivePort]s.
  *
  * [SendPort]s are created from [ReceivePort]s. Any message sent through
  * a [SendPort] is delivered to its respective [ReceivePort]. There might be
  * many [SendPort]s for the same [ReceivePort].
  *
  * [SendPort]s can be transmitted to other isolates.
  */
 abstract class SendPort {
 
   /**
    * Sends an asynchronous [message] to this send port. The message is copied to
-   * the receiving isolate. If specified, the [replyTo] port will be provided to
-   * the receiver to facilitate exchanging sequences of messages.
+   * the receiving isolate.
    *
    * The content of [message] can be: primitive values (null, num, bool, double,
    * String), instances of [SendPort], and lists and maps whose elements are any
    * of these. List and maps are also allowed to be cyclic.
    *
    * In the special circumstances when two isolates share the same code and are
    * running in the same process (e.g. isolates created via [spawnFunction]), it
    * is also possible to send object instances (which would be copied in the
    * process). This is currently only supported by the dartvm.  For now, the
    * dart2js compiler only supports the restricted messages described above.
    *
-   * Deprecation note: it is no longer valid to transmit a [ReceivePort] in a
-   * message. Previously they were translated to the corresponding send port
-   * before being transmitted.
+   * The second argument [replyTo] is deprecated.
    */
   void send(var message, [SendPort replyTo]);
 
   /**
-   * Sends a message to this send port and returns a [Future] of the reply.
-   * Basically, this internally creates a new receive port, sends a
-   * message to this send port with replyTo set to such receive port, and, when
-   * a reply is received, it closes the receive port and completes the returned
-   * future.
-   */
-  Future call(var message);
-
-  /**
    * Tests whether [other] is a [SendPort] pointing to the same
    * [ReceivePort] as this one.
    */
   bool operator==(var other);
 
   /**
    * Returns an immutable hash code for this send port that is
    * consistent with the == operator.
    */
   int get hashCode;
-
-}
-
-/**
- * Together with [SendPort], the only means of
- * communication between isolates.
- *
- * [ReceivePort]s have a [:toSendPort:] method
- * which returns a [SendPort]. Any message that is sent through this [SendPort]
- * is delivered to the [ReceivePort] it has been created from. There, they are
- * dispatched to the callback that has been registered on the receive port.
- *
- * A [ReceivePort] may have many [SendPort]s.
- */
-abstract class ReceivePort {
-
-  /**
-   * Opens a long-lived port for receiving messages. The returned port
-   * must be explicitly closed through [ReceivePort.close].
-   */
-  external factory ReceivePort();
-
-  /**
-   * Sets up a callback function for receiving pending or future
-   * messages on this receive port.
-   */
-  void receive(void callback(var message, SendPort replyTo));
-
-  /**
-   * Closes this receive port immediately. Pending messages will not
-   * be processed and it is impossible to re-open the port. Single-shot
-   * reply ports, such as those created through [SendPort.call], are
-   * automatically closed when the reply has been received. Multiple
-   * invocations of [close] are allowed but ignored.
-   */
-  void close();
-
-  /**
-   * Creates a new send port that sends to this receive port. It is legal to
-   * create several [SendPort]s from the same [ReceivePort].
-   */
-  SendPort toSendPort();
-
-}
-
-/**
- * [SendPortSync]s are created from [ReceivePortSync]s. Any message sent through
- * a [SendPortSync] is delivered to its respective [ReceivePortSync]. There
- * might be many [SendPortSync]s for the same [ReceivePortSync].
- *
- * [SendPortSync]s can be transmitted to other isolates.
- */
-abstract class SendPortSync {
-  /**
-   * Sends a synchronous message to this send port and returns the result.
-   */
-  callSync(var message);
-
-  /**
-   * Tests whether [other] is a [SendPortSync] pointing to the same
-   * [ReceivePortSync] as this one.
-   */
-  bool operator==(var other);
-
-  /**
-   * Returns an immutable hash code for this send port that is
-   * consistent with the == operator.
-   */
-  int get hashCode;
 }
 
+/**
+ * Together with [SendPort], the only means of communication between isolates.
+ *
+ * [ReceivePort]s have a `sendport` getter which returns a [SendPort].
+ * Any message that is sent through this [SendPort]
+ * is delivered to the [ReceivePort] it has been created from. There, the
+ * message is dispatched to its listener.
+ *
+ * A [ReceivePort] is a non-broadcast stream. This means that it buffers
+ * incoming messages until a listener is registered. Only one listener can
+ * receive messages. See [Stream.asBroadcastStream] for transforming the port
+ * to a broadcast stream.
+ *
+ * A [ReceivePort] may have many [SendPort]s.
+ */
+abstract class ReceivePort implements Stream {
+
+  /**
+   * Opens a long-lived port for receiving messages.
+   *
+   * A [ReceivePort] is a non-broadcast stream. This means that it buffers
+   * incoming messages until a listener is registered. Only one listener can
+   * receive messages. See [Stream.asBroadcastStream] for transforming the port
+   * to a broadcast stream.
+   *
+   * A receive port is closed by canceling its subscription.
+   */
+  external factory ReceivePort();
+
+  /**
+   * Creates a [ReceivePort] from a [RawReceivePort].
+   *
+   * The handler of the given [rawPort] is overwritten during the construction
+   * of the result.
+   */
+  external factory ReceivePort.fromRawReceivePort(RawReceivePort rawPort);
+
+  /**
+   * Inherited from [Stream].
+   *
+   * Note that all named arguments are ignored since a ReceivePort will never
+   * receive an error, or done message.
+   */
+  StreamSubscription listen(void onData(var message),
+                            { Function onError,
+                              void onDone(),
+                              bool cancelOnError });
+
+  /**
+   * Closes `this`.
+   *
+   * If the stream has not been canceled yet, adds a close-event to the event
+   * queue and discards any further incoming messages.
+   *
+   * If the stream has already been canceled this method has no effect.
+   */
+  void close();
+
+  /**
+   * Returns a send port that sends to this receive port.
+   */
+  SendPort get sendPort;
+}
+
+abstract class RawReceivePort {
+  /**
+   * Opens a long-lived port for receiving messages.
+   *
+   * A [RawReceivePort] is low level and does not work with [Zone]s. It
+   * can not be paused. The data-handler must be set before the first
+   * event is received.
+   */
+  external factory RawReceivePort([void handler(event)]);
+
+  /**
+   * Sets the handler that is invoked for every incoming message.
+   *
+   * The handler is invoked in the root-zone ([Zone.ROOT]).
+   */
+  void set handler(Function newHandler);
+
+  /**
+   * Closes the port.
+   *
+   * After a call to this method any incoming message is silently dropped.
+   */
+  void close();
+}
+
 // The VM doesn't support accessing external globals in the same library. We
 // therefore create this wrapper class.
 // TODO(6997): Don't go through static class for external variables.
 abstract class _Isolate {
   external static ReceivePort get port;
-  external static SendPort spawnFunction(void topLevelFunction(),
-    [bool unhandledExceptionCallback(IsolateUnhandledException e)]);
   external static SendPort spawnUri(String uri);
 }
 
 /**
  * Wraps unhandled exceptions thrown during isolate execution. It is
  * used to show both the error message and the stack trace for unhandled
  * exceptions.
  */
+// TODO(floitsch): probably going to remove and replace with something else.
 class IsolateUnhandledException implements Exception {
   /** Message being handled when exception occurred. */
   final message;
 
   /** Wrapped exception. */
   final source;
 
   /** Trace for the wrapped exception. */
   final Object stackTrace;
 
   const IsolateUnhandledException(this.message, this.source, this.stackTrace);
 
   String toString() {
     return 'IsolateUnhandledException: exception while handling message: '
         '${message} \n  '
         '${source.toString().replaceAll("\n", "\n  ")}\n'
         'original stack trace:\n  '
         '${stackTrace.toString().replaceAll("\n","\n  ")}';
   }
 }