Add ability to mock VMService's WebSocket connection (#8145)
diff --git a/packages/flutter_tools/lib/src/commands/trace.dart b/packages/flutter_tools/lib/src/commands/trace.dart index 75edcc2..43b59c3 100644 --- a/packages/flutter_tools/lib/src/commands/trace.dart +++ b/packages/flutter_tools/lib/src/commands/trace.dart
@@ -60,7 +60,7 @@ Tracing tracing; try { - tracing = await Tracing.connect(observatoryUri); + tracing = Tracing.connect(observatoryUri); } catch (error) { throwToolExit('Error connecting to observatory: $error'); } @@ -102,8 +102,9 @@ class Tracing { Tracing(this.vmService); - static Future<Tracing> connect(Uri uri) { - return VMService.connect(uri).then((VMService observatory) => new Tracing(observatory)); + static Tracing connect(Uri uri) { + VMService observatory = VMService.connect(uri); + return new Tracing(observatory); } final VMService vmService;
diff --git a/packages/flutter_tools/lib/src/resident_runner.dart b/packages/flutter_tools/lib/src/resident_runner.dart index 235ccdc..5d81a1c 100644 --- a/packages/flutter_tools/lib/src/resident_runner.dart +++ b/packages/flutter_tools/lib/src/resident_runner.dart
@@ -194,7 +194,7 @@ if (!debuggingOptions.debuggingEnabled) { return new Future<Null>.error('Error the service protocol is not enabled.'); } - vmService = await VMService.connect(uri); + vmService = VMService.connect(uri); printTrace('Connected to service protocol: $uri'); await vmService.getVM();
diff --git a/packages/flutter_tools/lib/src/vmservice.dart b/packages/flutter_tools/lib/src/vmservice.dart index 63632cf..95a9bb6 100644 --- a/packages/flutter_tools/lib/src/vmservice.dart +++ b/packages/flutter_tools/lib/src/vmservice.dart
@@ -7,13 +7,23 @@ import 'package:json_rpc_2/error_code.dart' as rpc_error_code; import 'package:json_rpc_2/json_rpc_2.dart' as rpc; +import 'package:meta/meta.dart'; import 'package:stream_channel/stream_channel.dart'; import 'package:web_socket_channel/io.dart'; +import 'package:web_socket_channel/web_socket_channel.dart'; +import 'base/common.dart'; import 'base/file_system.dart'; -import 'base/io.dart'; import 'globals.dart'; +/// A function that opens a two-way communication channel to the specified [uri]. +typedef StreamChannel<dynamic> OpenChannel(Uri uri); + +OpenChannel _openChannel = _defaultOpenChannel; + +StreamChannel<dynamic> _defaultOpenChannel(Uri uri) => + new IOWebSocketChannel.connect(uri.toString()); + /// The default VM service request timeout. const Duration kDefaultRequestTimeout = const Duration(seconds: 10); @@ -22,35 +32,41 @@ /// A connection to the Dart VM Service. class VMService { - VMService._(this.peer, this.httpAddress, this.wsAddress, this._requestTimeout) { + VMService._(this._peer, this.httpAddress, this.wsAddress, this._requestTimeout) { _vm = new VM._empty(this); + _peer.listen().catchError((dynamic e, StackTrace stackTrace) { + _connectionError.completeError(e, stackTrace); + }); - peer.registerMethod('streamNotify', (rpc.Parameters event) { + _peer.registerMethod('streamNotify', (rpc.Parameters event) { _handleStreamNotify(event.asMap); }); } + @visibleForTesting + static void setOpenChannelForTesting(OpenChannel openChannel) { + _openChannel = openChannel; + } + /// Connect to a Dart VM Service at [httpUri]. /// /// Requests made via the returns [VMService] time out after [requestTimeout] /// amount of time, which is [kDefaultRequestTimeout] by default. - static Future<VMService> connect(Uri httpUri, { Duration requestTimeout: kDefaultRequestTimeout }) async { + static VMService connect( + Uri httpUri, { + Duration requestTimeout: kDefaultRequestTimeout, + }) { Uri wsUri = httpUri.replace(scheme: 'ws', path: fs.path.join(httpUri.path, 'ws')); - WebSocket ws; - try { - ws = await WebSocket.connect(wsUri.toString()); - } catch (e) { - return new Future<VMService>.error('Failed to connect to $wsUri\n $e'); - } - rpc.Peer peer = new rpc.Peer.withoutJson(jsonDocument.bind(new IOWebSocketChannel(ws).cast())); - peer.listen(); + StreamChannel<dynamic> channel = _openChannel(wsUri); + rpc.Peer peer = new rpc.Peer.withoutJson(jsonDocument.bind(channel.cast())); return new VMService._(peer, httpUri, wsUri, requestTimeout); } final Uri httpAddress; final Uri wsAddress; - final rpc.Peer peer; + final rpc.Peer _peer; final Duration _requestTimeout; + final Completer<Map<String, dynamic>> _connectionError = new Completer<Map<String, dynamic>>(); VM _vm; /// The singleton [VM] object. Owns [Isolate] and [FlutterView] objects. @@ -61,8 +77,9 @@ Set<String> _listeningFor = new Set<String>(); - bool get isClosed => peer.isClosed; - Future<Null> get done => peer.done; + /// Whether our connection to the VM service has been closed; + bool get isClosed => _peer.isClosed; + Future<Null> get done => _peer.done; // Events Stream<ServiceEvent> get onDebugEvent => onEvent('Debug'); @@ -78,6 +95,16 @@ return _getEventController(streamId).stream; } + Future<Map<String, dynamic>> _sendRequest( + String method, + Map<String, dynamic> params, + ) { + return Future.any(<Future<Map<String, dynamic>>>[ + _peer.sendRequest(method, params), + _connectionError.future, + ]); + } + StreamController<ServiceEvent> _getEventController(String eventName) { StreamController<ServiceEvent> controller = _eventControllers[eventName]; if (controller == null) { @@ -114,8 +141,7 @@ Future<Null> _streamListen(String streamId) async { if (!_listeningFor.contains(streamId)) { _listeningFor.add(streamId); - await peer.sendRequest('streamListen', - <String, dynamic>{ 'streamId': streamId }); + await _sendRequest('streamListen', <String, dynamic>{ 'streamId': streamId }); } } @@ -299,7 +325,7 @@ _inProgressReload = null; } - return _inProgressReload; + return await _inProgressReload; } /// Update [this] using [map] as a source. [map] can be a service reference. @@ -578,8 +604,8 @@ assert(params != null); timeout ??= _vmService._requestTimeout; try { - Map<String, dynamic> result = await _vmService.peer - .sendRequest(method, params) + Map<String, dynamic> result = await _vmService + ._sendRequest(method, params) .timeout(timeout); return result; } on TimeoutException { @@ -587,6 +613,9 @@ if (timeoutFatal) throw new TimeoutException('Request to Dart VM Service timed out: $method($params)'); return null; + } on WebSocketChannelException catch (error) { + throwToolExit('Error connecting to observatory: $error'); + return null; } }
diff --git a/packages/flutter_tools/pubspec.yaml b/packages/flutter_tools/pubspec.yaml index 1e70c2b..8b8fc38 100644 --- a/packages/flutter_tools/pubspec.yaml +++ b/packages/flutter_tools/pubspec.yaml
@@ -15,7 +15,9 @@ file: 2.0.1 http: ^0.11.3 intl: '>=0.14.0 <0.15.0' - json_rpc_2: ^2.0.0 + # TODO(tvolkert): Change to ^2.0.0 after manually vetting 2.0.4 release, + # which contains https://github.com/dart-lang/json_rpc_2/pull/19 + json_rpc_2: 2.0.3 json_schema: 1.0.6 linter: 0.1.30-alpha.1 meta: ^1.0.4