Add a RunnerSuite.channel() call that can connect to a test worker

Partially addresses flutter#711

Author: nex3

lib/src/runner/load_suite.dart

@@ -5,6 +5,7 @@
 import 'dart:async';
 
 import 'package:stack_trace/stack_trace.dart';
+import 'package:stream_channel/stream_channel.dart';
 
 import '../../test.dart';
 import '../backend/group.dart';
@@ -193,5 +194,8 @@ class LoadSuite extends Suite implements RunnerSuite {
     return new LoadSuite._filtered(this, filtered);
   }
 
+  StreamChannel channel(String name) =>
+      throw new UnsupportedError("LoadSuite.channel() is not supported.");
+
   Future close() async {}
 }

lib/src/runner/plugin/platform_helpers.dart

@@ -48,6 +48,7 @@ Future<RunnerSuiteController> deserializeSuite(
   var suiteChannel = new MultiChannel(channel.transform(disconnector));
 
   suiteChannel.sink.add({
+    'type': 'initial',
     'platform': platform.serialize(),
     'metadata': suiteConfig.metadata.serialize(),
     'os': (platform == TestPlatform.vm || platform == TestPlatform.nodeJS)
@@ -112,7 +113,7 @@ Future<RunnerSuiteController> deserializeSuite(
       });
 
   return new RunnerSuiteController(
-      environment, suiteConfig, await completer.future,
+      environment, suiteConfig, suiteChannel, await completer.future,
       path: path,
       platform: platform,
       os: currentOS,

lib/src/runner/plugin/remote_platform_helpers.dart

@@ -5,6 +5,7 @@
 import 'package:stream_channel/stream_channel.dart';
 
 import '../remote_listener.dart';
+import '../suite_channel_manager.dart';
 
 /// Returns a channel that will emit a serialized representation of the tests
 /// defined in [getMain].
@@ -23,3 +24,20 @@ import '../remote_listener.dart';
 /// where they'll be visible in the development console.
 StreamChannel serializeSuite(Function getMain(), {bool hidePrints: true}) =>
     RemoteListener.start(getMain, hidePrints: hidePrints);
+
+/// Returns a channel that communicates with a plugin in the test runner.
+///
+/// This connects to a channel created by code in the test runner calling
+/// `RunnerSuite.channel()` with the same name. It can be used used to send and
+/// receive any JSON-serializable object.
+///
+/// Throws a [StateError] if [name] has already been used for a channel.
+StreamChannel suiteChannel(String name) {
+  var manager = SuiteChannelManager.current;
+  if (manager == null) {
+    throw new StateError(
+        'suiteChannel() may only be called within a test worker.');
+  }
+
+  return manager.connectOut(name);
+}

lib/src/runner/remote_listener.dart

@@ -4,6 +4,7 @@
 
 import 'dart:async';
 
+import 'package:async/async.dart';
 import 'package:stream_channel/stream_channel.dart';
 import 'package:term_glyph/term_glyph.dart' as glyph;
 
@@ -20,6 +21,7 @@ import '../frontend/test_chain.dart';
 import '../util/remote_exception.dart';
 import '../util/stack_trace_mapper.dart';
 import '../utils.dart';
+import 'suite_channel_manager.dart';
 
 class RemoteListener {
   /// The test suite to run.
@@ -52,66 +54,79 @@ class RemoteListener {
     var verboseChain = true;
 
     var printZone = hidePrints ? null : Zone.current;
-    runZoned(() async {
-      var main;
-      try {
-        main = getMain();
-      } on NoSuchMethodError catch (_) {
-        _sendLoadException(channel, "No top-level main() function defined.");
-        return;
-      } catch (error, stackTrace) {
-        _sendError(channel, error, stackTrace, verboseChain);
-        return;
-      }
-
-      if (main is! Function) {
-        _sendLoadException(channel, "Top-level main getter is not a function.");
-        return;
-      } else if (main is! AsyncFunction) {
-        _sendLoadException(
-            channel, "Top-level main() function takes arguments.");
-        return;
-      }
-
-      var message = await channel.stream.first;
-
-      if (message['asciiGlyphs'] ?? false) glyph.ascii = true;
-      var metadata = new Metadata.deserialize(message['metadata']);
-      verboseChain = metadata.verboseTrace;
-      var declarer = new Declarer(
-          metadata: metadata,
-          platformVariables: new Set.from(message['platformVariables']),
-          collectTraces: message['collectTraces'],
-          noRetry: message['noRetry']);
-
-      configureTestChaining(
-          mapper: StackTraceMapper.deserialize(message['stackTraceMapper']),
-          exceptPackages: _deserializeSet(message['foldTraceExcept']),
-          onlyPackages: _deserializeSet(message['foldTraceOnly']));
-
-      await declarer.declare(main);
-
-      var suite = new Suite(declarer.build(),
-          platform: new TestPlatform.deserialize(message['platform']),
-          os: message['os'] == null
-              ? null
-              : OperatingSystem.find(message['os']),
-          path: message['path']);
-
-      runZoned(() {
-        Invoker.guard(
-            () => new RemoteListener._(suite, printZone)._listen(channel));
-      },
-          // Make the declarer visible to running tests so that they'll throw
-          // useful errors when calling `test()` and `group()` within a test,
-          // and so they can add to the declarer's `tearDownAll()` list.
-          zoneValues: {#test.declarer: declarer});
-    }, onError: (error, stackTrace) {
-      _sendError(channel, error, stackTrace, verboseChain);
-    }, zoneSpecification: new ZoneSpecification(print: (_, __, ___, line) {
+    var spec = new ZoneSpecification(print: (_, __, ___, line) {
       if (printZone != null) printZone.print(line);
       channel.sink.add({"type": "print", "line": line});
-    }));
+    });
+
+    runZoned(() {
+      new SuiteChannelManager().asCurrent(() async {
+        var main;
+        try {
+          main = getMain();
+        } on NoSuchMethodError catch (_) {
+          _sendLoadException(channel, "No top-level main() function defined.");
+          return;
+        } catch (error, stackTrace) {
+          _sendError(channel, error, stackTrace, verboseChain);
+          return;
+        }
+
+        if (main is! Function) {
+          _sendLoadException(
+              channel, "Top-level main getter is not a function.");
+          return;
+        } else if (main is! AsyncFunction) {
+          _sendLoadException(
+              channel, "Top-level main() function takes arguments.");
+          return;
+        }
+
+        var queue = new StreamQueue(channel.stream);
+        var message = await queue.next;
+        assert(message['type'] == 'initial');
+
+        queue.rest.listen((message) {
+          assert(message["type"] == "suiteChannel");
+          SuiteChannelManager.current.connectIn(
+              message['name'], channel.virtualChannel(message['id']));
+        });
+
+        if (message['asciiGlyphs'] ?? false) glyph.ascii = true;
+        var metadata = new Metadata.deserialize(message['metadata']);
+        verboseChain = metadata.verboseTrace;
+        var declarer = new Declarer(
+            metadata: metadata,
+            platformVariables: new Set.from(message['platformVariables']),
+            collectTraces: message['collectTraces'],
+            noRetry: message['noRetry']);
+
+        configureTestChaining(
+            mapper: StackTraceMapper.deserialize(message['stackTraceMapper']),
+            exceptPackages: _deserializeSet(message['foldTraceExcept']),
+            onlyPackages: _deserializeSet(message['foldTraceOnly']));
+
+        await declarer.declare(main);
+
+        var suite = new Suite(declarer.build(),
+            platform: new TestPlatform.deserialize(message['platform']),
+            os: message['os'] == null
+                ? null
+                : OperatingSystem.find(message['os']),
+            path: message['path']);
+
+        runZoned(() {
+          Invoker.guard(
+              () => new RemoteListener._(suite, printZone)._listen(channel));
+        },
+            // Make the declarer visible to running tests so that they'll throw
+            // useful errors when calling `test()` and `group()` within a test,
+            // and so they can add to the declarer's `tearDownAll()` list.
+            zoneValues: {#test.declarer: declarer});
+      });
+    }, onError: (error, stackTrace) {
+      _sendError(channel, error, stackTrace, verboseChain);
+    }, zoneSpecification: spec);
 
     return controller.foreign;
   }

lib/src/runner/runner_suite.dart

@@ -5,6 +5,7 @@
 import 'dart:async';
 
 import 'package:async/async.dart';
+import 'package:stream_channel/stream_channel.dart';
 
 import '../backend/group.dart';
 import '../backend/operating_system.dart';
@@ -45,15 +46,22 @@ class RunnerSuite extends Suite {
   /// The event is `true` when debugging starts and `false` when it ends.
   Stream<bool> get onDebugging => _controller._onDebuggingController.stream;
 
+  /// Returns a channel that communicates with the remote suite.
+  ///
+  /// This connects to a channel created by code in the test worker calling
+  /// `suiteChannel()` from `remote_platform_helpers.dart` with the same name.
+  /// It can be used used to send and receive any JSON-serializable object.
+  StreamChannel channel(String name) => _controller.channel(name);
+
   /// A shortcut constructor for creating a [RunnerSuite] that never goes into
-  /// debugging mode.
+  /// debugging mode and doesn't support suite channels.
   factory RunnerSuite(
       Environment environment, SuiteConfiguration config, Group group,
       {String path,
       TestPlatform platform,
       OperatingSystem os,
       AsyncFunction onClose}) {
-    var controller = new RunnerSuiteController(environment, config, group,
+    var controller = new RunnerSuiteController(environment, config, null, group,
         path: path, platform: platform, os: os, onClose: onClose);
     return controller.suite;
   }
@@ -84,6 +92,9 @@ class RunnerSuiteController {
   /// The configuration for this suite.
   final SuiteConfiguration _config;
 
+  /// A channel that communicates with the remote suite.
+  final MultiChannel _suiteChannel;
+
   /// The function to call when the suite is closed.
   final AsyncFunction _onClose;
 
@@ -93,7 +104,11 @@ class RunnerSuiteController {
   /// The controller for [suite.onDebugging].
   final _onDebuggingController = new StreamController<bool>.broadcast();
 
-  RunnerSuiteController(this._environment, this._config, Group group,
+  /// The channel names that have already been used.
+  final _channelNames = new Set<String>();
+
+  RunnerSuiteController(
+      this._environment, this._config, this._suiteChannel, Group group,
       {String path,
       TestPlatform platform,
       OperatingSystem os,
@@ -112,6 +127,27 @@ class RunnerSuiteController {
     _onDebuggingController.add(debugging);
   }
 
+  /// Returns a channel that communicates with the remote suite.
+  ///
+  /// This connects to a channel created by code in the test worker calling
+  /// `suiteChannel()` from `remote_platform_helpers.dart` with the same name.
+  /// It can be used used to send and receive any JSON-serializable object.
+  ///
+  /// This is exposed on the [RunnerSuiteController] so that runner plugins can
+  /// communicate with the workers they spawn before the associated [suite] is
+  /// fully loaded.
+  StreamChannel channel(String name) {
+    if (!_channelNames.add(name)) {
+      throw new StateError(
+          'Duplicate RunnerSuite.channel() connection "$name".');
+    }
+
+    var channel = _suiteChannel.virtualChannel();
+    _suiteChannel.sink
+        .add({"type": "suiteChannel", "name": name, "id": channel.id});
+    return channel;
+  }
+
   /// The backing function for [suite.close].
   Future _close() => _closeMemo.runOnce(() async {
         _onDebuggingController.close();

lib/src/runner/suite_channel_manager.dart

@@ -0,0 +1,61 @@
+// Copyright (c) 2018, 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 'package:stream_channel/stream_channel.dart';
+
+/// The key used to look up [SuiteChannelManager.current] in a zone.
+final _currentKey = new Object();
+
+/// A class that connects incoming and outgoing channels with the same names.
+class SuiteChannelManager {
+  /// Connections from the test runner that have yet to connect to corresponding
+  /// calls to [suiteChannel] within this worker.
+  final _incomingConnections = <String, StreamChannel>{};
+
+  /// Connections from calls to [suiteChannel] that have yet to connect to
+  /// corresponding connections from the test runner.
+  final _outgoingConnections = <String, StreamChannelCompleter>{};
+
+  /// The channel names that have already been used.
+  final _names = new Set<String>();
+
+  /// Returns the current manager, or `null` if this isn't called within a call
+  /// to [asCurrent].
+  static SuiteChannelManager get current =>
+      Zone.current[_currentKey] as SuiteChannelManager;
+
+  /// Runs [body] with [this] as [SuiteChannelManager.current].
+  ///
+  /// This is zone-scoped, so [this] will be the current configuration in any
+  /// asynchronous callbacks transitively created by [body].
+  T asCurrent<T>(T body()) => runZoned(body, zoneValues: {_currentKey: this});
+
+  /// Creates a connection to the test runnner's channel with the given [name].
+  StreamChannel connectOut(String name) {
+    if (_incomingConnections.containsKey(name)) {
+      return _incomingConnections[name];
+    } else if (_names.contains(name)) {
+      throw new StateError('Duplicate suiteChannel() connection "$name".');
+    } else {
+      _names.add(name);
+      var completer = new StreamChannelCompleter();
+      _outgoingConnections[name] = completer;
+      return completer.channel;
+    }
+  }
+
+  /// Connects [channel] to this worker's channel with the given [name].
+  void connectIn(String name, StreamChannel channel) {
+    if (_outgoingConnections.containsKey(name)) {
+      _outgoingConnections.remove(name).setChannel(channel);
+    } else if (_incomingConnections.containsKey(name)) {
+      throw new StateError(
+          'Duplicate RunnerSuite.channel() connection "$name".');
+    } else {
+      _incomingConnections[name] = channel;
+    }
+  }
+}