Avoid an open Peer with a closed Client

lib/src/peer.dart

@@ -35,11 +35,11 @@ class Peer implements Client, Server {
   /// they're responses.
   final _clientIncomingForwarder = StreamController(sync: true);
 
-  final _done = Completer<void>();
+  Future<void> _done;
   @override
-  Future get done => _done.future;
+  Future get done => _done ??= Future.wait([_client.done, _server.done]);
   @override
-  bool get isClosed => _done.isCompleted;
+  bool get isClosed => _client.isClosed || _server.isClosed;
 
   @override
   ErrorCallback get onUnhandledError => _server?.onUnhandledError;
@@ -142,15 +142,15 @@ class Peer implements Client, Server {
         _serverIncomingForwarder.add(message);
       }
     }, onError: (error, stackTrace) {
-      _done.completeError(error, stackTrace);
-      _channel.sink.close();
-    }, onDone: () {
-      if (!_done.isCompleted) _done.complete();
-      close();
-    });
+      _serverIncomingForwarder.addError(error, stackTrace);
+    }, onDone: close);
     return done;
   }
 
   @override
-  Future close() => Future.wait([_client.close(), _server.close()]);
+  Future close() {
+    _client.close();
+    _server.close();
+    return done;
+  }
 }

test/peer_test.dart

@@ -113,6 +113,21 @@ void main() {
     await peer.close();
   });
 
+  test('considered closed with misbehaving StreamChannel', () async {
+    // If a StreamChannel does not enforce the guarantees stated in it's
+    // contract - specifically that "Closing the sink causes the stream to close
+    // before it emits any more events." - The `Peer` should still understand
+    // when it has been closed manually.
+    var channel = StreamChannel(
+      StreamController().stream,
+      StreamController(),
+    );
+    var peer = json_rpc.Peer.withoutJson(channel);
+    unawaited(peer.listen());
+    unawaited(peer.close());
+    expect(peer.isClosed, true);
+  });
+
   group('like a server,', () {
     test('can receive a call and return a response', () {
       expect(outgoing.first,