Add ReentrantSynchronousStream and use it for cache invalidation events.

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

 // Copyright (c) 2015, 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 analyzer.src.context.cache;
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'package:analyzer/src/generated/engine.dart'
@@ skipping 20 matching lines @@
   static bool _TRACE_CHANGES = false;
 
   /**
    * An array containing the partitions of which this cache is comprised.
    */
   final List<CachePartition> _partitions;
 
   /**
    * 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
    * partitions will be searched in the order in which they appear in the array,
    * so the most specific partition (usually an [SdkCachePartition]) should be
    * first and the most general (usually a [UniversalCachePartition]) last.
    */
   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.
 //   */
 //  List<AnalysisContextStatistics_PartitionData> get partitionData {
 //    int count = _partitions.length;
 //    List<AnalysisContextStatistics_PartitionData> data =
 //        new List<AnalysisContextStatistics_PartitionData>(count);
 //    for (int i = 0; i < count; i++) {
 //      CachePartition partition = _partitions[i];
@@ skipping 478 matching lines @@
       if (entry != null) {
         entry._invalidate(dependentResult.result, delta);
       }
     }
     // If empty, remove the entry altogether.
     if (_resultMap.isEmpty) {
       _partition._targetMap.remove(target);
       _partition._removeIfSource(target);
     }
     // Notify controller.
-    _partition._onResultInvalidated
-        .add(new InvalidatedResult(this, descriptor));
+    _partition.onResultInvalidated
+        .add(new InvalidatedResult(this, descriptor, thisData.value));
   }
 
   /**
    * Invalidates all the results of this entry, with propagation.
    */
   void _invalidateAll() {
     List<ResultDescriptor> results = _resultMap.keys.toList();
     for (ResultDescriptor result in results) {
       _invalidate(result, null);
     }
@@ skipping 226 matching lines @@
 
   /**
    * A table mapping caching policies to the cache flush managers.
    */
   final HashMap<ResultCachingPolicy, CacheFlushManager> _flushManagerMap =
       new HashMap<ResultCachingPolicy, CacheFlushManager>();
 
   /**
    * 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
    * known about those targets.
    */
   HashMap<AnalysisTarget, CacheEntry> _targetMap =
       new HashMap<AnalysisTarget, CacheEntry>();
 
   /**
    * A set of the [Source] targets.
@@ skipping 14 matching lines @@
   /**
    * Return a table mapping the targets known to the context to the information
    * known about the target.
    *
    * <b>Note:</b> This method is only visible for use by [AnalysisCache] and
    * should not be used for any other purpose.
    */
   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() {
     for (CacheEntry entry in _targetMap.values) {
       entry.dispose();
     }
     _targetMap.clear();
   }
 
   /**
@@ skipping 173 matching lines @@
   /**
    * The target in which the result was invalidated.
    */
   final CacheEntry entry;
 
   /**
    * The descriptor of the result which was invalidated.
    */
   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
 // can be typed.
 class ResultData {
   /**
    * The [ResultDescriptor] this result is for.
    */
   final ResultDescriptor descriptor;
 
@@ skipping 82 matching lines @@
   void resultAccessed(TargetedResult result) {}
 
   @override
   List<TargetedResult> resultStored(TargetedResult newResult, newValue) {
     return TargetedResult.EMPTY_LIST;
   }
 
   @override
   void targetRemoved(AnalysisTarget target) {}
 }