Fix a race condition.

lib/src/channel_manager.dart

 // Copyright (c) 2016, 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.
 
 import 'dart:async';
 
 import 'package:stream_channel/stream_channel.dart';
 
 /// Wraps a [StreamChannel] and handles logic that's shared between [Server],
 /// [Client], and [Peer].
@@ skipping 38 matching lines @@
     if (_listenCalled) {
       throw new StateError("Can only call $_name.listen() once.");
     }
     _listenCalled = true;
 
     _channel.stream.listen(handleInput,
         onError: (error, stackTrace) {
           _doneCompleter.completeError(error, stackTrace);
           _channel.sink.close();
         },
-        onDone: _doneCompleter.complete,
+        onDone: () {
+          if (!_doneCompleter.isCompleted) _doneCompleter.complete();
+        },
         cancelOnError: true);
 
     return done;
   }
 
   /// Emit [event].
   void add(event) {
     if (isClosed && !_closeCalled) return;
     _channel.sink.add(event);
   }
 
   /// Closes the channel.
   Future close() {
     _closeCalled = true;
     if (!_doneCompleter.isCompleted) {
       _doneCompleter.complete(_channel.sink.close());
     }
     return done;
   }
 }