Fix a barback bug where a result could be emitted incorrectly.

pkg/barback/lib/src/asset_cascade.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 barback.asset_cascade;
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'asset.dart';
 import 'asset_id.dart';
 import 'asset_node.dart';
 import 'asset_set.dart';
 import 'build_result.dart';
 import 'cancelable_future.dart';
 import 'errors.dart';
 import 'package_graph.dart';
 import 'phase.dart';
+import 'stream_pool.dart';
 import 'transformer.dart';
 import 'utils.dart';
 
 /// The asset cascade for an individual package.
 ///
 /// This keeps track of which [Transformer]s are applied to which assets, and
 /// re-runs those transformers when their dependencies change. The transformed
 /// asset nodes are accessible via [getAssetNode].
 ///
 /// A cascade consists of one or more [Phases], each of which has one or more
@@ skipping 33 matching lines @@
   /// A stream that emits any errors from the cascade or the transformers.
   ///
   /// This emits errors as they're detected. If an error occurs in one part of
   /// the cascade, unrelated parts will continue building.
   ///
   /// This will not emit programming errors from barback itself. Those will be
   /// emitted through the [results] stream's error channel.
   Stream<BarbackException> get errors => _errorsController.stream;
   final _errorsController = new StreamController<BarbackException>.broadcast();
 
+  /// A stream that emits an event whenever this cascade becomes dirty.
+  ///
+  /// After this stream emits an event, [results] will emit an event once the
+  /// cascade is no longer dirty.
+  ///
+  /// This may emit events when the cascade was already dirty. Events are
+  /// emitted synchronously to ensure that the dirty state is thoroughly
+  /// propagated as soon as any assets are changed.
+  Stream get onDirty => _onDirtyPool.stream;
+  final _onDirtyPool = new StreamPool.broadcast();
+
+  /// A controller whose stream feeds into [_onDirtyPool].
+  final _onDirtyController = new StreamController.broadcast(sync: true);
+
   /// The errors that have occurred since the current build started.
   ///
   /// This will be empty if no build is occurring.
   Queue<BarbackException> _accumulatedErrors;
 
   /// A future that completes when the currently running build process finishes.
   ///
   /// If no build it in progress, is `null`.
   Future _processDone;
 
   /// Whether any source assets have been updated or removed since processing
   /// last began.
   var _newChanges = false;
 
   /// Returns all currently-available output assets from this cascade.
   AssetSet get availableOutputs => _phases.last.availableOutputs;
 
   /// Creates a new [AssetCascade].
   ///
   /// It loads source assets within [package] using [provider].
   AssetCascade(this.graph, this.package) {
+    _onDirtyPool.add(_onDirtyController.stream);
     _addPhase(new Phase(this, []));
   }
 
   /// Gets the asset identified by [id].
   ///
   /// If [id] is for a generated or transformed asset, this will wait until it
   /// has been created and return it. This means that the returned asset will
   /// always be [AssetState.AVAILABLE].
   ///
   /// If the asset cannot be found, returns null.
@@ skipping 91 matching lines @@
     }
   }
 
   void reportError(BarbackException error) {
     _accumulatedErrors.add(error);
     _errorsController.add(error);
   }
 
   /// Add [phase] to the end of [_phases] and watch its [onDirty] stream.
   void _addPhase(Phase phase) {
+    _onDirtyPool.add(phase.onDirty);
     phase.onDirty.listen((_) {
       _newChanges = true;
       _waitForProcess();
     });
     _phases.add(phase);
   }
 
   /// Starts the build process asynchronously if there is work to be done.
   ///
   /// Returns a future that completes with the background processing is done.
@@ skipping 45 matching lines @@
 
         // Otherwise, everything is done.
         return;
       }
 
       // Process that phase and then loop onto the next.
       return future.then((_) => _process());
     });
   }
 }