Fix Peer requests not terminating when the channel closes

The `listen()` method of `Peer` never propagates close events from its manager to the `client` field. This causes in-flight requests to never terminate as the clean up handler at https://github.com/dart-lang/json_rpc_2/blob/d589e635d8ccb7cda6a804bd571f88abbabab146/lib/src/client.dart#L65-L72 is never called.

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 2.2.1
+
+* Fix `Peer` requests not terminating when the underlying channel is closed.
+
 ## 2.2.0
 
 * Added `strictProtocolChecks` named parameter to `Server` and `Peer`

lib/src/peer.dart

@@ -4,6 +4,7 @@
 
 import 'dart:async';
 
+import 'package:pedantic/pedantic.dart';
 import 'package:stream_channel/stream_channel.dart';
 
 import 'channel_manager.dart';
@@ -118,10 +119,10 @@ class Peer implements Client, Server {
   // Shared methods.
 
   @override
-  Future listen() {
-    _client.listen();
-    _server.listen();
-    return _manager.listen((message) {
+  Future listen() async {
+    unawaited(_client.listen());
+    unawaited(_server.listen());
+    await _manager.listen((message) {
       if (message is Map) {
         if (message.containsKey('result') || message.containsKey('error')) {
           _clientIncomingForwarder.add(message);
@@ -143,6 +144,8 @@ class Peer implements Client, Server {
         _serverIncomingForwarder.add(message);
       }
     });
+
+    return close();
   }
 
   @override

pubspec.yaml

@@ -8,9 +8,9 @@ environment:
   sdk: ">=2.2.0 <3.0.0"
 
 dependencies:
+  pedantic: ^1.8.0
   stack_trace: ^1.0.0
   stream_channel: ">=1.1.0 <3.0.0"
 
 dev_dependencies:
-  pedantic: ^1.8.0
   test: ^1.0.0

test/peer_test.dart

@@ -5,6 +5,7 @@
 import 'dart:async';
 import 'dart:convert';
 
+import 'package:pedantic/pedantic.dart';
 import 'package:stream_channel/stream_channel.dart';
 import 'package:test/test.dart';
 
@@ -84,6 +85,20 @@ void main() {
         expect(peer.sendRequest('w', {'x': 'y'}), completion(equals('z')));
       });
     });
+
+    test('requests terminates when the channel is closed', () async {
+      var incomingController = StreamController();
+      var channel = StreamChannel.withGuarantees(
+        incomingController.stream, StreamController(),
+      );
+      var peer = json_rpc.Peer.withoutJson(channel);
+      unawaited(peer.listen());
+
+      var response = peer.sendRequest('foo');
+      await incomingController.close();
+
+      expect(response, throwsStateError);
+    });
   });
 
   group('like a server,', () {