Add a done getter to Client, Server, and Peer.

lib/src/two_way_stream.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 json_rpc_2.two_way_stream;
 
 import 'dart:async';
 import 'dart:convert';
 
 import 'utils.dart';
@@ skipping 14 matching lines @@
   final Stream _input;
 
   /// The subscription to [_input].
   StreamSubscription _inputSubscription;
 
   /// The output sink.
   ///
   /// This takes decoded JSON objects.
   final StreamSink _output;
 
-  /// The completer for [listen].
+  /// Returns a [Future] that completes when the connection is closed.
   ///
-  /// This is non-`null` after [listen] has been called.
-  Completer _listenCompleter;
+  /// This is the same future that's returned by [listen].
+  Future get done => _doneCompleter.future;
+  final _doneCompleter = new Completer();
 
   /// Creates a two-way stream.
   ///
   /// [input] and [output] should emit and take (respectively) JSON-encoded
   /// strings.
   ///
   /// [inputName] is used in error messages as the name of the input parameter.
   /// [outputName] is likewise used as the name of the output parameter.
   ///
   /// If [onInvalidInput] is passed, any errors parsing messages from [input]
@@ skipping 39 matching lines @@
       throw new ArgumentError("Either `$inputName` must be a StreamSink or "
           "`$outputName` must be passed.");
     }
   }
 
   /// Starts listening to the input stream.
   ///
   /// The returned Future will complete when the input stream is closed. If the
   /// input stream emits an error, that will be piped to the returned Future.
   Future listen(void handleInput(input)) {
-    if (_listenCompleter != null) {
+    if (_inputSubscription != null) {
       throw new StateError("Can only call $_name.listen once.");
     }
 
-    _listenCompleter = new Completer();
     _inputSubscription = _input.listen(handleInput,
         onError: (error, stackTrace) {
-      if (_listenCompleter.isCompleted) return;
+      if (_doneCompleter.isCompleted) return;
       _output.close();
-      _listenCompleter.completeError(error, stackTrace);
+      _doneCompleter.completeError(error, stackTrace);
     }, onDone: () {
-      if (_listenCompleter.isCompleted) return;
+      if (_doneCompleter.isCompleted) return;
       _output.close();
-      _listenCompleter.complete();
+      _doneCompleter.complete();
     }, cancelOnError: true);
 
-    return _listenCompleter.future;
+    return _doneCompleter.future;
   }
 
   /// Emit [event] on the output stream.
   void add(event) => _output.add(event);
 
   /// Stops listening to the input stream and closes the output stream.
   Future close() {
-    if (_listenCompleter == null) {
+    if (_inputSubscription == null) {
       throw new StateError("Can't call $_name.close before $_name.listen.");
     }
 
-    if (!_listenCompleter.isCompleted) _listenCompleter.complete();
+    if (!_doneCompleter.isCompleted) _doneCompleter.complete();
 
     var inputFuture = _inputSubscription.cancel();
     // TODO(nweiz): include the output future in the return value when issue
     // 19095 is fixed.
     _output.close();
     return inputFuture == null ? new Future.value() : inputFuture;
   }
 }