Stream isolates.

sdk/lib/isolate/isolate_stream.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.
+
+/**
+ * The initial [IsolateStream] available by default for this isolate. This
+ * [IsolateStream] is created automatically and it is commonly used to establish
+ * the first communication between isolates (see [streamSpawnFunction] and
+ * [streamSpawnUri]).
+ */
+final IsolateStream stream = new IsolateStream._fromOriginalReceivePort(port);

[Lasse Reichstein Nielsen, 2012/11/22 10:23:15]

Is 'port' defined in base.dart?

[floitsch, 2012/11/22 11:58:04]

yes.

+
+/**
+ * A [MessageBox] creates an [IsolateStream] [stream] and an [IsolateSink]

[Lasse Reichstein Nielsen, 2012/11/22 10:23:15]

Commas around [stream] and [sink].

[floitsch, 2012/11/22 11:58:04]

Done.

+ * [sink].
+ *
+ * Any message that is written into the [sink] (independent of the isolate) is
+ * sent to the [stream] where its subscribers can react to the messages.
+ */
+class MessageBox {
+  final IsolateStream stream;
+  final IsolateSink sink;
+
+  MessageBox.oneShot() : this._oneShot(new ReceivePort());
+  MessageBox._oneShot(ReceivePort receivePort)
+      : stream = new IsolateStream._fromOriginalReceivePortOneShot(receivePort),
+        sink = new IsolateSink._fromPort(receivePort.toSendPort());
+
+  MessageBox() : this._(new ReceivePort());
+  MessageBox._(ReceivePort receivePort)
+      : stream = new IsolateStream._fromOriginalReceivePort(receivePort),
+        sink = new IsolateSink._fromPort(receivePort.toSendPort());
+}
+
+// Used for mangling.
+const int _ISOLATE_STREAM_TOKEN = 132421112;

[Lasse Reichstein Nielsen, 2012/11/22 10:23:15]

Pah, doesn't look random enough :P

[floitsch, 2012/11/22 11:58:04]

Changed to 132421119. ;)

+
+/**
+ * [IsolateStream]s, together with [IsolateSink]s, are the only means of
+ * communication between isolates. Each IsolateStream has a corresponding
+ * [IsolateSink]. Any message written into that sink will be delivered to
+ * the stream and then dispatched to the stream's subscribers.
+ */
+class IsolateStream extends Stream<dynamic> {
+  final ReceivePort _port;
+  StreamController _controller = new StreamController();

[Lasse Reichstein Nielsen, 2012/11/22 10:23:15]

The base StreamController doesn't handle 'pause' at all (i.e., it receives
requests to pause, but does nothing).
I assume this is sufficient here.

[floitsch, 2012/11/22 11:58:04]

It is (I think), but iirc when I started this CL the pausing Streams weren't in
yet.

+
+  IsolateStream._fromOriginalReceivePort(this._port) {
+    _port.receive((message, replyTo) {
+      assert(replyTo == null);
+      _write(message);
+    });
+  }
+
+  IsolateStream._fromOriginalReceivePortOneShot(this._port) {
+    _port.receive((message, replyTo) {
+      assert(replyTo == null);
+      _write(message);
+      close();
+    });
+  }
+
+  void _write(var message) {
+    message = _unmangleMessage(message);
+    _controller.sink.write(message);
+  }
+
+  void close() {
+    _controller.close()
+    _port.close();
+  }
+
+  StreamSubscription<T> subscribe({void onData(T event),
+                                   void onError(StreamError error),
+                                   void onDone(),
+                                   bool unsubscribeOnError}) {
+      return _controller.subscribe(onData: onData,
+                                   onError: onError,
+                                   onDone: onDone,
+                                   unsubscribeOnError: unsubscribeOnError);
+  }
+
+  dynamic _unmangleMessage(var message) {
+    _IsolateDecoder decoder = new _IsolateDecoder(
+        _ISOLATE_STREAM_TOKEN,
+        (data) {
+          if (data is! List) return data;
+          if (data.length != 2) return data;
+          if (data[0] != "Sink" || data[1] is! SendPort) return data;
+          return new IsolateSink._fromPort(data[1]);
+        });
+    return decoder.decode(message);
+  }
+}
+
+/**
+ * [IsolateSink]s represent the feed for [IsolateStream]s. Any message written
+ * to [this] is delivered to its respective [IsolateStream]. [IsolateSink]s are
+ * created by [MessageBox]es.
+ *
+ * [IsolateSink]s can be transmitted to other isolates.
+ */
+class IsolateSink extends StreamSink<dynamic> {
+  final SendPort _port;
+  IsolateSink._fromPort(this._port);
+
+  /**
+   * Sends an asynchronous [message] to the linked [IsolateStream]. The message
+   * is copied to the receiving isolate.
+   *
+   * The content of [message] can be: primitive values (null, num, bool, double,
+   * String), instances of [IsolateSink]s, 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.
+   */
+  void write(dynamic message) {
+    _port.send(_mangleMessage(message));
+  }
+
+  void signalError(StreamError errorEvent) {
+    throw new UnimplementedError("signalError on isolate streams");
+  }
+
+  dynamic _mangleMessage(var message) {
+    _IsolateEncoder encoder = new _IsolateEncoder(
+        _ISOLATE_STREAM_TOKEN,
+        (data) => (data is IsolateSink) ? ["Sink", data._port] : data);
+    return encoder.encode(message);
+  }
+
+  void close() {
+    // TODO(floitsch): what should happen when an IsolateSink is closed?

[Lasse Reichstein Nielsen, 2012/11/22 10:23:15]

You are sending the 'done' event to the other end, and you can't send more
events on this end. You can do whatever is consistent with that.
The big question is ofcourse whether the associated stream is also closed, but
the other end should be deciding that - it's the one who can close that stream.

[floitsch, 2012/11/22 11:58:04]

We will discuss this in person.

+  }
+
+  /**
+   * Tests whether [other] is an [IsolateSink] feeding into the same
+   * [IsolateStream] as this one.
+   */
+  bool operator==(var other) {
+    return other is IsolateSink && _port == other._port;
+  }
+
+  int get hashCode => _port.hashCode + 499;
+}
+
+
+/**
+ * Creates and spawns an isolate that shares the same code as the current
+ * isolate, but that starts from [topLevelFunction]. 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.

[Lasse Reichstein Nielsen, 2012/11/22 10:23:15]

"closure" isn't defined. The previous two cases define the valid cases well
enough, so just stop after that.
(If you want to elaborate, say that a top-level function is a declared function,
not just a function expression living at the top level).

[floitsch, 2012/11/22 11:58:04]

Done.

+ *
+ * When any isolate starts (even the main script of the application), a default
+ * [IsolateStream] is created for it. This sink is available from the top-level
+ * getter [stream] defined in this library.
+ *
+ * [spawnFunction] returns an [IsolateSink] feeding into the child isolate's
+ * default stream.
+ *
+ * See comments at the top of this library for more details.
+ */
+IsolateSink streamSpawnFunction(void topLevelFunction()) {
+  SendPort sendPort = spawnFunction(topLevelFunction);
+  return new IsolateSink(sendPort);
+}
+
+/**
+ * Creates and spawns an isolate whose code is available at [uri].  Like with
+ * [streamSpawnFunction], the child isolate will have a default [IsolateStream],
+ * and a this function returns an [IsolateSink] feeding into it.
+ *
+ * See comments at the top of this library for more details.
+ */
+IsolateSink streamSpawnUri(String uri) {
+  SendPort sendPort = spawnUri(uri);
+  return new IsolateSink(sendPort);
+}