Convert json_rpc.Server to take a Stream and StreamSink.

pkg/json_rpc_2/test/server/utils.dart

Index: pkg/json_rpc_2/test/server/utils.dart
diff --git a/pkg/json_rpc_2/test/server/utils.dart b/pkg/json_rpc_2/test/server/utils.dart
index 6f92c0a5456b46bb80fb4efb62c61d4c91935ac5..a4fd36af674b69f1e73a6091f2aad1505770cb96 100644
--- a/pkg/json_rpc_2/test/server/utils.dart
+++ b/pkg/json_rpc_2/test/server/utils.dart
@@ -4,16 +4,53 @@
 
 library json_rpc_2.test.server.util;
 
-import 'package:unittest/unittest.dart';
+import 'dart:async';
+import 'dart:convert';
+
 import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
+import 'package:json_rpc_2/error_code.dart' as error_code;
+import 'package:unittest/unittest.dart';
+
+/// A controller used to test a [json_rpc.Server].
+class ServerController {
+  /// The controller for the server's request stream.
+  final _requestController = new StreamController<String>();
+
+  /// The controller for the server's response sink.
+  final _responseController = new StreamController<String>();
+
+  /// The server.
+  json_rpc.Server get server => _server;
+  json_rpc.Server _server;
 
-void expectErrorResponse(json_rpc.Server server, request, int errorCode,
+  ServerController() {
+    _server = new json_rpc.Server(
+        _requestController.stream, _responseController.sink);
+    _server.listen();
+  }
+
+  /// Passes [request], a decoded request, to [server] and returns its decoded
+  /// response.
+  Future handleRequest(request) =>
+      handleJsonRequest(JSON.encode(request)).then(JSON.decode);
+
+  /// Passes [request], a JSON-encoded request, to [server] and returns its
+  /// encoded response.
+  Future handleJsonRequest(String request) {
+    _requestController.add(request);
+    return _responseController.stream.first;
+  }
+}
+
+/// Expects that [controller]'s server will return an error response to
+/// [request] with the given [errorCode], [message], and [data].
+void expectErrorResponse(ServerController controller, request, int errorCode,
     String message, {data}) {
   var id;
   if (request is Map) id = request['id'];
   if (data == null) data = {'request': request};
 
-  expect(server.handleRequest(request), completion(equals({
+  expect(controller.handleRequest(request), completion(equals({
     'jsonrpc': '2.0',
     'id': id,
     'error': {
@@ -24,10 +61,25 @@ void expectErrorResponse(json_rpc.Server server, request, int errorCode,
   })));
 }
 
+/// Returns a matcher that matches a [json_rpc.RpcException] with an
+/// `invalid_params` error code.
 Matcher throwsInvalidParams(String message) {
   return throwsA(predicate((error) {
     expect(error, new isInstanceOf<json_rpc.RpcException>());
+    expect(error.code, equals(error_code.INVALID_PARAMS));
     expect(error.message, equals(message));
     return true;
   }));
 }
+
+/// Returns a [Future] that completes after pumping the event queue [times]
+/// times. By default, this should pump the event queue enough times to allow
+/// any code to run, as long as it's not waiting on some external event.
+Future pumpEventQueue([int times = 20]) {
+  if (times == 0) return new Future.value();
+  // We use a delayed future to allow microtask events to finish. The
+  // Future.value or Future() constructors use scheduleMicrotask themselves and
+  // would therefore not wait for microtask callbacks that are scheduled after
+  // invoking this method.
+  return new Future.delayed(Duration.ZERO, () => pumpEventQueue(times - 1));
+}