implement ByteStreamClientChannel close

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';
 
@@ skipping 10 matching lines @@
     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('close', ByteStreamClientChannelTest.close);
     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',
@@ skipping 136 matching lines @@
     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.
+   * coming from the server.
    */
   static IOSink inputSink;
 
   /**
+   * Sink through which the channel delivers data to the server.
+   */
+  static IOSink outputSink;
+
+  /**
    * 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);
+    outputSink = new IOSink(outputStream);
     channel = new ByteStreamClientChannel(inputStream.stream, outputSink);
   }
 
+  static Future close() {
+    bool doneCalled = false;
+    bool closeCalled = false;
+    // add listener so that outputSink will trigger done/close futures
+    outputLineStream.listen((_) { /* no-op */ });
+    outputSink.done.then((_) {
+      doneCalled = true;
+    });
+    channel.close().then((_) {
+      closeCalled = true;
+    });
+    return pumpEventQueue().then((_) {
+      expect(doneCalled, isTrue);
+      expect(closeCalled, isTrue);
+    });
+  }
+
   static Future listen_notification() {
     List<Notification> notifications = [];
     channel.notificationStream.forEach((n) => notifications.add(n));
     inputSink.writeln('{"event":"server.connected"}');
     return pumpEventQueue().then((_) {
       expect(notifications.length, equals(1));
       expect(notifications[0].event, equals('server.connected'));
     });
   }
 
@@ skipping 147 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'));
     });
   }
 }