Don't leave dangling request futures. (#19)

Closes #18

CHANGELOG.md

@@ -1,3 +1,12 @@
+## 2.0.4
+
+* `Client.sendRequest()` now throws a `StateError` if the client is closed while
+  the request is in-flight. This avoids dangling `Future`s that will never be
+  completed.
+
+* Both `Client.sendRequest()` and `Client.sendNotification()` now throw
+  `StateError`s if they're called after the client is closed.
+
 ## 2.0.3
 
 * Fix new strong-mode warnings.

lib/src/channel_manager.dart

@@ -25,7 +25,7 @@ class ChannelManager {
   ///
   /// This is the same future that's returned by [listen].
   Future get done => _doneCompleter.future;
-  final _doneCompleter = new Completer();
+  final _doneCompleter = new Completer.sync();
 
   /// Whether the underlying communication channel is closed.
   bool get isClosed => _doneCompleter.isCompleted;

lib/src/client.dart

@@ -61,7 +61,18 @@ class Client {
   /// Note that the client won't begin listening to [responses] until
   /// [Client.listen] is called.
   Client.withoutJson(StreamChannel channel)
-      : _manager = new ChannelManager("Client", channel);
+      : _manager = new ChannelManager("Client", channel) {
+    _manager.done.whenComplete(() {
+      for (var request in _pendingRequests.values) {
+        request.completer.completeError(
+            new StateError("The client closed with pending requests."),
+            StackTrace.current);
+      }
+      _pendingRequests.clear();
+    }).catchError((_) {
+      // Avoid an unhandled error.
+    });
+  }
 
   /// Starts listening to the underlying stream.
   ///
@@ -87,6 +98,9 @@ class Client {
   /// If the request succeeds, this returns the response result as a decoded
   /// JSON-serializable object. If it fails, it throws an [RpcException]
   /// describing the failure.
+  ///
+  /// Throws a [StateError] if the client is closed while the request is in
+  /// flight, or if the client is closed when this method is called.
   Future sendRequest(String method, [parameters]) {
     var id = _id++;
     _send(method, parameters, id);
@@ -106,6 +120,8 @@ class Client {
   ///
   /// Since this is just a notification to which the server isn't expected to
   /// send a response, it has no return value.
+  ///
+  /// Throws a [StateError] if the client is closed when this method is called.
   void sendNotification(String method, [parameters]) =>
       _send(method, parameters);
 
@@ -119,6 +135,7 @@ class Client {
       throw new ArgumentError('Only maps and lists may be used as JSON-RPC '
           'parameters, was "$parameters".');
     }
+    if (isClosed) throw new StateError("The client is closed.");
 
     var message = <String, dynamic>{
       "jsonrpc": "2.0",

pubspec.yaml

@@ -1,5 +1,5 @@
 name: json_rpc_2
-version: 2.0.3
+version: 2.0.4
 author: Dart Team <[email protected]>
 description: An implementation of the JSON-RPC 2.0 spec.
 homepage: http://github.com/dart-lang/json_rpc_2

test/client/client_test.dart

@@ -184,6 +184,12 @@ void main() {
     })));
   });
 
+  test("requests throw StateErrors if the client is closed", () {
+    controller.client.close();
+    expect(() => controller.client.sendRequest("foo"), throwsStateError);
+    expect(() => controller.client.sendNotification("foo"), throwsStateError);
+  });
+
   test("ignores bogus responses", () {
     // Make a request so we have something to respond to.
     controller.expectRequest((request) {

test/client/stream_test.dart

@@ -76,4 +76,27 @@ void main() {
     expect(() => responseController.stream.listen((_) {}), throwsStateError);
     expect(requestController.isClosed, isTrue);
   });
+
+  group("a stream error", () {
+    test("is reported through .done", () {
+      expect(client.listen(), throwsA("oh no!"));
+      expect(client.done, throwsA("oh no!"));
+      responseController.addError("oh no!");
+    });
+
+    test("cause a pending request to throw a StateError", () {
+      expect(client.listen(), throwsA("oh no!"));
+      expect(client.sendRequest('foo'), throwsStateError);
+      responseController.addError("oh no!");
+    });
+
+    test("causes future requests to throw StateErrors", () async {
+      expect(client.listen(), throwsA("oh no!"));
+      responseController.addError("oh no!");
+      await pumpEventQueue();
+
+      expect(() => client.sendRequest('foo'), throwsStateError);
+      expect(() => client.sendNotification('foo'), throwsStateError);
+    });
+  });
 }