Very simple version of Isolates.

sdk/lib/_internal/pub/lib/src/barback/load_transformers.dart

 // Copyright (c) 2013, 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.
 
 library pub.load_transformers;
 
 import 'dart:async';
 import 'dart:convert';
 import 'dart:isolate';
 
@@ skipping 14 matching lines @@
 import 'dart:async';
 import 'dart:isolate';
 import 'dart:convert';
 import 'dart:mirrors';
 
 import 'http://<<HOST_AND_PORT>>/packages/source_maps/span.dart';
 import 'http://<<HOST_AND_PORT>>/packages/stack_trace/stack_trace.dart';
 import 'http://<<HOST_AND_PORT>>/packages/barback/barback.dart';
 
 /// Sets up the initial communication with the host isolate.
-void main() {
-  port.receive((args, replyTo) {
-    _sendFuture(replyTo, new Future.sync(() {
+void main({message}) {
+  var port = new ReceivePort();
+  message.send(['success', port.sendPort]);
+  port.listen((args) {
+    _sendFuture(args['replyTo'], new Future.sync(() {
       var library = Uri.parse(args['library']);
       var configuration = JSON.decode(args['configuration']);
       return initialize(library, configuration).
           map(_serializeTransformerOrGroup).toList();
     }));
   });
 }
 
 /// Loads all the transformers and groups defined in [uri].
 ///
@@ skipping 57 matching lines @@
         'type': 'log',
         'level': level.name,
         'message': message,
         'assetId': assetId == null ? null : _serializeId(assetId),
         'span': span == null ? null : _serializeSpan(span)
       });
     });
   }
 
   Future<Asset> getInput(AssetId id) {
-    return _receiveFuture(_port.call({
+    return _callAndReceiveFuture(_port, {
       'type': 'getInput',
       'id': _serializeId(id)
-    })).then(_deserializeAsset);
+    }).then(_deserializeAsset);
   }
 
   Future<String> readInputAsString(AssetId id, {Encoding encoding}) {
     if (encoding == null) encoding = UTF8;
     return getInput(id).then((input) => input.readAsString(encoding: encoding));
   }
 
   Stream<List<int>> readInput(AssetId id) =>
       _futureStream(getInput(id).then((input) => input.read()));
 
@@ skipping 45 matching lines @@
     return _serializeTransformer(transformerOrGroup);
   } else {
     assert(transformerOrGroup is TransformerGroup);
     return _serializeTransformerGroup(transformerOrGroup);
   }
 }
 
 /// Converts [transformer] into a serializable map.
 Map _serializeTransformer(Transformer transformer) {
   var port = new ReceivePort();
-  port.receive((message, replyTo) {
+  port.listen((message) {
+    var replyTo = message['replyTo'];
     _sendFuture(replyTo, new Future.sync(() {
       if (message['type'] == 'isPrimary') {
         return transformer.isPrimary(_deserializeAsset(message['asset']));
       } else {
         assert(message['type'] == 'apply');
         return transformer.apply(
             new ForeignTransform(message['transform']));
       }
     }));
   });
 
   return {
     'type': 'Transformer',
     'toString': transformer.toString(),
-    'port': port.toSendPort()
+    'port': port.sendPort
   };
 }
 
 // Converts [group] into a serializable map.
 Map _serializeTransformerGroup(TransformerGroup group) {
   return {
     'type': 'TransformerGroup',
     'toString': group.toString(),
     'phases': group.phases.map((phase) {
       return phase.map(_serializeTransformerOrGroup).toList();
     }).toList()
   };
 }
 
+/// When the input receives a 'done' as data-event, transforms it to a 
+/// done event and cancels the subscription.
+StreamSubscription doneTransformer(Stream input, bool cancelOnError) {
+  var subscription;
+  var transformed = input.transform(new StreamTransformer.fromHandlers(
+      handleData: (data, sink) {
+        if (data == 'done') {
+          sink.close();
+          subscription.cancel();
+        } else {
+          sink.add(data);
+        }
+      }));
+  subscription = transformed.listen(null, cancelOnError: cancelOnError);
+  return subscription;
+}
+
 /// Converts a serializable map into an [Asset].
 Asset _deserializeAsset(Map asset) {
-  var box = new MessageBox();
-  asset['sink'].add(box.sink);
-  return new Asset.fromStream(_deserializeId(asset['id']), box.stream);
+  var receivePort = new ReceivePort();
+  asset['sendPort'].send(receivePort.sendPort);
+  var stream = receivePort.transform(const StreamTransformer(doneTransformer));
+  return new Asset.fromStream(_deserializeId(asset['id']), stream);
 }
 
 /// Converts a serializable map into an [AssetId].
 AssetId _deserializeId(Map id) => new AssetId(id['package'], id['path']);
 
 /// Converts [asset] into a serializable map.
 Map _serializeAsset(Asset asset) {
   // We can't send IsolateStreams (issue 12437), so instead we send a sink and
   // get the isolate to send us back another sink.
-  var box = new MessageBox();
-  box.stream.first.then((sink) {
-    asset.read().listen(sink.add,
-        onError: sink.addError,
-        onDone: sink.close);
+  var receivePort = new ReceivePort();
+  receivePort.first.then((sendPort) {
+    asset.read().listen(sendPort.send,
+        onError: (error, stackTrace) {
+          throw new UnimplementedError('Error during asset serialization');
+        },
+        onDone: () { sendPort.send('done'); });
   });
 
   return {
     'id': _serializeId(asset.id),
-    'sink': box.sink
+    'sendPort': receivePort.sendPort
   };
 }
 
 /// Converts [id] into a serializable map.
 Map _serializeId(AssetId id) => {'package': id.package, 'path': id.path};
 
 /// Converts [span] into a serializable map.
 Map _serializeSpan(Span span) {
   // TODO(nweiz): convert FileSpans to FileSpans.
   return {
@@ skipping 25 matching lines @@
   future.then((result) {
     port.send({'success': result});
   }).catchError((error) {
     // TODO(nweiz): at least MissingInputException should be preserved here.
     port.send({'error': CrossIsolateException.serialize(error)});
   });
 }
 
 /// Receives the result of [_sendFuture] from [portCall], which should be the
 /// return value of [SendPort.call].
-Future _receiveFuture(Future portCall) {
-  return portCall.then((response) {
-    if (response.containsKey('success')) return response['success'];
-    return new Future.error(
-        new CrossIsolateException.deserialize(response['error']));
+///
+/// The [message] argument is modified to include the [replyTo] port.
+Future _callAndReceiveFuture(SendPort port, Map message) {
+  var responsePort = new ReceivePort();
+  message['replyTo'] = responsePort.sendPort;
+  return new Future.sync(() {
+    port.send(message);
+    return responsePort.first.then((response) {
+      if (response.containsKey('success')) return response['success'];
+      return new Future.error(
+          new CrossIsolateException.deserialize(response['error']));
+    });
   });
 }
 
 /// An exception that was originally raised in another isolate.
 ///
 /// Exception objects can't cross isolate boundaries in general, so this class
 /// wraps as much information as can be consistently serialized.
 class CrossIsolateException implements Exception {
   /// The name of the type of exception thrown.
   ///
@@ skipping 74 matching lines @@
   return id.getAssetId(server.barback).then((assetId) {
     var path = assetId.path.replaceFirst('lib/', '');
     // TODO(nweiz): load from a "package:" URI when issue 12474 is fixed.
     var hostAndPort = '${server.address.address}:${server.port}';
     var uri = 'http://$hostAndPort/packages/${id.package}/$path';
     var code = 'import "$uri";' +
         _TRANSFORMER_ISOLATE.replaceAll('<<HOST_AND_PORT>>', hostAndPort);
     log.fine("Loading transformers from $assetId");
 
     return dart.runInIsolate(code).then((sendPort) {
-      return _receiveFuture(sendPort.call({
+      return _callAndReceiveFuture(sendPort, {
         'library': uri,
         // TODO(nweiz): support non-JSON-encodable configuration maps.
         'configuration': JSON.encode(id.configuration)
-      })).then((transformers) {
+      }).then((transformers) {
         transformers = transformers.map(_deserializeTransformerOrGroup).toSet();
         log.fine("Transformers from $assetId: $transformers");
         return transformers;
       });
     }).catchError((error) {
       if (error is! dart.CrossIsolateException) throw error;
       if (error.type != 'IsolateSpawnException') throw error;
       // TODO(nweiz): don't parse this as a string once issues 12617 and 12689
       // are fixed.
       if (!error.message.split('\n')[1].startsWith('import "$uri";')) {
@@ skipping 18 matching lines @@
   final SendPort _port;
 
   /// The result of calling [toString] on the transformer in the isolate.
   final String _toString;
 
   _ForeignTransformer(Map map)
       : _port = map['port'],
         _toString = map['toString'];
 
   Future<bool> isPrimary(Asset asset) {
-    return _receiveFuture(_port.call({
+    return _callAndReceiveFuture(_port, {
       'type': 'isPrimary',
       'asset': _serializeAsset(asset)
-    }));
+    });
   }
 
   Future apply(Transform transform) {
-    return _receiveFuture(_port.call({
+    return _callAndReceiveFuture(_port, {
       'type': 'apply',
       'transform': _serializeTransform(transform)
-    }));
+    });
   }
 
   String toString() => _toString;
 }
 
 /// A wrapper for a transformer group that's in a different isolate.
 class _ForeignGroup implements TransformerGroup {
   final Iterable<Iterable> phases;
 
   /// The result of calling [toString] on the transformer group in the isolate.
@@ skipping 11 matching lines @@
 /// Converts a serializable map into a [Transformer] or a [TransformerGroup].
 _deserializeTransformerOrGroup(Map map) {
   if (map['type'] == 'Transformer') return new _ForeignTransformer(map);
   assert(map['type'] == 'TransformerGroup');
   return new _ForeignGroup(map);
 }
 
 /// Converts [transform] into a serializable map.
 Map _serializeTransform(Transform transform) {
   var receivePort = new ReceivePort();
-  receivePort.receive((message, replyTo) {
+  receivePort.listen((message) {
+    var replyTo = message['replyTo'];
     if (message['type'] == 'getInput') {
       _sendFuture(replyTo, transform.getInput(_deserializeId(message['id']))
           .then(_serializeAsset));
     } else if (message['type'] == 'addOutput') {
       transform.addOutput(_deserializeAsset(message['output']));
     } else {
       assert(message['type'] == 'log');
 
       var method;
       if (message['level'] == 'Info') {
         method = transform.logger.info;
       } else if (message['level'] == 'Warning') {
         method = transform.logger.warning;
       } else {
         assert(message['level'] == 'Error');
         method = transform.logger.error;
       }
 
       var assetId = message['assetId'] == null ? null :
         _deserializeId(message['assetId']);
       var span = message['span'] == null ? null :
         _deserializeSpan(message['span']);
       method(message['message'], asset: assetId, span: span);
     }
   });
 
   return {
-    'port': receivePort.toSendPort(),
+    'port': receivePort.sendPort,
     'primaryInput': _serializeAsset(transform.primaryInput)
   };
 }
 
+/// When the input receives a 'done' as data-event, transforms it to a
+/// done event and cancels the subscription.
+StreamSubscription doneTransformer(Stream input, bool cancelOnError) {
+  var subscription;
+  var transformed = input.transform(new StreamTransformer.fromHandlers(
+      handleData: (data, sink) {
+        if (data == 'done') {
+          sink.close();
+          subscription.cancel();
+        } else {
+          sink.add(data);
+        }
+      }));
+  subscription = transformed.listen(null, cancelOnError: cancelOnError);
+  return subscription;
+}
+
 /// Converts a serializable map into an [Asset].
 Asset _deserializeAsset(Map asset) {
-  var box = new MessageBox();
-  asset['sink'].add(box.sink);
-  return new Asset.fromStream(_deserializeId(asset['id']), box.stream);
+  var receivePort = new ReceivePort();
+  asset['sendPort'].send(receivePort.sendPort);
+  var stream = receivePort.transform(const StreamTransformer(doneTransformer));
+  return new Asset.fromStream(_deserializeId(asset['id']), stream);
 }
 
 /// Converts a serializable map into an [AssetId].
 AssetId _deserializeId(Map id) => new AssetId(id['package'], id['path']);
 
 /// Converts a serializable map into a [Span].
 Span _deserializeSpan(Map span) {
   assert(span['type'] == 'fixed');
   var location = _deserializeLocation(span['start']);
   return new FixedSpan(span['sourceUrl'], location.offset, location.line,
       location.column, text: span['text'], isIdentifier: span['isIdentifier']);
 }
 
 /// Converts a serializable map into a [Location].
 Location _deserializeLocation(Map location) {
   assert(location['type'] == 'fixed');
   return new FixedLocation(location['offset'], location['sourceUrl'],
       location['line'], location['column']);
 }
 
 // TODO(nweiz): add custom serialization code for assets that can be more
 // efficiently serialized.
 /// Converts [asset] into a serializable map.
 Map _serializeAsset(Asset asset) {
   // We can't send IsolateStreams (issue 12437), so instead we send a sink and
   // get the isolate to send us back another sink.
-  var box = new MessageBox();
-  box.stream.first.then((sink) {
-    asset.read().listen(sink.add,
-        onError: sink.addError,
-        onDone: sink.close);
+  var receivePort = new ReceivePort();
+  receivePort.first.then((sendPort) {
+    asset.read().listen(sendPort.send,
+        onError: (error, stackTrace) {
+          throw new UnimplementedError('Error during asset serialization');
+        },
+        onDone: () { sendPort.send('done'); });
   });
 
   return {
     'id': _serializeId(asset.id),
-    'sink': box.sink
+    'sendPort': receivePort.sendPort
   };
 }
 
 /// Converts [id] into a serializable map.
 Map _serializeId(AssetId id) => {'package': id.package, 'path': id.path};
 
 /// Sends the result of [future] through [port].
 ///
-/// This should be received on the other end using [_receiveFuture]. It
+/// This should be received on the other end using [_callAndReceiveFuture]. It
 /// re-raises any exceptions on the other side as [dart.CrossIsolateException]s.
 void _sendFuture(SendPort port, Future future) {
   future.then((result) {
     port.send({'success': result});
   }).catchError((error) {
     // TODO(nweiz): at least MissingInputException should be preserved here.
     port.send({'error': dart.CrossIsolateException.serialize(error)});
   });
 }
 
 /// Receives the result of [_sendFuture] from [portCall], which should be the
 /// return value of [SendPort.call].
-Future _receiveFuture(Future portCall) {
-  return portCall.then((response) {
-    if (response.containsKey('success')) return response['success'];
-    return new Future.error(
-        new dart.CrossIsolateException.deserialize(response['error']));
+///
+/// The [message] argument is modified to include the [replyTo] port.
+Future _callAndReceiveFuture(SendPort port, Map message) {
+  var responsePort = new ReceivePort();
+  message['replyTo'] = responsePort.sendPort;
+  return new Future.sync(() {
+    port.send(message);
+    return responsePort.first.then((response) {
+      if (response.containsKey('success')) return response['success'];
+      return new Future.error(
+          new dart.CrossIsolateException.deserialize(response['error']));
+    });
   });
 }