Code generate integration test streams for notifications.

pkg/analysis_server/test/integration/integration_tests.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 test.integration.analysis;
 
 import 'dart:async';
 import 'dart:collection';
 import 'dart:convert';
 import 'dart:io';
@@ skipping 42 matching lines @@
    * Data associated with the "server.connected" notification that was received
    * when the server started up.
    */
   var serverConnectedParams;
 
   /**
    * True if we are currently subscribed to [SERVER_STATUS] updates.
    */
   bool _subscribedToServerStatus = false;
 
+  AbstractAnalysisServerIntegrationTest() {
+    initializeInttestMixin();
+  }
+
   /**
    * Write a source file with the given absolute [pathname] and [contents].
    *
    * If the file didn't previously exist, it is created.  If it did, it is
    * overwritten.
    *
    * Parent directories are created as necessary.
    */
   void writeFile(String pathname, String contents) {
     new Directory(dirname(pathname)).createSync(recursive: true);
@@ skipping 32 matching lines @@
    * received after this function call.  So it is safe to use this getter
    * multiple times in one test; each time it is used it will wait afresh for
    * analysis to finish.
    */
   Future get analysisFinished {
     Completer completer = new Completer();
     StreamSubscription subscription;
     // This will only work if the caller has already subscribed to
     // SERVER_STATUS (e.g. using sendServerSetSubscriptions(['STATUS']))
     expect(_subscribedToServerStatus, isTrue);
-    subscription = server.onNotification(SERVER_STATUS).listen((params) {
+    subscription = onServerStatus.listen((params) {
       bool analysisComplete = false;
       try {
         analysisComplete = !params['analysis']['analyzing'];
       } catch (_) {
         // Status message was mal-formed or missing optional parameters.  That's
         // fine, since we'll detect a mal-formed status message below.
       }
       if (analysisComplete) {
         completer.complete(params);
         subscription.cancel();
@@ skipping 19 matching lines @@
         checkTypes);
   }
 
   /**
    * The server is automatically started before every test, and a temporary
    * [sourceDirectory] is created.
    */
   Future setUp() {
     sourceDirectory = Directory.systemTemp.createTempSync('analysisServer');
 
-    server.onNotification(ANALYSIS_ERRORS).listen((params) {
-      expect(params, isMap);
-      expect(params['file'], isString);
+    onAnalysisErrors.listen((params) {
       currentAnalysisErrors[params['file']] = params['errors'];
     });
     Completer serverConnected = new Completer();
-    server.onNotification(SERVER_CONNECTED).listen((_) {
+    onServerConnected.listen((_) {
       expect(serverConnected.isCompleted, isFalse);
       serverConnected.complete();
     });
-    return server.start().then((params) {
+    return server.start(dispatchNotification).then((params) {
       serverConnectedParams = params;
       server.exitCode.then((_) {
         skipShutdown = true;
       });
       return serverConnected.future;
     });
   }
 
   /**
    * After every test, the server is stopped and [sourceDirectory] is deleted.
@@ skipping 309 matching lines @@
 
   @override
   Description describe(Description description) => description.add('Map from '
       ).addDescriptionOf(keyMatcher).add(' to ').addDescriptionOf(valueMatcher);
 }
 
 Matcher isMapOf(Matcher keyMatcher, Matcher valueMatcher) => new _MapOf(
     keyMatcher, valueMatcher);
 
 /**
+ * Type of callbacks used to process notifications.
+ */
+typedef void NotificationProcessor(String event, params);
+
+/**
  * Instances of the class [Server] manage a connection to a server process, and
  * facilitate communication to and from the server.
  */
 class Server {
   /**
    * Server process object, or null if server hasn't been started yet.
    */
   Process _process = null;
 
   /**
    * Commands that have been sent to the server but not yet acknowledged, and
    * the [Completer] objects which should be completed when acknowledgement is
    * received.
    */
   final HashMap<String, Completer> _pendingCommands = <String, Completer> {};
 
   /**
    * Number which should be used to compute the 'id' to send in the next command
    * sent to the server.
    */
   int _nextId = 0;
 
   /**
-   * [StreamController]s to which notifications should be sent, organized by
-   * event type.
+   * [StreamController] to which notifications will be sent.
    */
-  final HashMap<String, StreamController> _notificationControllers =
-      new HashMap<String, StreamController>();
-
-  /**
-   * [Stream]s associated with the controllers in [_notificationControllers],
-   * but converted to broadcast streams.
-   */
-  final HashMap<String, Stream> _notificationStreams = new HashMap<String,
-      Stream>();
+  final StreamController _notifications = new StreamController();
 
   /**
    * Messages which have been exchanged with the server; we buffer these
    * up until the test finishes, so that they can be examined in the debugger
    * or printed out in response to a call to [debugStdio].
    */
   final List<String> _recordedStdio = <String>[];
 
   /**
    * True if we are currently printing out messages exchanged with the server.
    */
   bool _debuggingStdio = false;
 
   /**
    * True if we've received bad data from the server, and we are aborting the
    * test.
    */
   bool _receivedBadDataFromServer = false;
 
   /**
    * Stopwatch that we use to generate timing information for debug output.
    */
   Stopwatch _time = new Stopwatch();
 
   /**
-   * Get a stream which will receive notifications of the given event type.
-   * The values delivered to the stream will be the contents of the 'params'
-   * field of the notification message.
-   */
-  Stream onNotification(String event) {
-    Stream notificationStream = _notificationStreams[event];
-    if (notificationStream == null) {
-      StreamController notificationController = new StreamController();
-      _notificationControllers[event] = notificationController;
-      notificationStream = notificationController.stream.asBroadcastStream();
-      _notificationStreams[event] = notificationStream;
-    }
-    return notificationStream;
-  }
-
-  /**
    * Start the server.  If [debugServer] is true, the server will be started
    * with "--debug", allowing a debugger to be attached.
    */
-  Future start({bool debugServer: false}) {
+  Future start(NotificationProcessor notificationProcessor, {bool debugServer:
+      false}) {
     if (_process != null) {
       throw new Exception('Process already started');
     }
     _time.start();
     // TODO(paulberry): move the logic for finding the script, the dart
     // executable, and the package root into a shell script.
     String dartBinary = Platform.executable;
     String scriptDir = dirname(Platform.script.toFilePath(windows:
         Platform.isWindows));
     String serverPath = normalize(join(scriptDir, '..', '..', 'bin',
@@ skipping 40 matching lines @@
           }
           // Check that the message is well-formed.  We do this after calling
           // completer.complete() or completer.completeError() so that we don't
           // stall the test in the event of an error.
           expect(message, isResponse);
         } else {
           // Message is a notification.  It should have an event and possibly
           // params.
           expect(messageAsMap, contains('event'));
           expect(messageAsMap['event'], isString);
-          String event = messageAsMap['event'];
-          StreamController notificationController =
-              _notificationControllers[event];
-          if (notificationController != null) {
-            notificationController.add(messageAsMap['params']);
-          }
+          notificationProcessor(messageAsMap['event'], messageAsMap['params']);
           // Check that the message is well-formed.  We do this after calling
           // notificationController.add() so that we don't stall the test in the
           // event of an error.
           expect(message, isNotification);
         }
       });
       process.stderr.transform((new Utf8Codec()).decoder).transform(
           new LineSplitter()).listen((String line) {
         String trimmedLine = line.trim();
         _recordStdio('ERR:  $trimmedLine');
@@ skipping 88 matching lines @@
    */
   void _recordStdio(String line) {
     double elapsedTime = _time.elapsedTicks / _time.frequency;
     line = "$elapsedTime: $line";
     if (_debuggingStdio) {
       print(line);
     }
     _recordedStdio.add(line);
   }
 }