Serialize barback assets more efficiently.

pkg/barback/lib/src/serialize.dart

Index: pkg/barback/lib/src/serialize.dart
diff --git a/pkg/barback/lib/src/serialize.dart b/pkg/barback/lib/src/serialize.dart
new file mode 100644
index 0000000000000000000000000000000000000000..049afc9613743bf89b1b465e4953e6f49f4b8253
--- /dev/null
+++ b/pkg/barback/lib/src/serialize.dart
@@ -0,0 +1,69 @@
+// 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 barback.serialize;
+
+import 'dart:async';
+import 'dart:isolate';
+
+import 'asset_id.dart';
+import 'utils.dart';
+
+/// Converts [id] into a serializable map.
+Map serializeId(AssetId id) => {'package': id.package, 'path': id.path};
+
+/// Converts [stream] into a [SendPort] with which another isolate can request
+/// the data from [stream].
+SendPort serializeStream(Stream stream) {
+  var receivePort = new ReceivePort();

[Bob Nystrom, 2013/11/06 21:29:08]

Does this get closed by .first?

[nweiz, 2013/11/07 00:44:48]

Yes, thank goodness. [Stream.first] cancels its subscription after the first
message is received, and canceling a subscription on a [ReceivePort] closes the
port.

+  receivePort.first.then((sendPort) {
+    stream.listen((data) => sendPort.send({'type': 'data', 'data': data}),
+        onDone: () => sendPort.send({'type': 'done'}),
+        onError: (error, stackTrace) {
+      sendPort.send({
+        'type': 'error',
+        'error': CrossIsolateException.serialize(error, stackTrace)
+      });
+    });
+  });
+
+  return receivePort.sendPort;
+}
+
+/// Converts a serializable map into an [AssetId].
+AssetId deserializeId(Map id) => new AssetId(id['package'], id['path']);
+
+/// 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));
+  });
+}
+
+/// 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['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;
+}