Add ReentrantSynchronousStream and use it for cache invalidation events.

pkg/analyzer/lib/src/context/cache.dart

Index: pkg/analyzer/lib/src/context/cache.dart
diff --git a/pkg/analyzer/lib/src/context/cache.dart b/pkg/analyzer/lib/src/context/cache.dart
index 99e08638fe2363613597a5b10b480a84494958e8..4e6607b5628fbea9fa3925f7c0627165a6df5ed8 100644
--- a/pkg/analyzer/lib/src/context/cache.dart
+++ b/pkg/analyzer/lib/src/context/cache.dart
@@ -38,8 +38,8 @@ class AnalysisCache {
   /**
    * The [StreamController] reporting [InvalidatedResult]s.
    */
-  final StreamController<InvalidatedResult> _onResultInvalidated =
-      new StreamController<InvalidatedResult>.broadcast(sync: true);
+  final ReentrantSynchronousStream<InvalidatedResult> onResultInvalidated =
+      new ReentrantSynchronousStream<InvalidatedResult>();
 
   /**
    * Initialize a newly created cache to have the given [partitions]. The
@@ -50,17 +50,11 @@ class AnalysisCache {
   AnalysisCache(this._partitions) {
     for (CachePartition partition in _partitions) {
       partition.onResultInvalidated.listen((InvalidatedResult event) {
-        _onResultInvalidated.add(event);
+        onResultInvalidated.add(event);
       });
     }
   }
 
-  /**
-   * Return the stream that is notified when a value is invalidated.
-   */
-  Stream<InvalidatedResult> get onResultInvalidated =>
-      _onResultInvalidated.stream;
-
   // TODO(brianwilkerson) Implement or delete this.
 //  /**
 //   * Return information about each of the partitions in this cache.
@@ -559,8 +553,8 @@ class CacheEntry {
       _partition._removeIfSource(target);
     }
     // Notify controller.
-    _partition._onResultInvalidated
-        .add(new InvalidatedResult(this, descriptor));
+    _partition.onResultInvalidated
+        .add(new InvalidatedResult(this, descriptor, thisData.value));
   }
 
   /**
@@ -807,8 +801,8 @@ abstract class CachePartition {
   /**
    * The [StreamController] reporting [InvalidatedResult]s.
    */
-  final StreamController<InvalidatedResult> _onResultInvalidated =
-      new StreamController<InvalidatedResult>.broadcast(sync: true);
+  final ReentrantSynchronousStream<InvalidatedResult> onResultInvalidated =
+      new ReentrantSynchronousStream<InvalidatedResult>();
 
   /**
    * A table mapping the targets belonging to this partition to the information
@@ -843,12 +837,6 @@ abstract class CachePartition {
   Map<AnalysisTarget, CacheEntry> get map => _targetMap;
 
   /**
-   * Return the stream that is notified when a value is invalidated.
-   */
-  Stream<InvalidatedResult> get onResultInvalidated =>
-      _onResultInvalidated.stream;
-
-  /**
    * Notifies the partition that the client is going to stop using it.
    */
   void dispose() {
@@ -1042,13 +1030,47 @@ class InvalidatedResult {
    */
   final ResultDescriptor descriptor;
 
-  InvalidatedResult(this.entry, this.descriptor);
+  /**
+   * The value of the result which was invalidated.
+   */
+  final Object value;
+
+  InvalidatedResult(this.entry, this.descriptor, this.value);
 
   @override
   String toString() => '$descriptor of ${entry.target}';
 }
 
 /**
+ * A Stream-like interface, which broadcasts events synchronously.
+ * If a second event is fired while delivering a first event, then the second
+ * event will be delivered first, and then delivering of the first will be
+ * continued.
+ */
+class ReentrantSynchronousStream<T> {
+  final List<Function> listeners = <Function>[];
+
+  /**
+   * Send the given [event] to the stream.
+   */
+  void add(T event) {
+    List<Function> listeners = this.listeners.toList();
+    for (Function listener in listeners) {
+      listener(event);
+    }
+  }
+
+  /**
+   * Listen for the events in this stream.
+   * Note that if the [listener] fires a new event, then the [listener] will be
+   * invoked again before returning from the [add] invocation.
+   */
+  void listen(void listener(T event)) {
+    listeners.add(listener);
+  }
+}
+
+/**
  * The data about a single analysis result that is stored in a [CacheEntry].
  */
 // TODO(brianwilkerson) Consider making this a generic class so that the value