Remove ChannelManager class (flutter#64)

Fixes flutter#55

This class was intended to manage lifecycle around a `StreamChannel`,
but in the case of a `Peer` there is a conflict having 3 instances of a
`ChannelManager` attempting to manage lifecycle constraints which
results in completing the same `Completer` instance twice. Since `Peer`
needs a different behavior than `Server` or `Client` the
`ChannelManager` abstraction isn't helpful.

- Inline the important behavior around completing `done` appropriately
  for the end of the channel `stream`, or for errors, into `Peer`,
  `Server`, and `Client`.
- Inline the important behavior around closing the channel `sink` and
  completing `done` when a `Server` or `Client` is closed. `Peer` has
  different behavior and only forwards to it's `Server` and `Client`.
  The returned future from `channel.sink.close` is ignored, since this
  future does not complete in the case where a done even can't be
  delivered to listeners.

Author: natebosch

CHANGELOG.md

@@ -1,4 +1,6 @@
-## 2.2.2-dev
+## 2.2.2
+
+* Fix `Peer.close()` throwing `Bad state: Future already completed`.
 
 ## 2.2.1
 

lib/src/channel_manager.dart

@@ -7,7 +7,6 @@ import 'dart:async';
 import 'package:stack_trace/stack_trace.dart';
 import 'package:stream_channel/stream_channel.dart';
 
-import 'channel_manager.dart';
 import 'exception.dart';
 import 'utils.dart';
 
@@ -17,7 +16,7 @@ import 'utils.dart';
 /// those method calls. Methods can be called with [sendRequest], or with
 /// [sendNotification] if no response is expected.
 class Client {
-  final ChannelManager _manager;
+  final StreamChannel<dynamic> _channel;
 
   /// The next request id.
   var _id = 0;
@@ -30,19 +29,21 @@ class Client {
   /// The map of request ids to pending requests.
   final _pendingRequests = <int, _Request>{};
 
+  final _done = Completer<void>();
+
   /// Returns a [Future] that completes when the underlying connection is
   /// closed.
   ///
   /// This is the same future that's returned by [listen] and [close]. It may
   /// complete before [close] is called if the remote endpoint closes the
   /// connection.
-  Future get done => _manager.done;
+  Future get done => _done.future;
 
   /// Whether the underlying connection is closed.
   ///
   /// Note that this will be `true` before [close] is called if the remote
   /// endpoint closes the connection.
-  bool get isClosed => _manager.isClosed;
+  bool get isClosed => _done.isCompleted;
 
   /// Creates a [Client] that communicates over [channel].
   ///
@@ -60,9 +61,8 @@ class Client {
   ///
   /// Note that the client won't begin listening to [responses] until
   /// [Client.listen] is called.
-  Client.withoutJson(StreamChannel channel)
-      : _manager = ChannelManager('Client', channel) {
-    _manager.done.whenComplete(() {
+  Client.withoutJson(this._channel) {
+    done.whenComplete(() {
       for (var request in _pendingRequests.values) {
         request.completer.completeError(
             StateError(
@@ -81,13 +81,26 @@ class Client {
   /// when it has an error. This is the same as [done].
   ///
   /// [listen] may only be called once.
-  Future listen() => _manager.listen(_handleResponse);
+  Future listen() {
+    _channel.stream.listen(_handleResponse, onError: (error, stackTrace) {
+      _done.completeError(error, stackTrace);
+      _channel.sink.close();
+    }, onDone: () {
+      if (!_done.isCompleted) _done.complete();
+      close();
+    });
+    return done;
+  }
 
   /// Closes the underlying connection.
   ///
   /// Returns a [Future] that completes when all resources have been released.
   /// This is the same as [done].
-  Future close() => _manager.close();
+  Future close() {
+    _channel.sink.close();
+    if (!_done.isCompleted) _done.complete();
+    return done;
+  }
 
   /// Sends a JSON-RPC 2 request to invoke the given [method].
   ///
@@ -145,7 +158,7 @@ class Client {
     if (_batch != null) {
       _batch.add(message);
     } else {
-      _manager.add(message);
+      _channel.sink.add(message);
     }
   }
 
@@ -166,7 +179,7 @@ class Client {
 
     _batch = [];
     return tryFinally(callback, () {
-      _manager.add(_batch);
+      _channel.sink.add(_batch);
       _batch = null;
     });
   }

lib/src/client.dart

@@ -6,7 +6,6 @@ import 'dart:async';
 
 import 'package:stream_channel/stream_channel.dart';
 
-import 'channel_manager.dart';
 import 'client.dart';
 import 'parameters.dart';
 import 'server.dart';
@@ -18,7 +17,7 @@ import 'utils.dart';
 /// 2.0 endpoint. It sends both requests and responses across the same
 /// communication channel and expects to connect to a peer that does the same.
 class Peer implements Client, Server {
-  final ChannelManager _manager;
+  final StreamChannel<dynamic> _channel;
 
   /// The underlying client that handles request-sending and response-receiving
   /// logic.
@@ -36,10 +35,11 @@ class Peer implements Client, Server {
   /// they're responses.
   final _clientIncomingForwarder = StreamController(sync: true);
 
+  final _done = Completer<void>();
   @override
-  Future get done => _manager.done;
+  Future get done => _done.future;
   @override
-  bool get isClosed => _manager.isClosed;
+  bool get isClosed => _done.isCompleted;
 
   @override
   ErrorCallback get onUnhandledError => _server?.onUnhandledError;
@@ -81,15 +81,14 @@ class Peer implements Client, Server {
   /// some requests which are not conformant with the JSON-RPC 2.0
   /// specification. In particular, requests missing the `jsonrpc` parameter
   /// will be accepted.
-  Peer.withoutJson(StreamChannel channel,
-      {ErrorCallback onUnhandledError, bool strictProtocolChecks = true})
-      : _manager = ChannelManager('Peer', channel) {
+  Peer.withoutJson(this._channel,
+      {ErrorCallback onUnhandledError, bool strictProtocolChecks = true}) {
     _server = Server.withoutJson(
-        StreamChannel(_serverIncomingForwarder.stream, channel.sink),
+        StreamChannel(_serverIncomingForwarder.stream, _channel.sink),
         onUnhandledError: onUnhandledError,
         strictProtocolChecks: strictProtocolChecks);
     _client = Client.withoutJson(
-        StreamChannel(_clientIncomingForwarder.stream, channel.sink));
+        StreamChannel(_clientIncomingForwarder.stream, _channel.sink));
   }
 
   // Client methods.
@@ -121,7 +120,7 @@ class Peer implements Client, Server {
   Future listen() {
     _client.listen();
     _server.listen();
-    return _manager.listen((message) {
+    _channel.stream.listen((message) {
       if (message is Map) {
         if (message.containsKey('result') || message.containsKey('error')) {
           _clientIncomingForwarder.add(message);
@@ -142,10 +141,16 @@ class Peer implements Client, Server {
         // server since it knows how to send error responses.
         _serverIncomingForwarder.add(message);
       }
-    }).whenComplete(close);
+    }, onError: (error, stackTrace) {
+      _done.completeError(error, stackTrace);
+      _channel.sink.close();
+    }, onDone: () {
+      if (!_done.isCompleted) _done.complete();
+      close();
+    });
+    return done;
   }
 
   @override
-  Future close() =>
-      Future.wait([_client.close(), _server.close(), _manager.close()]);
+  Future close() => Future.wait([_client.close(), _server.close()]);
 }

lib/src/peer.dart

@@ -10,7 +10,6 @@ import 'package:stack_trace/stack_trace.dart';
 import 'package:stream_channel/stream_channel.dart';
 
 import '../error_code.dart' as error_code;
-import 'channel_manager.dart';
 import 'exception.dart';
 import 'parameters.dart';
 import 'utils.dart';
@@ -29,7 +28,7 @@ typedef ErrorCallback = void Function(dynamic error, dynamic stackTrace);
 /// asynchronously, it's possible for multiple methods to be invoked at the same
 /// time, or even for a single method to be invoked multiple times at once.
 class Server {
-  final ChannelManager _manager;
+  final StreamChannel<dynamic> _channel;
 
   /// The methods registered for this server.
   final _methods = <String, Function>{};
@@ -40,19 +39,21 @@ class Server {
   /// [RpcException.methodNotFound] exception.
   final _fallbacks = Queue<Function>();
 
+  final _done = Completer<void>();
+
   /// Returns a [Future] that completes when the underlying connection is
   /// closed.
   ///
   /// This is the same future that's returned by [listen] and [close]. It may
   /// complete before [close] is called if the remote endpoint closes the
   /// connection.
-  Future get done => _manager.done;
+  Future get done => _done.future;
 
   /// Whether the underlying connection is closed.
   ///
   /// Note that this will be `true` before [close] is called if the remote
   /// endpoint closes the connection.
-  bool get isClosed => _manager.isClosed;
+  bool get isClosed => _done.isCompleted;
 
   /// A callback that is fired on unhandled exceptions.
   ///
@@ -102,23 +103,34 @@ class Server {
   /// If [strictProtocolChecks] is false, this [Server] will accept some
   /// requests which are not conformant with the JSON-RPC 2.0 specification. In
   /// particular, requests missing the `jsonrpc` parameter will be accepted.
-  Server.withoutJson(StreamChannel channel,
-      {this.onUnhandledError, this.strictProtocolChecks = true})
-      : _manager = ChannelManager('Server', channel);
+  Server.withoutJson(this._channel,
+      {this.onUnhandledError, this.strictProtocolChecks = true});
 
   /// Starts listening to the underlying stream.
   ///
   /// Returns a [Future] that will complete when the connection is closed or
   /// when it has an error. This is the same as [done].
   ///
   /// [listen] may only be called once.
-  Future listen() => _manager.listen(_handleRequest);
+  Future listen() {
+    _channel.stream.listen(_handleRequest, onError: (error, stackTrace) {
+      _done.completeError(error, stackTrace);
+      _channel.sink.close();
+    }, onDone: () {
+      if (!_done.isCompleted) _done.complete();
+    });
+    return done;
+  }
 
   /// Closes the underlying connection.
   ///
   /// Returns a [Future] that completes when all resources have been released.
   /// This is the same as [done].
-  Future close() => _manager.close();
+  Future close() {
+    _channel.sink.close();
+    if (!_done.isCompleted) _done.complete();
+    return done;
+  }
 
   /// Registers a method named [name] on this server.
   ///
@@ -177,7 +189,7 @@ class Server {
       if (response == null) return;
     }
 
-    if (!isClosed) _manager.add(response);
+    if (!isClosed) _channel.sink.add(response);
   }
 
   /// Handles an individual parsed request.

lib/src/server.dart

@@ -1,5 +1,5 @@
 name: json_rpc_2
-version: 2.2.2-dev
+version: 2.2.2
 description: >-
   Utilities to write a client or server using the JSON-RPC 2.0 spec.
 homepage: https://github.com/dart-lang/json_rpc_2

pubspec.yaml

@@ -111,7 +111,7 @@ void main() {
     var peer = json_rpc.Peer.withoutJson(channel);
     unawaited(peer.listen());
     await peer.close();
-  }, skip: 'https://github.com/dart-lang/json_rpc_2/issues/55');
+  });
 
   group('like a server,', () {
     test('can receive a call and return a response', () {