new ByteStreamClientChannel

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 1f23745d5ae86bca23e5923599099acd40f7e28e..4d66c0917ae6563b91097ffd27ca421c2e60fda3 100644
--- a/pkg/analysis_server/test/channel_test.dart
+++ b/pkg/analysis_server/test/channel_test.dart
@@ -26,6 +26,12 @@ main() {
     test('requestResponse', WebSocketChannelTest.requestResponse);
     test('response', WebSocketChannelTest.response);
   });
+  group('ByteStreamClientChannel',  () {
+    setUp(ByteStreamClientChannelTest.setUp);
+    test('listen_notification', ByteStreamClientChannelTest.listen_notification);
+    test('listen_response', ByteStreamClientChannelTest.listen_response);
+    test('sendRequest', ByteStreamClientChannelTest.sendRequest);
+  });
   group('ByteStreamServerChannel', () {
     setUp(ByteStreamServerChannelTest.setUp);
     test('closed', ByteStreamServerChannelTest.closed);
@@ -173,6 +179,67 @@ class WebSocketChannelTest {
   }
 }
 
+class ByteStreamClientChannelTest {
+  static ByteStreamClientChannel 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;
+
+  static void setUp() {
+    var inputStream = new StreamController<List<int>>();
+    inputSink = new IOSink(inputStream);
+    var outputStream = new StreamController<List<int>>();
+    outputLineStream = outputStream.stream.transform((new Utf8Codec()).decoder
+        ).transform(new LineSplitter());
+    var outputSink = new IOSink(outputStream);
+    channel = new ByteStreamClientChannel(inputStream.stream, outputSink);
+  }
+
+  static Future listen_notification() {
+    List<Notification> notifications = [];
+    channel.notificationStream.forEach((n) => notifications.add(n));
+    inputSink.writeln('{"event":"server.connected"}');
+    return inputSink.flush().timeout(new Duration(seconds: 1)).then((_) {

[Paul Berry, 2014/05/11 05:09:00]

Using timeouts like this leads to nondeterministic tests (see r35221).  You
should be able to achieve the same effect deterministically using
pumpEventQueue, e.g.:

return inputSink.flush().then(pumpEventQueue()).then((_) {

(similar comment applies to the 2 methods below).

[danrubel, 2014/05/12 20:16:25]

Good idea. Done.

+      expect(notifications.length, equals(1));
+      expect(notifications[0].event, equals('server.connected'));
+    });
+  }
+
+  static Future listen_response() {
+    List<Response> responses = [];
+    channel.responseStream.forEach((n) => responses.add(n));
+    inputSink.writeln('{"id":"72"}');
+    return inputSink.flush().timeout(new Duration(seconds: 1)).then((_) {
+      expect(responses.length, equals(1));
+      expect(responses[0].id, equals('72'));
+    });
+  }
+
+  static Future sendRequest() {
+    Request request = new Request('72', 'foo.bar');
+    outputLineStream.first.timeout(new Duration(seconds: 1))
+        .then((line) => new JsonDecoder().convert(line))
+        .then((json) {
+          expect(json[Request.ID], equals('72'));
+          expect(json[Request.METHOD], equals('foo.bar'));
+          inputSink.writeln('{"id":"73"}');
+          inputSink.writeln('{"id":"72"}');
+        });
+    return channel.sendRequest(request).timeout(new Duration(seconds: 1))
+        .then((Response response) {
+          expect(response.id, equals('72'));
+        });
+  }
+}
+
 class ByteStreamServerChannelTest {
   static ByteStreamServerChannel channel;