Replace Stream with monitor.

pkg/analyzer/lib/src/dart/analysis/driver.dart

 // Copyright (c) 2016, 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.
 
 import 'dart:async';
 import 'dart:collection';
 import 'dart:convert';
 
 import 'package:analyzer/dart/ast/ast.dart';
 import 'package:analyzer/dart/ast/token.dart';
@@ skipping 140 matching lines @@
    * It maps file paths to the unlinked API signatures.
    */
   final _fileApiSignatureMap = <String, String>{};
 
   /**
    * Mapping from library URIs to the dependency signature of the library.
    */
   final _dependencySignatureMap = <Uri, String>{};
 
   /**
-   * TODO(scheglov) document and improve
+   * The monitor that is signalled when there is work to do.
    */
-  final _hasWorkStreamController = new StreamController<String>();
+  final _Monitor _hasWork = new _Monitor();
 
   AnalysisDriver(this._logger, this._resourceProvider, this._byteStore,
       this._contentCache, this._sourceFactory, this._analysisOptions) {
     _sdkBundle = _sourceFactory.dartSdk.getLinkedBundle();
   }
 
   /**
    * Set the list of files that the driver should try to analyze sooner.
    *
    * Every path in the list must be absolute and normalized.
    *
    * The driver will produce the results through the [results] stream. The
    * exact order in which results are produced is not defined, neither
    * between priority files, nor between priority and non-priority files.
    */
   void set priorityFiles(List<String> priorityPaths) {
     // TODO(scheglov) implement
   }
 
   /**
    * Return the [Stream] that produces [AnalysisResult]s for added files.
    *
    * Analysis starts when the client starts listening to the stream, and stops
    * when the client cancels the subscription.

[Paul Berry, 2016/10/31 16:40:47]

It looks to me like it's not safe to call `.results` more than once on a given
AnalysisDriver.  Is that correct?  If so, could you please add a note to this
effect in the doc comment?

[scheglov, 2016/10/31 17:45:43]

Yes, this is a normal Stream, not broadcast.
So, only one subscriber over the lifetime.

    *
    * When the client starts listening, the analysis state transitions to
    * "analyzing" and an analysis result is produced for every added file prior
    * to the next time the analysis state transitions to "idle".
    *
    * At least one analysis result is produced for every file passed to
    * [addFile] or [changeFile] prior to the next time the analysis state
    * transitions to "idle". Analysis results for other files are produced
    * only if the changes affect analysis results of other files.
    *
    * More than one result might be produced for the same file, even if the
    * client does not change the state of the files.
    *
    * Results might be produced even for files that have never been added
    * using [addFile], for example when [getResult] was called for a file.
    */
   Stream<AnalysisResult> get results async* {
     try {
+      PerformanceLogSection analysisSection = null;
       while (true) {
         // TODO(scheglov) implement state transitioning
-        await for (String why in _hasWorkStreamController.stream) {
-          _verifyUnlinkedSignatureOfChangedFiles();
+        await _hasWork.signal;
 
-          // Analyze the first file in the general queue.
-          if (_filesToAnalyze.isNotEmpty) {
-            PerformanceLogSection analysisSection =
-                _logger.enter('Analyze ${_filesToAnalyze.length} files');
-            try {
-              // TODO(scheglov) The loop is strange for now.
-              while (_filesToAnalyze.isNotEmpty) {
-                String path = _filesToAnalyze.first;
-                _filesToAnalyze.remove(path);
-                _File file = _fileForPath(path);
-                AnalysisResult result = _computeAnalysisResult(file);
-                yield result;
-              }
-            } finally {
-              analysisSection.exit();
-            }
-          }
+        if (analysisSection == null) {
+          analysisSection = _logger.enter('Analyzing');
         }
-        // TODO(scheglov) implement
+
+        // TODO(scheglov) verify one file at a time
+        _verifyUnlinkedSignatureOfChangedFiles();
+
+        // Analyze the first file in the general queue.
+        if (_filesToAnalyze.isNotEmpty) {
+          String path = _filesToAnalyze.first;
+          _filesToAnalyze.remove(path);
+          _File file = _fileForPath(path);
+          AnalysisResult result = _computeAnalysisResult(file);
+          yield result;
+        }
+
+        // If there is work to do, notify the monitor.
+        if (_filesToAnalyze.isNotEmpty) {
+          _hasWork.notify();
+        } else {
+          analysisSection.exit();
+          analysisSection = null;
+        }
       }
+      // TODO(scheglov) implement
     } finally {
       print('The stream was cancelled.');
     }
   }
 
   /**
    * Add the file with the given [path] to the set of files to analyze.
    *
    * The [path] must be absolute and normalized.
    *
    * The results of analysis are eventually produced by the [results] stream.
    */
   void addFile(String path) {
     _explicitFiles.add(path);
     _filesToAnalyze.add(path);
-    _hasWorkStreamController.add('do it!');
+    _hasWork.notify();
   }
 
   /**
    * The file with the given [path] might have changed - updated, added or
    * removed. Or not, we don't know. Or it might have, but then changed back.
    *
    * The [path] must be absolute and normalized.
    *
    * The [path] can be any file - explicitly or implicitly analyzed, or neither.
    *
    * Causes the analysis state to transition to "analyzing" (if it is not in
    * that state already). Schedules the file contents for [path] to be read
    * into the current file state prior to the next time the analysis state
    * transitions to "idle".
    *
    * Invocation of this method will not prevent a [Future] returned from
    * [getResult] from completing with a result, but the result is not
    * guaranteed to be consistent with the new current file state after this
    * [changeFile] invocation.
    */
   void changeFile(String path) {
     _filesToVerifyUnlinkedSignature.add(path);
     _filesToAnalyze.add(path);
-    _hasWorkStreamController.add('do it!');
+    _hasWork.notify();
   }
 
   /**
    * Return the [Future] that completes with a [AnalysisResult] for the file
    * with the given [path].
    *
    * The [path] must be absolute and normalized.
    *
    * The [path] can be any file - explicitly or implicitly analyzed, or neither.
    *
    * Causes the analysis state to transition to "analyzing" (if it is not in
    * that state already), the driver will read the file and produce the analysis
    * result for it, which is consistent with the current file state (including
    * the new state of the file), prior to the next time the analysis state
    * transitions to "idle".
    */
   Future<AnalysisResult> getResult(String path) {
     var completer = new Completer<AnalysisResult>();
     _requestedFiles
         .putIfAbsent(path, () => <Completer<AnalysisResult>>[])
         .add(completer);
-    _hasWorkStreamController.add(path);
+    _hasWork.notify();
     return completer.future;
   }
 
   /**
    * Remove the file with the given [path] from the list of files to analyze.
    *
    * The [path] must be absolute and normalized.
    *
    * The results of analysis of the file might still be produced by the
    * [results] stream. The driver will try to stop producing these results,
@@ skipping 632 matching lines @@
 
       appendDependencies(this);
     }
   }
 
   @override
   String toString() => uri.toString();
 }
 
 /**
+ * [_Monitor] can be used to wait for a signal.
+ *
+ * Signals are not queued, the client will receive exactly one signal
+ * regardless of the number of [notify] invocations. The [signal] is reset
+ * after completion and will not complete until [notify] is called next time.
+ */
+class _Monitor {
+  Completer<Null> _completer = new Completer<Null>();
+
+  /**
+   * Return a [Future] that completes when [notify] is called at least once.
+   */
+  Future<Null> get signal async {
+    await _completer.future;
+    _completer = new Completer<Null>();
+  }
+
+  /**
+   * Complete the [signal] future if it is not completed yet. It is safe to
+   * call this method multiple times, but the [signal] will complete only once.
+   */
+  void notify() {
+    if (!_completer.isCompleted) {
+      _completer.complete(true);
+    }
+  }
+}
+
+/**
  * TODO(scheglov) document
  */
 class _ReferencedUris {
   bool isLibrary = true;
   final List<String> imported = <String>[];
   final List<String> exported = <String>[];
   final List<String> parted = <String>[];
 
   factory _ReferencedUris(UnlinkedUnit unit) {
     _ReferencedUris referenced = new _ReferencedUris._();
     referenced.parted.addAll(unit.publicNamespace.parts);
     for (UnlinkedImport import in unit.imports) {
       if (!import.isImplicit) {
         referenced.imported.add(import.uri);
       }
     }
     for (UnlinkedExportPublic export in unit.publicNamespace.exports) {
       referenced.exported.add(export.uri);
     }
     return referenced;
   }
 
   _ReferencedUris._();
 }