Avoid sending needless data between isolates in "pub serve" and "pub build"

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

Index: sdk/lib/_internal/pub/lib/src/barback/load_transformers.dart
diff --git a/sdk/lib/_internal/pub/lib/src/barback/load_transformers.dart b/sdk/lib/_internal/pub/lib/src/barback/load_transformers.dart
index 3eb02b7219878d829d38d9ca8e5fdeb69f79f8af..7a5e0f32c54efd09540340298c45e1244e429d4a 100644
--- a/sdk/lib/_internal/pub/lib/src/barback/load_transformers.dart
+++ b/sdk/lib/_internal/pub/lib/src/barback/load_transformers.dart
@@ -214,29 +214,45 @@ Map _serializeTransformerGroup(TransformerGroup group) {
   };
 }
 
-/// 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) {
+/// Converts a serializable map into an [Asset].
+Asset _deserializeAsset(Map asset) {
+  return new Asset.fromStream(
+      _deserializeId(asset['id']),
+      _deserializeStream(asset['stream']));
+}
+
+/// The body of a [StreamTransformer] that deserializes the values in a stream
+/// sent by [_serializeStream].
+StreamSubscription _deserializeTransformer(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);
-        }
-      }));
+    if (data['type'] == 'data') {
+      sink.add(data['data']);
+    } else if (data['type'] == 'error') {
+      sink.addError(CrossIsolateException.deserialize(data['error']));
+    } else {
+      assert(data['type'] == 'done');
+      sink.close();
+      subscription.cancel();
+    }
+  }));
   subscription = transformed.listen(null, cancelOnError: cancelOnError);
   return subscription;
 }
 
-/// Converts a serializable map into an [Asset].
-Asset _deserializeAsset(Map asset) {
-  var receivePort = new ReceivePort();
-  asset['sendPort'].send(receivePort.sendPort);
-  var stream = receivePort.transform(const StreamTransformer(doneTransformer));
-  return new Asset.fromStream(_deserializeId(asset['id']), stream);
+/// Convert a [SendPort] whose opposite is waiting to send us a stream into a
+/// [Stream].
+///
+/// No stream data will actually be sent across the isolate boundary until
+/// someone subscribes to the returned stream.
+Stream _deserializeStream(SendPort sendPort) {
+  return callbackStream(() {
+    var receivePort = new ReceivePort();
+    sendPort.send(receivePort.sendPort);
+    return receivePort.transform(
+        const StreamTransformer(_deserializeTransformer));
+  });
 }
 
 /// Converts a serializable map into an [AssetId].
@@ -244,21 +260,28 @@ 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.
+  return {
+    'id': _serializeId(asset.id),
+    'stream': _serializeStream(asset.read())
+  };
+}
+
+/// Converts [stream] into a [SendPort] with which another isolate can request
+/// the data from [stream].
+SendPort _serializeStream(Stream stream) {
   var receivePort = new ReceivePort();
   receivePort.first.then((sendPort) {
-    asset.read().listen(sendPort.send,
+    stream.listen((data) => sendPort.send({'type': 'data', 'data': data}),
+        onDone: () => sendPort.send({'type': 'done'}),
         onError: (error, stackTrace) {
-          throw new UnimplementedError('Error during asset serialization');
-        },
-        onDone: () { sendPort.send('done'); });
+      sendPort.send({
+        'type': 'error',
+        'error': CrossIsolateException.serialize(error, stackTrace)
+      });
+    });
   });
 
-  return {
-    'id': _serializeId(asset.id),
-    'sendPort': receivePort.sendPort
-  };
+  return receivePort.sendPort;
 }
 
 /// Converts [id] into a serializable map.
@@ -392,6 +415,21 @@ Stream _futureStream(Future<Stream> future) {
   });
   return controller.stream;
 }
+
+Stream callbackStream(Stream callback()) {
+  var subscription;
+  var controller;
+  controller = new StreamController(onListen: () {
+    subscription = callback().listen(controller.add,
+        onError: controller.addError,
+        onDone: controller.close);
+  },
+      onCancel: () => subscription.cancel(),
+      onPause: () => subscription.pause(),
+      onResume: () => subscription.resume(),
+      sync: true);
+  return controller.stream;
+}
 """;
 
 /// Load and return all transformers and groups from the library identified by
@@ -528,29 +566,45 @@ Map _serializeTransform(Transform transform) {
   };
 }
 
-/// 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) {
+/// Converts a serializable map into an [Asset].
+Asset _deserializeAsset(Map asset) {
+  return new Asset.fromStream(
+      _deserializeId(asset['id']),
+      _deserializeStream(asset['stream']));
+}
+
+/// A transformer that deserializes the values in a stream sent by
+/// [_serializeStream].
+StreamSubscription _deserializeTransformer(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);
-        }
-      }));
+    if (data['type'] == 'data') {
+      sink.add(data['data']);
+    } else if (data['type'] == 'error') {
+      sink.addError(CrossIsolateException.deserialize(data['error']));
+    } else {
+      assert(data['type'] == 'done');
+      sink.close();
+      subscription.cancel();
+    }
+  }));
   subscription = transformed.listen(null, cancelOnError: cancelOnError);
   return subscription;
 }
 
-/// Converts a serializable map into an [Asset].
-Asset _deserializeAsset(Map asset) {
-  var receivePort = new ReceivePort();
-  asset['sendPort'].send(receivePort.sendPort);
-  var stream = receivePort.transform(const StreamTransformer(doneTransformer));
-  return new Asset.fromStream(_deserializeId(asset['id']), stream);
+/// Convert a [SendPort] whose opposite is waiting to send us a stream into a
+/// [Stream].
+///
+/// No stream data will actually be sent across the isolate boundary until
+/// someone subscribes to the returned stream.
+Stream _deserializeStream(SendPort sendPort) {
+  return callbackStream(() {
+    var receivePort = new ReceivePort();

[Bob Nystrom, 2013/10/28 23:52:49]

This needs to be explicitly closed now, right?

Does this just rely on the stream returned by this being closed?

[nweiz, 2013/10/29 00:15:55]

Once a "done" event is received, the subscription on the receivePort will be
closed, which will close the port itself. Alternately if a user closes the
returned stream early the port will be closed as well.

+    sendPort.send(receivePort.sendPort);
+    return receivePort.transform(
+        const StreamTransformer(_deserializeTransformer));
+  });
 }
 
 /// Converts a serializable map into an [AssetId].
@@ -575,21 +629,28 @@ Location _deserializeLocation(Map location) {
 // 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.
+  return {
+    'id': _serializeId(asset.id),
+    'stream': _serializeStream(asset.read())
+  };
+}
+
+/// Converts [stream] into a [SendPort] with which another isolate can request
+/// the data from [stream].
+SendPort _serializeStream(Stream stream) {
   var receivePort = new ReceivePort();
   receivePort.first.then((sendPort) {
-    asset.read().listen(sendPort.send,
+    stream.listen((data) => sendPort.send({'type': 'data', 'data': data}),
+        onDone: () => sendPort.send({'type': 'done'}),
         onError: (error, stackTrace) {
-          throw new UnimplementedError('Error during asset serialization');
-        },
-        onDone: () { sendPort.send('done'); });
+      sendPort.send({
+        'type': 'error',
+        'error': CrossIsolateException.serialize(error, stackTrace)
+      });
+    });
   });
 
-  return {
-    'id': _serializeId(asset.id),
-    'sendPort': receivePort.sendPort
-  };
+  return receivePort.sendPort;
 }
 
 /// Converts [id] into a serializable map.