Replace AnalysisServerListener with Stream.

pkg/analysis_server/lib/src/analysis_server.dart

 // Copyright (c) 2014, 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 analysis.server;
 
 import 'dart:async';
 import 'dart:collection';
 
 import 'package:analyzer/file_system/file_system.dart';
@@ skipping 190 matching lines @@
       new HashMap<AnalysisService, Set<String>>();
 
   /**
    * A table mapping [AnalysisContext]s to the completers that should be
    * completed when analysis of this context is finished.
    */
   Map<AnalysisContext, Completer<AnalysisDoneReason>> contextAnalysisDoneCompleters =
       new HashMap<AnalysisContext, Completer<AnalysisDoneReason>>();
 
   /**
-   * The listeners that are listening for lifecycle events from this server.
+   * The controller that is notified when analysis is started.
    */
-  List<AnalysisServerListener> listeners = <AnalysisServerListener>[];
+  StreamController<AnalysisContext> _onAnalysisStartedController;
+
+  /**
+   * The controller that is notified when analysis is complete.
+   */
+  StreamController _onAnalysisCompleteController;
 
   /**
    * True if any exceptions thrown by analysis should be propagated up the call
    * stack.
    */
   bool rethrowExceptions;
 
   /**
    * Initialize a newly created server to receive requests from and send
    * responses to the given [channel].
    *
    * If [rethrowExceptions] is true, then any exceptions thrown by analysis are
    * propagated up the call stack.  The default is true to allow analysis
    * exceptions to show up in unit tests, but it should be set to false when
    * running a full analysis server.
    */
   AnalysisServer(this.channel, this.resourceProvider,
       PackageMapProvider packageMapProvider, this.index, this.defaultSdk,
       {this.rethrowExceptions: true}) {
     searchEngine = createSearchEngine(index);
     operationQueue = new ServerOperationQueue(this);
     contextDirectoryManager =
         new ServerContextManager(this, resourceProvider, packageMapProvider);
     AnalysisEngine.instance.logger = new AnalysisLogger();
+    _onAnalysisStartedController = new StreamController.broadcast();
+    _onAnalysisCompleteController = new StreamController.broadcast();
     running = true;
     Notification notification = new ServerConnectedParams().toNotification();
     channel.sendNotification(notification);
     channel.listen(handleRequest, onDone: done, onError: error);
   }
 
   /**
    * Schedules execution of the given [ServerOperation].
    */
   void scheduleOperation(ServerOperation operation) {
     bool wasEmpty = operationQueue.isEmpty;
     addOperation(operation);
     if (wasEmpty) {
       _schedulePerformOperation();
     }
   }
 
   /**
    * Schedules analysis of the given context.
    */
   void schedulePerformAnalysisOperation(AnalysisContext context) {
-    _notifyAnalysisStarted(context);
+    _onAnalysisStartedController.add(context);
     scheduleOperation(new PerformAnalysisOperation(context, false));
   }
 
   /**
    * Send the given [notification] to the client.
    */
   void sendNotification(Notification notification) {
     channel.sendNotification(notification);
   }
 
@@ skipping 58 matching lines @@
     //
     // Update the defaults used to create new contexts.
     //
     AnalysisOptionsImpl options = contextDirectoryManager.defaultOptions;
     optionUpdaters.forEach((OptionUpdater optionUpdater) {
       optionUpdater(options);
     });
   }
 
   /**
-   * Add the given [listener] to the list of listeners that are listening for
-   * lifecycle events from this server.
+   * The stream that is notified when analysis of a context is started.
    */
-  void addAnalysisServerListener(AnalysisServerListener listener) {
-    if (!listeners.contains(listener)) {
-      listeners.add(listener);
-    }
+  Stream<AnalysisContext> get onAnalysisStarted {
+    return _onAnalysisStartedController.stream;
   }
 
   /**
+   * The stream that is notified when analysis is complete.
+   */
+  Stream get onAnalysisComplete => _onAnalysisCompleteController.stream;
+
+  /**
    * Adds the given [ServerOperation] to the queue, but does not schedule
    * operations execution.
    */
   void addOperation(ServerOperation operation) {
     operationQueue.add(operation);
   }
 
   /**
    * The socket from which requests are being read has been closed.
    */
@@ skipping 118 matching lines @@
             'Unexpected exception during analysis',
             new CaughtException(exception, stackTrace));
       }
       _sendServerErrorNotification(exception, stackTrace);
       shutdown();
     } finally {
       if (!operationQueue.isEmpty) {
         _schedulePerformOperation();
       } else {
         sendStatusNotification(null);
-        _notifyAnalysisComplete();
+        _onAnalysisCompleteController.add(null);
       }
     }
   }
 
   /**
    * Trigger reanalysis of all files from disk.
    */
   void reanalyze() {
     // Clear any operations that are pending.
     operationQueue.clear();
     // Instruct the contextDirectoryManager to rebuild all contexts from
     // scratch.
     contextDirectoryManager.refresh();
   }
 
   /**
-   * Remove the given [listener] from the list of listeners that are listening
-   * for lifecycle events from this server.
-   */
-  void removeAnalysisServerListener(AnalysisServerListener listener) {
-    listeners.remove(listener);
-  }
-
-  /**
    * Send status notification to the client. The `operation` is the operation
    * being performed or `null` if analysis is complete.
    */
   void sendStatusNotification(ServerOperation operation) {
     // Only send status when subscribed.
     if (!serverServices.contains(ServerService.STATUS)) {
       return;
     }
     // Only send status when it changes
     bool isAnalyzing = operation != null;
@@ skipping 376 matching lines @@
   }
 
   /**
    * Return `true` if all operations have been performed in this [AnalysisServer].
    */
   bool test_areOperationsFinished() {
     return operationQueue.isEmpty;
   }
 
   /**
-   * Notify all listeners that analysis of [context] is started.
-   */
-  void _notifyAnalysisStarted(AnalysisContext context) {
-    listeners.forEach((AnalysisServerListener listener) {
-      listener.analysisStarted(context);
-    });
-  }
-
-  /**
-   * Notify all listeners that analysis is complete.
-   */
-  void _notifyAnalysisComplete() {
-    listeners.forEach((AnalysisServerListener listener) {
-      listener.analysisComplete();
-    });
-  }
-
-  /**
    * Schedules [performOperation] exection.
    */
   void _schedulePerformOperation() {
     new Future(performOperation);
   }
 
   /**
    * Sends a fatal `server.error` notification.
    */
   void _sendServerErrorNotification(exception, stackTrace) {
@@ skipping 13 matching lines @@
     }
     // send the notification
     channel.sendNotification(
         new ServerErrorParams(
             true,
             exceptionString,
             stackTraceString).toNotification());
   }
 }
 
-/**
- * An object that is listening for lifecycle events from an analysis server.
- */
-abstract class AnalysisServerListener {
-  /**
-   * Analysis is complete.
-   */
-  void analysisComplete();
-
-  /**
-   * Analysis is started.
-   */
-  void analysisStarted(AnalysisContext context);
-}
-
 typedef void OptionUpdater(AnalysisOptionsImpl options);