new ByteStreamClientChannel

pkg/analysis_server/test/channel_test.dart

 // Copyright (c) 2014, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 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';
 import 'package:unittest/unittest.dart';
 
 import 'mocks.dart';
 
 main() {
   group('WebSocketChannel', () {
     setUp(WebSocketChannelTest.setUp);
     test('close', WebSocketChannelTest.close);
     test('invalidJsonToClient', WebSocketChannelTest.invalidJsonToClient);
     test('invalidJsonToServer', WebSocketChannelTest.invalidJsonToServer);
     test('notification', WebSocketChannelTest.notification);
     test('notificationAndResponse', WebSocketChannelTest.notificationAndResponse);
     test('request', WebSocketChannelTest.request);
     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);
     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);
@@ skipping 127 matching lines @@
 
   static void expectMsgCount({requestCount: 0,
                               responseCount: 0,
                               notificationCount: 0}) {
     expect(requestsReceived, hasLength(requestCount));
     expect(responsesReceived, hasLength(responseCount));
     expect(notificationsReceived, hasLength(notificationCount));
   }
 }
 
+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;
 
   /**
    * Sink that may be used to deliver data to the channel, as though it's
    * coming from the client.
    */
   static IOSink inputSink;
 
   /**
@@ skipping 106 matching lines @@
     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'));
     });
   }
 }