Fix a bug where barback transformers would produce stale output.

pkg/barback/lib/src/transformer/aggregate_transform.dart

 // Copyright (c) 2014, 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.transformer.aggregate_transform;
 
 import 'dart:async';
 import 'dart:convert';
 
 import '../asset/asset.dart';
@@ skipping 22 matching lines @@
   ///
   /// This is exposed as a stream so that the transformer can start working
   /// before all its inputs are available. The stream is closed not just when
   /// all inputs are provided, but when barback is confident no more inputs will
   /// be forthcoming.
   ///
   /// A transformer may complete its `apply` method before this stream is
   /// closed. For example, it may know that each key will only have two inputs
   /// associated with it, and so use `transform.primaryInputs.take(2)` to access
   /// only those inputs.
-  Stream<Asset> get primaryInputs => _primaryInputs;
-  Stream<Asset> _primaryInputs;
-
-  /// The controller for [primaryInputs].
-  ///
-  /// This is a broadcast controller so that the transform can keep
-  /// [_emittedPrimaryInputs] up to date.
-  final _inputController = new StreamController<Asset>.broadcast();
+  Stream<Asset> get primaryInputs => _inputController.stream;
+  final _inputController = new StreamController<Asset>();
 
   /// The set of all primary inputs that have been emitted by [primaryInputs].
+  ///
+  /// This is populated by the transform's controller so that
+  /// [AggregateTransformController.addedId] synchronously returns the correct
+  /// result after [AggregateTransformController.addInput] is called.
   final _emittedPrimaryInputs = new AssetSet();
 
   AggregateTransform._(TransformNode node)
       : _node = node,
-        super(node) {
-    _inputController.stream.listen(_emittedPrimaryInputs.add);
-    // [primaryInputs] should be a non-broadcast stream.
-    _primaryInputs = broadcastToSingleSubscription(_inputController.stream);
-  }
+        super(node);
 
   /// Gets the asset for an input [id].
   ///
   /// If an input with [id] cannot be found, throws an [AssetNotFoundException].
   Future<Asset> getInput(AssetId id) {
     if (_emittedPrimaryInputs.containsId(id)) {
       return syncFuture(() => _emittedPrimaryInputs[id]);
     } else {
       return _node.getInput(id);
     }
@@ skipping 60 matching lines @@
   /// The set of assets that the transformer has emitted.
   AssetSet get outputs => transform._outputs;
 
   bool get isDone => transform._inputController.isClosed;
 
   AggregateTransformController(TransformNode node)
       : super(new AggregateTransform._(node));
 
   /// Adds a primary input asset to the [AggregateTransform.primaryInputs]
   /// stream.
-  void addInput(Asset input) => transform._inputController.add(input);
+  void addInput(Asset input) {
+    transform._emittedPrimaryInputs.add(input);
+    transform._inputController.add(input);
+  }
 
   /// Returns whether an input with the given [id] was added via [addInput].
-  bool addedId(AssetId id) {
-    return transform._emittedPrimaryInputs.ids.contains(id);
-  }
+  bool addedId(AssetId id) =>
+      transform._emittedPrimaryInputs.containsId(id);
 
   void done() {
     transform._inputController.close();
   }
 }