Create analysis server ByteStreamServerChannel class.

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 6d588873245b7ff4e33da793f34b1880255a7752..2a11638014cb17125fb6e6537d95748bf07c504a 100644
--- a/pkg/analysis_server/test/channel_test.dart
+++ b/pkg/analysis_server/test/channel_test.dart
@@ -5,6 +5,8 @@
 library test.channel;
 
 import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
 
 import 'package:analysis_server/src/channel.dart';
 import 'package:analysis_server/src/protocol.dart';
@@ -24,6 +26,18 @@ main() {
     test('requestResponse', WebSocketChannelTest.requestResponse);
     test('response', WebSocketChannelTest.response);
   });
+  group('ByteStreamServerChannel', () {
+    setUp(ByteStreamServerChannelTest.setUp);
+    test('listen_wellFormedRequest',
+        ByteStreamServerChannelTest.listen_wellFormedRequest);
+    test('listen_invalidRequest',
+        ByteStreamServerChannelTest.listen_invalidRequest);
+    test('listen_invalidJson', ByteStreamServerChannelTest.listen_invalidJson);
+    test('listen_streamError', ByteStreamServerChannelTest.listen_streamError);
+    test('listen_streamDone', ByteStreamServerChannelTest.listen_streamDone);
+    test('sendNotification', ByteStreamServerChannelTest.sendNotification);
+    test('sendResponse', ByteStreamServerChannelTest.sendResponse);
+  });
 }
 
 class WebSocketChannelTest {
@@ -156,4 +170,126 @@ class WebSocketChannelTest {
     expect(responsesReceived, hasLength(responseCount));
     expect(notificationsReceived, hasLength(notificationCount));
   }
-}
+}
+
+class ByteStreamServerChannelTest {
+  static ByteStreamServerChannel channel;
+
+  /**
+   * Sink that may be used to deliver data to the channel, as though it's
+   * coming from the client.
+   */
+  static IOSink inputSink;
+
+  /**
+   * Stream of lines sent back to the client by the channel.
+   */
+  static Stream<String> outputLineStream;
+
+  /**
+   * Stream of requests received from the channel via [listen()].
+   */
+  static Stream<Request> requestStream;
+
+  /**
+   * Stream of errors received from the channel via [listen()].
+   */
+  static Stream errorStream;
+
+  /**
+   * Future which is completed when then [listen()] reports [onDone].
+   */
+  static Future doneFuture;
+
+  static void setUp() {
+    StreamController<List<int>> inputStream = new StreamController<List<int>>();
+    inputSink = new IOSink(inputStream);
+    StreamController<List<int>> outputStream = new StreamController<List<int>>(
+        );
+    outputLineStream = outputStream.stream.transform((new Utf8Codec()).decoder
+        ).transform(new LineSplitter());
+    IOSink outputSink = new IOSink(outputStream);
+    channel = new ByteStreamServerChannel(inputStream.stream, outputSink);
+    StreamController<Request> requestStreamController =
+        new StreamController<Request>();
+    requestStream = requestStreamController.stream;
+    StreamController errorStreamController = new StreamController();
+    errorStream = errorStreamController.stream;
+    Completer doneCompleter = new Completer();
+    doneFuture = doneCompleter.future;
+    channel.listen((Request request) {
+      requestStreamController.add(request);
+    }, onError: (error) {
+      errorStreamController.add(error);
+    }, onDone: () {
+      doneCompleter.complete();
+    });
+  }
+
+  static Future listen_wellFormedRequest() {
+    inputSink.writeln('{"id":"0","method":"server.version"}');
+    return inputSink.flush().then((_) => requestStream.first.timeout(
+        new Duration(seconds: 1))).then((Request request) {
+      expect(request.id, equals("0"));
+      expect(request.method, equals("server.version"));
+    });
+  }
+
+  static Future listen_invalidRequest() {
+    inputSink.writeln('{"id":"0"}');
+    return inputSink.flush().then((_) => outputLineStream.first.timeout(
+        new Duration(seconds: 1))).then((String response) {
+      var jsonResponse = new JsonCodec().decode(response);
+      expect(jsonResponse, isMap);
+      expect(jsonResponse, contains('error'));
+      expect(jsonResponse['error'], isNotNull);
+    });
+  }
+
+  static Future listen_invalidJson() {
+    inputSink.writeln('{"id":');
+    return inputSink.flush().then((_) => outputLineStream.first.timeout(
+        new Duration(seconds: 1))).then((String response) {
+      var jsonResponse = new JsonCodec().decode(response);
+      expect(jsonResponse, isMap);
+      expect(jsonResponse, contains('error'));
+      expect(jsonResponse['error'], isNotNull);
+    });
+  }
+
+  static Future listen_streamError() {
+    var error = new Error();
+    inputSink.addError(error);
+    return inputSink.flush().then((_) => errorStream.first.timeout(new Duration(
+        seconds: 1))).then((var receivedError) {
+      expect(receivedError, same(error));
+    });
+  }
+
+  static Future listen_streamDone() {
+    return inputSink.close().then((_) => doneFuture.timeout(new Duration(
+        seconds: 1)));
+  }
+
+  static Future sendNotification() {
+    channel.sendNotification(new Notification('foo'));
+    return outputLineStream.first.timeout(new Duration(seconds: 1)).then((String
+        notification) {
+      var jsonNotification = new JsonCodec().decode(notification);
+      expect(jsonNotification, isMap);
+      expect(jsonNotification, contains('event'));
+      expect(jsonNotification['event'], equals('foo'));
+    });
+  }
+
+  static Future sendResponse() {
+    channel.sendResponse(new Response('foo'));
+    return outputLineStream.first.timeout(new Duration(seconds: 1)).then((String
+        response) {
+      var jsonResponse = new JsonCodec().decode(response);
+      expect(jsonResponse, isMap);
+      expect(jsonResponse, contains('id'));
+      expect(jsonResponse['id'], equals('foo'));
+    });
+  }
+}