Revert "Reland "[ Service / package:dds ] Add stream support to packa…

…ge:dds and enable DDS for VM service tests""

This reverts commit e5b8579 as it seems to have broken service/pause_on_start_and_exit_with_child_test/service https://ci.chromium.org/p/dart/builders/ci.sandbox/app-kernel-linux-debug-x64/5877

Change-Id: Idb9df51816eebfb58137c449c6461731c77409f0
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/143881
Reviewed-by: Ben Konyi <bkonyi@google.com>
Commit-Queue: Ben Konyi <bkonyi@google.com>

DEPS

@@ -146,7 +146,7 @@ vars = {
   "usage_tag": "3.4.0",
   "watcher_rev": "0.9.7+14",
   "web_components_rev": "8f57dac273412a7172c8ade6f361b407e2e4ed02",
-  "web_socket_channel_tag": "1.1.0",
+  "web_socket_channel_tag": "1.0.15",
   "WebCore_rev": "fb11e887f77919450e497344da570d780e078bc8",
   "yaml_tag": "2.2.0",
   "zlib_rev": "c44fb7248079cc3d5563b14b3f758aee60d6b415",

pkg/dds/lib/dds.dart

@@ -7,24 +7,15 @@
 library dds;
 
 import 'dart:async';
-import 'dart:convert';
 import 'dart:io';
-import 'dart:typed_data';
 
-import 'package:async/async.dart';
-import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
-import 'package:pedantic/pedantic.dart';
 import 'package:shelf/shelf.dart';
 import 'package:shelf/shelf_io.dart' as io;
 import 'package:shelf_proxy/shelf_proxy.dart';
 import 'package:shelf_web_socket/shelf_web_socket.dart';
-import 'package:stream_channel/stream_channel.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
 
-part 'src/binary_compatible_peer.dart';
-part 'src/client.dart';
 part 'src/dds_impl.dart';
-part 'src/stream_manager.dart';
 
 /// An intermediary between a Dart VM service and its clients that offers
 /// additional functionality on top of the standard VM service protocol.

pkg/dds/lib/src/dds_impl.dart

@@ -5,20 +5,12 @@
 part of dds;
 
 class _DartDevelopmentService implements DartDevelopmentService {
-  _DartDevelopmentService(this._remoteVmServiceUri, this._uri) {
-    _streamManager = _StreamManager(this);
-  }
+  _DartDevelopmentService(this._remoteVmServiceUri, this._uri);
 
   Future<void> startService() async {
     // Establish the connection to the VM service.
-    _vmServiceSocket = WebSocketChannel.connect(remoteVmServiceWsUri);
-    _vmServiceClient = _BinaryCompatiblePeer(_vmServiceSocket, _streamManager);
-    // Setup the JSON RPC client with the VM service.
-    unawaited(_vmServiceClient.listen());
-
-    // Setup stream event handling.
-    streamManager.listen();
-
+    _vmServiceSocket = await WebSocket.connect(remoteVmServiceWsUri.toString());
+    _vmServiceStream = _vmServiceSocket.asBroadcastStream();
     // Once we have a connection to the VM service, we're ready to spawn the intermediary.
     await _startDDSServer();
   }
@@ -36,20 +28,8 @@ class _DartDevelopmentService implements DartDevelopmentService {
 
   /// Stop accepting requests after gracefully handling existing requests.
   Future<void> shutdown() async {
-    // Don't accept anymore HTTP requests.
     await _server.close();
-
-    // Close all incoming websocket connections.
-    final futures = <Future>[];
-    for (final client in _clients) {
-      futures.add(client.close());
-    }
-    await Future.wait(futures);
-
-    // Close connection to VM service.
-    await _vmServiceSocket.sink.close();
-
-    _done.complete();
+    await _vmServiceSocket.close();
   }
 
   // Attempt to upgrade HTTP requests to a websocket before processing them as
@@ -58,18 +38,16 @@ class _DartDevelopmentService implements DartDevelopmentService {
   Cascade _handlers() => Cascade().add(_webSocketHandler()).add(_httpHandler());
 
   Handler _webSocketHandler() => webSocketHandler((WebSocketChannel ws) {
-        final client = _DartDevelopmentServiceClient(
-          this,
-          ws,
-          _vmServiceClient,
+        // TODO(bkonyi): actually process requests instead of blindly forwarding them.
+        _vmServiceStream.listen(
+          (event) => ws.sink.add(event),
+          onDone: () => ws.sink.close(),
         );
-        _clients.add(client);
-        client.listen().then((_) => _clients.remove(client));
+        ws.stream.listen((event) => _vmServiceSocket.add(event));
       });
 
   Handler _httpHandler() {
-    // DDS doesn't support any HTTP requests itself, so we just forward all of
-    // them to the VM service.
+    // TODO(bkonyi): actually process requests instead of blindly forwarding them.
     final cascade = Cascade().add(proxyHandler(remoteVmServiceUri));
     return cascade.handler;
   }
@@ -101,15 +79,7 @@ class _DartDevelopmentService implements DartDevelopmentService {
 
   bool get isRunning => _uri != null;
 
-  Future<void> get done => _done.future;
-  Completer _done = Completer<void>();
-
-  _StreamManager get streamManager => _streamManager;
-  _StreamManager _streamManager;
-
-  final List<_DartDevelopmentServiceClient> _clients = [];
-
-  json_rpc.Peer _vmServiceClient;
-  WebSocketChannel _vmServiceSocket;
+  WebSocket _vmServiceSocket;
+  Stream _vmServiceStream;
   HttpServer _server;
 }

pkg/dds/pubspec.yaml

@@ -11,15 +11,13 @@ environment:
   sdk: '>=2.6.0 <3.0.0'
 
 dependencies:
-  async: ^2.4.1
   json_rpc_2: ^2.1.0
-  pedantic: ^1.7.0
   shelf: ^0.7.5
   shelf_proxy: ^0.1.0+7
   shelf_web_socket: ^0.2.3
-  stream_channel: ^2.0.0
   web_socket_channel: ^1.1.0
 
 dev_dependencies:
+  pedantic: ^1.7.0
   test: ^1.0.0
   vm_service: ^4.0.0