Fix for leaking AnalysisCache and AnalysisContext instances.

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 23 matching lines @@
    * An array containing the partitions of which this cache is comprised.
    */
   final List<CachePartition> _partitions;
 
   /**
    * The [StreamController] reporting [InvalidatedResult]s.
    */
   final ReentrantSynchronousStream<InvalidatedResult> onResultInvalidated =
       new ReentrantSynchronousStream<InvalidatedResult>();
 
+  final List<
+      ReentrantSynchronousStreamSubscription> onResultInvalidatedPartitionSubscriptions = <
+      ReentrantSynchronousStreamSubscription>[];
+
   /**
-   * Initialize a newly created cache to have the given [partitions]. The
+   * 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) {
+      ReentrantSynchronousStreamSubscription<InvalidatedResult> subscription =
+          partition.onResultInvalidated.listen((InvalidatedResult event) {
         onResultInvalidated.add(event);
       });
+      onResultInvalidatedPartitionSubscriptions.add(subscription);
     }
   }
 
+  /**
+   * Return an iterator returning all of the [Source] targets.
+   */
+  Iterable<Source> get sources {
+    return _partitions
+        .map((CachePartition partition) => partition._sources)
+        .expand((Iterable<Source> sources) => sources);
+  }
+
   // 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];
 //      data[i] = new AnalysisContextStatisticsImpl_PartitionDataImpl(
 //          partition.astSize,
 //          partition.map.length);
 //    }
 //    return data;
 //  }
 
   /**
-   * Return an iterator returning all of the [Source] targets.
+   * Free any allocated resources and references.
    */
-  Iterable<Source> get sources {
-    return _partitions
-        .map((CachePartition partition) => partition._sources)
-        .expand((Iterable<Source> sources) => sources);
+  void dispose() {
+    for (ReentrantSynchronousStreamSubscription subscription
+        in onResultInvalidatedPartitionSubscriptions) {
+      subscription.cancel();
+    }
   }
 
   /**
    * Return the entry associated with the given [target].
    */
   CacheEntry get(AnalysisTarget target) {
     int count = _partitions.length;
     for (int i = 0; i < count; i++) {
       CachePartition partition = _partitions[i];
       if (partition.isResponsibleFor(target)) {
@@ skipping 943 matching lines @@
    * Check whether this delta affects the result described by the given
    * [descriptor] and [target].
    */
   DeltaResult validate(InternalAnalysisContext context, AnalysisTarget target,
       ResultDescriptor descriptor) {
     return DeltaResult.INVALIDATE;
   }
 }
 
 /**
- * The possible results of validating analysis results againt a [Delta].
+ * The possible results of validating analysis results against a [Delta].
  */
 enum DeltaResult {
   /**
    * Invalidate this result and continue visiting dependent results
    * with this [Delta].
    */
   INVALIDATE,
 
   /**
    * Invalidate this result and stop using this [Delta], so unconditionally
@@ skipping 55 matching lines @@
     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)) {
+  ReentrantSynchronousStreamSubscription<T> listen(void listener(T event)) {
     listeners.add(listener);
+    return new ReentrantSynchronousStreamSubscription<T>(this, listener);
   }
 }
 
+/**
+ * A subscription on events from a [ReentrantSynchronousStream].
+ */
+class ReentrantSynchronousStreamSubscription<T> {
+  final ReentrantSynchronousStream<T> _stream;
+  final Function _listener;
+
+  ReentrantSynchronousStreamSubscription(this._stream, this._listener);
+
+  /**
+   * Cancels this subscription.
+   * It will no longer receive events.
+   */
+  void cancel() {
+    _stream.listeners.remove(_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 90 matching lines @@
   void resultAccessed(TargetedResult result) {}
 
   @override
   List<TargetedResult> resultStored(TargetedResult newResult, newValue) {
     return TargetedResult.EMPTY_LIST;
   }
 
   @override
   void targetRemoved(AnalysisTarget target) {}
 }