restructure client api to use streams

pkg/analysis_server/test/channel_test.dart

Index: pkg/analysis_server/test/channel_test.dart
diff --git a/pkg/analysis_server/test/channel_test.dart b/pkg/analysis_server/test/channel_test.dart
index 0c639ad64a3259b0b3aa50e7120c05d5a776d96c..1ea2f4b30dbb9079f372ba988cde89ec9e1dca5c 100644
--- a/pkg/analysis_server/test/channel_test.dart
+++ b/pkg/analysis_server/test/channel_test.dart
@@ -4,102 +4,128 @@
 
 library test.channel;
 
+import 'dart:async';
+
+import 'mocks.dart';
+
+import 'package:analysis_server/src/channel.dart';
 import 'package:analysis_server/src/protocol.dart';
 import 'package:unittest/matcher.dart';
 import 'package:unittest/unittest.dart';
-import 'package:analysis_server/src/channel.dart';
-import 'mocks.dart';
 
 main() {
   group('Channel', () {
+    setUp(ChannelTest.setUp);
     test('invalidJsonToClient', ChannelTest.invalidJsonToClient);
     test('invalidJsonToServer', ChannelTest.invalidJsonToServer);
     test('notification', ChannelTest.notification);
     test('request', ChannelTest.request);
+    test('requestResponse', ChannelTest.requestResponse);
     test('response', ChannelTest.response);
   });
 }
 
 class ChannelTest {
-
-  static void invalidJsonToClient() {
-    InvalidJsonMockSocket mockSocket = new InvalidJsonMockSocket();
-    WebSocketClientChannel client = new WebSocketClientChannel(mockSocket);
-    var responsesReceived = new List();
-    var notificationsReceived = new List();
-    client.listen((Response response) => responsesReceived.add(response),
-        (Notification notification) => notificationsReceived.add(notification));
-
-    mockSocket.addInvalid('"blat"');
-    mockSocket.addInvalid('{foo:bar}');
-
-    expect(responsesReceived.length, equals(0));
-    expect(notificationsReceived.length, equals(0));
-    expect(mockSocket.responseCount, equals(0));
+  static MockSocket socket;
+  static WebSocketClientChannel client;
+  static WebSocketServerChannel server;
+
+  static List requestsReceived;
+  static List responsesReceived;
+  static List notificationsReceived;
+
+  static void setUp() {
+    socket = new MockSocket.pair();
+    client = new WebSocketClientChannel(socket);
+    server = new WebSocketServerChannel(socket.twin);
+
+    requestsReceived = new List();
+    responsesReceived = new List();
+    notificationsReceived = new List();
+
+    server.listen((data) => requestsReceived.add(data));
+    client.responseStream.listen((data) => responsesReceived.add(data));
+    client.notificationStream.listen((data) => notificationsReceived.add(data));
   }
 
-  static void invalidJsonToServer() {
-    InvalidJsonMockSocket mockSocket = new InvalidJsonMockSocket();
-    WebSocketServerChannel server = new WebSocketServerChannel(mockSocket);
-    var received = new List();
-    server.listen((Request request) => received.add(request));
-
-    mockSocket.addInvalid('"blat"');
-    mockSocket.addInvalid('{foo:bar}');
-
-    expect(received.length, equals(0));
-    expect(mockSocket.responseCount, equals(2));
+  static Future invalidJsonToClient() {
+    socket.twin.add('{"foo":"bar"}');
+    server.sendResponse(new Response('myId'));
+    return client.responseStream
+        .first
+        .timeout(new Duration(seconds: 1))
+        .then((Response response) {
+          expect(response.id, equals('myId'));
+          expectMsgCount(0, 1, 0);
+        });
   }
 
-  static void notification() {
-    MockSocket mockSocket = new MockSocket();
-    WebSocketClientChannel client = new WebSocketClientChannel(mockSocket);
-    WebSocketServerChannel server = new WebSocketServerChannel(mockSocket);
-    var responsesReceived = new List();
-    var notificationsReceived = new List();
-    client.listen((Response response) => responsesReceived.add(response),
-        (Notification notification) => notificationsReceived.add(notification));
+  static Future invalidJsonToServer() {
+    socket.add('"blat"');
+    return client.responseStream
+        .first
+        .timeout(new Duration(seconds: 1))
+        .then((Response response) {
+          expect(response.id, equals(''));
+          expect(response.error, isNotNull);
+          expectMsgCount(0, 1, 0);
+        });
+  }
 
+  static Future notification() {
     server.sendNotification(new Notification('myEvent'));
-
-    expect(responsesReceived.length, equals(0));
-    expect(notificationsReceived.length, equals(1));
-    expect(notificationsReceived.first.runtimeType, equals(Notification));
-    Notification actual = notificationsReceived.first;
-    expect(actual.event, equals('myEvent'));
+    return client.notificationStream
+        .first
+        .timeout(new Duration(seconds: 1))
+        .then((Notification notification) {
+          expect(notification.event, equals('myEvent'));
+          expectMsgCount(0, 0, 1);
+
+          expect(notificationsReceived.first, equals(notification));
+        });
   }
 
   static void request() {
-    MockSocket mockSocket = new MockSocket();
-    WebSocketClientChannel client = new WebSocketClientChannel(mockSocket);
-    WebSocketServerChannel server = new WebSocketServerChannel(mockSocket);
-    var requestsReceived = new List();
-    server.listen((Request request) => requestsReceived.add(request));
-
-    client.sendRequest(new Request('myId', 'aMethod'));
-
-    expect(requestsReceived.length, equals(1));
-    expect(requestsReceived.first.runtimeType, equals(Request));
-    Request actual = requestsReceived.first;
-    expect(actual.id, equals('myId'));
-    expect(actual.method, equals('aMethod'));
+    client.sendRequest(new Request('myId', 'myMth'));
+    server.listen((Request request) {
+      expect(request.id, equals('myId'));
+      expect(request.method, equals('myMth'));
+      expectMsgCount(1, 0, 0);
+    });
   }
 
-  static void response() {
-    MockSocket mockSocket = new MockSocket();
-    WebSocketClientChannel client = new WebSocketClientChannel(mockSocket);
-    WebSocketServerChannel server = new WebSocketServerChannel(mockSocket);
-    var responsesReceived = new List();
-    var notificationsReceived = new List();
-    client.listen((Response response) => responsesReceived.add(response),
-        (Notification notification) => notificationsReceived.add(notification));
+  static Future requestResponse() {
+    // Simulate server sending a response by echoing the request
+    server.listen((Request request) =>
+        server.sendResponse(new Response(request.id)));
+    return client.sendRequest(new Request('myId', 'myMth'))
+        .timeout(new Duration(seconds: 1))
+        .then((Response response) {
+          expect(response.id, equals('myId'));
+          expectMsgCount(1, 1, 0);
+
+          expect(requestsReceived.first.runtimeType, equals(Request));
+          Request request = requestsReceived.first;
+          expect(request.id, equals('myId'));
+          expect(request.method, equals('myMth'));
+          expect(responsesReceived.first, equals(response));
+        });
+  }
 
+  static Future response() {
     server.sendResponse(new Response('myId'));
+    return client.responseStream
+        .first
+        .timeout(new Duration(seconds: 1))
+        .then((Response response) {
+          expect(response.id, equals('myId'));
+          expectMsgCount(0, 1, 0);
+        });
+  }
 
-    expect(responsesReceived.length, equals(1));
-    expect(notificationsReceived.length, equals(0));
-    expect(responsesReceived.first.runtimeType, equals(Response));
-    Response actual = responsesReceived.first;
-    expect(actual.id, equals('myId'));
+  static void expectMsgCount(requestCount, responseCount, notificationCount) {
+    expect(requestsReceived, hasLength(requestCount));
+    expect(responsesReceived, hasLength(responseCount));
+    expect(notificationsReceived, hasLength(notificationCount));
   }
 }