Expose aggregate transformers in pub.

dart/sdk/lib/_internal/pub/asset/dart/serialize/aggregate_transform.dart

Index: dart/sdk/lib/_internal/pub/asset/dart/serialize/aggregate_transform.dart
diff --git a/dart/sdk/lib/_internal/pub/asset/dart/serialize/transform.dart b/dart/sdk/lib/_internal/pub/asset/dart/serialize/aggregate_transform.dart
similarity index 58%
copy from dart/sdk/lib/_internal/pub/asset/dart/serialize/transform.dart
copy to dart/sdk/lib/_internal/pub/asset/dart/serialize/aggregate_transform.dart
index 38e2ff1d0f48ad1ecc18cebf081b1c11c150870b..da8a5afae86e90929574af115ba73e15a1143796 100644
--- a/dart/sdk/lib/_internal/pub/asset/dart/serialize/transform.dart
+++ b/dart/sdk/lib/_internal/pub/asset/dart/serialize/aggregate_transform.dart
@@ -2,26 +2,26 @@
 // 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.asset.serialize.transform;
+library pub.asset.serialize.aggregate_transform;
 
 import 'dart:async';
 import 'dart:isolate';
-import 'dart:convert';
 
 import 'package:barback/barback.dart';
 // TODO(nweiz): don't import from "src" once issue 14966 is fixed.
 import 'package:barback/src/internal_asset.dart';
 
 import '../serialize.dart';
-import '../utils.dart';
+import 'get_input_transform.dart';
 
-/// Serialize the methods shared between [Transform] and [DeclaringTransform].
+/// Serialize the methods shared between [AggregateTransform] and
+/// [DeclaringAggregateTransform].
 ///
 /// [additionalFields] contains additional serialized fields to add to the
 /// serialized transform. [methodHandlers] is a set of additional methods. Each
 /// value should take a JSON message and return the response (which may be a
 /// Future).
-Map _serializeBaseTransform(transform, Map additionalFields,
+Map _serializeBaseAggregateTransform(transform, Map additionalFields,
     Map<String, Function> methodHandlers) {
   var receivePort = new ReceivePort();
   receivePort.listen((wrappedMessage) {
@@ -30,7 +30,7 @@ Map _serializeBaseTransform(transform, Map additionalFields,
       if (handler != null) return handler(message);
 
       if (message['type'] == 'consumePrimary') {
-        transform.consumePrimary();
+        transform.consumePrimary(deserializeId(message['assetId']));
         return null;
       }
 
@@ -51,13 +51,17 @@ Map _serializeBaseTransform(transform, Map additionalFields,
     });
   });
 
-  return {'port': receivePort.sendPort}..addAll(additionalFields);
+  return {
+    'port': receivePort.sendPort,
+    'key': transform.key,
+    'package': transform.package
+  }..addAll(additionalFields);
 }
 
 /// Converts [transform] into a serializable map.
-Map serializeTransform(Transform transform) {
-  return _serializeBaseTransform(transform, {
-    'primaryInput': serializeAsset(transform.primaryInput)
+Map serializeAggregateTransform(AggregateTransform transform) {
+  return _serializeBaseAggregateTransform(transform, {
+    'primaryInputs': serializeStream(transform.primaryInputs, serializeAsset)
   }, {
     'getInput': (message) => transform.getInput(deserializeId(message['id']))
         .then((asset) => serializeAsset(asset)),
@@ -67,27 +71,35 @@ Map serializeTransform(Transform transform) {
 }
 
 /// Converts [transform] into a serializable map.
-Map serializeDeclaringTransform(DeclaringTransform transform) {
-  return _serializeBaseTransform(transform, {
-    'primaryId': serializeId(transform.primaryId)
+Map serializeDeclaringAggregateTransform(
+    DeclaringAggregateTransform transform) {
+  return _serializeBaseAggregateTransform(transform, {
+    'primaryIds': serializeStream(transform.primaryIds, serializeId)
   }, {
     'declareOutput': (message) =>
         transform.declareOutput(deserializeId(message['output']))
   });
 }
 
-/// The base class for wrappers for [Transform]s that are in the host isolate.
-class _ForeignBaseTransform {
+/// The base class for wrappers for [AggregateTransform]s that are in the host
+/// isolate.
+class _ForeignBaseAggregateTransform {
   /// The port with which we communicate with the host isolate.
   ///
   /// This port and all messages sent across it are specific to this transform.
   final SendPort _port;
 
+  final String key;
+
+  final String package;
+
   TransformLogger get logger => _logger;
   TransformLogger _logger;
 
-  _ForeignBaseTransform(Map transform)
-      : _port = transform['port'] {
+  _ForeignBaseAggregateTransform(Map transform)
+      : _port = transform['port'],
+        key = transform['key'],
+        package = transform['package'] {
     _logger = new TransformLogger((assetId, level, message, span) {
       call(_port, {
         'type': 'log',
@@ -99,20 +111,28 @@ class _ForeignBaseTransform {
     });
   }
 
-  void consumePrimary() {
-    call(_port, {'type': 'consumePrimary'});
+  void consumePrimary(AssetId id) {
+    call(_port, {'type': 'consumePrimary', 'assetId': serializeId(id)});
   }
 }
 
-/// A wrapper for a [Transform] that's in the host isolate.
+// We can get away with only removing the class declarations in incompatible
+// barback versions because merely referencing undefined types in type
+// annotations isn't a static error. Only implementing an undefined interface is
+// a static error.
+//# if barback >=0.14.1-dev
+
+/// A wrapper for an [AggregateTransform] that's in the host isolate.
 ///
 /// This retrieves inputs from and sends outputs and logs to the host isolate.
-class ForeignTransform extends _ForeignBaseTransform implements Transform {
-  final Asset primaryInput;
+class ForeignAggregateTransform extends _ForeignBaseAggregateTransform
+    with GetInputTransform implements AggregateTransform {
+  final Stream<Asset> primaryInputs;
 
   /// Creates a transform from a serialized map sent from the host isolate.
-  ForeignTransform(Map transform)
-      : primaryInput = deserializeAsset(transform['primaryInput']),
+  ForeignAggregateTransform(Map transform)
+      : primaryInputs = deserializeStream(
+            transform['primaryInputs'], deserializeAsset),
         super(transform);
 
   Future<Asset> getInput(AssetId id) {
@@ -122,21 +142,6 @@ class ForeignTransform extends _ForeignBaseTransform implements Transform {
     }).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()));
-
-  Future<bool> hasInput(AssetId id) {
-    return getInput(id).then((_) => true).catchError((error) {
-      if (error is AssetNotFoundException && error.id == id) return false;
-      throw error;
-    });
-  }
-
   void addOutput(Asset output) {
     call(_port, {
       'type': 'addOutput',
@@ -145,14 +150,16 @@ class ForeignTransform extends _ForeignBaseTransform implements Transform {
   }
 }
 
-/// A wrapper for a [DeclaringTransform] that's in the host isolate.
-class ForeignDeclaringTransform extends _ForeignBaseTransform
-    implements DeclaringTransform {
-  final AssetId primaryId;
+/// A wrapper for a [DeclaringAggregateTransform] that's in the host isolate.
+class ForeignDeclaringAggregateTransform
+    extends _ForeignBaseAggregateTransform
+    implements DeclaringAggregateTransform {
+  final Stream<AssetId> primaryIds;
 
   /// Creates a transform from a serializable map sent from the host isolate.
-  ForeignDeclaringTransform(Map transform)
-      : primaryId = deserializeId(transform['primaryId']),
+  ForeignDeclaringAggregateTransform(Map transform)
+      : primaryIds = deserializeStream(
+            transform['primaryIds'], deserializeId),
         super(transform);
 
   void declareOutput(AssetId id) {
@@ -162,3 +169,5 @@ class ForeignDeclaringTransform extends _ForeignBaseTransform
     });
   }
 }
+
+//# end