diff --git a/CHANGELOG.md b/CHANGELOG.md
index ba15fb3..34123f9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,15 @@
+## 0.1.0
+
+* Remove `Server.handleRequest()` and `Server.parseRequest()`. Instead, `new
+  Server()` takes a `Stream` and a `StreamSink` and uses those behind-the-scenes
+  for its communication.
+
+* Add `Server.listen()`, which causes the server to begin listening to the
+  underlying request stream.
+
+* Add `Server.close()`, which closes the underlying request stream and response
+  sink.
+
 ## 0.0.2+3
 
 * Widen the version constraint for `stack_trace`.
diff --git a/lib/src/server.dart b/lib/src/server.dart
index d05c54f..c7ece5b 100644
--- a/lib/src/server.dart
+++ b/lib/src/server.dart
@@ -26,6 +26,20 @@
 /// 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 {
+  /// The stream for decoded requests.
+  final Stream _requests;
+
+  /// The subscription to the decoded request stream.
+  StreamSubscription _requestSubscription;
+
+  /// The sink for decoded responses.
+  final StreamSink _responses;
+
+  /// The completer for [listen].
+  ///
+  /// This is non-`null` after [listen] has been called.
+  Completer _listenCompleter;
+
   /// The methods registered for this server.
   final _methods = new Map<String, Function>();
 
@@ -35,7 +49,104 @@
   /// [RpcException.methodNotFound] exception.
   final _fallbacks = new Queue<Function>();
 
-  Server();
+  /// Creates a [Server] that reads requests from [requests] and writes
+  /// responses to [responses].
+  ///
+  /// If [requests] is a [StreamSink] as well as a [Stream] (for example, a
+  /// `WebSocket`), [responses] may be omitted.
+  ///
+  /// Note that the server won't begin listening to [requests] until
+  /// [Server.listen] is called.
+  factory Server(Stream<String> requests, [StreamSink<String> responses]) {
+    if (responses == null) {
+      if (requests is! StreamSink) {
+        throw new ArgumentError("Either `requests` must be a StreamSink or "
+            "`responses` must be passed.");
+      }
+      responses = requests as StreamSink;
+    }
+
+    var wrappedResponses = mapStreamSink(responses, JSON.encode);
+    return new Server.withoutJson(requests.expand((request) {
+      var decodedRequest;
+      try {
+        decodedRequest = JSON.decode(request);
+      } on FormatException catch (error) {
+        wrappedResponses.add(new RpcException(error_code.PARSE_ERROR,
+            'Invalid JSON: ${error.message}').serialize(request));
+        return [];
+      }
+
+      return [decodedRequest];
+    }), wrappedResponses);
+  }
+
+  /// Creates a [Server] that reads decoded requests from [requests] and writes
+  /// decoded responses to [responses].
+  ///
+  /// Unlike [new Server], this doesn't read or write JSON strings. Instead, it
+  /// reads and writes decoded maps or lists.
+  ///
+  /// If [requests] is a [StreamSink] as well as a [Stream], [responses] may be
+  /// omitted.
+  ///
+  /// Note that the server won't begin listening to [requests] until
+  /// [Server.listen] is called.
+  Server.withoutJson(Stream requests, [StreamSink responses])
+      : _requests = requests,
+        _responses = responses == null && requests is StreamSink ?
+            requests : responses {
+    if (_responses == null) {
+      throw new ArgumentError("Either `requests` must be a StreamSink or "
+          "`responses` must be passed.");
+    }
+  }
+
+  /// Starts listening to the underlying stream.
+  ///
+  /// Returns a [Future] that will complete when the stream is closed or when it
+  /// has an error.
+  ///
+  /// [listen] may only be called once.
+  Future listen() {
+    if (_listenCompleter != null) {
+      throw new StateError(
+          "Can only call Server.listen once on a given server.");
+    }
+
+    _listenCompleter = new Completer();
+    _requestSubscription = _requests.listen(_handleRequest,
+        onError: (error, stackTrace) {
+      if (_listenCompleter.isCompleted) return;
+      _responses.close();
+      _listenCompleter.completeError(error, stackTrace);
+    }, onDone: () {
+      if (_listenCompleter.isCompleted) return;
+      _responses.close();
+      _listenCompleter.complete();
+    }, cancelOnError: true);
+
+    return _listenCompleter.future;
+  }
+
+  /// Closes the server's request subscription and response sink.
+  ///
+  /// Returns a [Future] that completes when all resources have been released.
+  ///
+  /// A server can't be closed before [listen] has been called.
+  Future close() {
+    if (_listenCompleter == null) {
+      throw new StateError("Can't call Server.close before Server.listen.");
+    }
+
+    if (!_listenCompleter.isCompleted) _listenCompleter.complete();
+
+    var subscriptionFuture = _requestSubscription.cancel();
+    // TODO(nweiz): include the response future in the return value when issue
+    // 19095 is fixed.
+    _responses.close();
+    return subscriptionFuture == null ? new Future.value() : subscriptionFuture;
+  }
 
   /// Registers a method named [name] on this server.
   ///
@@ -69,14 +180,14 @@
     _fallbacks.add(callback);
   }
 
-  /// Handle a request that's already been parsed from JSON.
+  /// Handle a request.
   ///
   /// [request] is expected to be a JSON-serializable object representing a
   /// request sent by a client. This calls the appropriate method or methods for
   /// handling that request and returns a JSON-serializable response, or `null`
   /// if no response should be sent. [callback] may send custom
   /// errors by throwing an [RpcException].
-  Future handleRequest(request) {
+  Future _handleRequest(request) {
     return syncFuture(() {
       if (request is! List) return _handleSingleRequest(request);
       if (request.isEmpty) {
@@ -88,28 +199,7 @@
         var nonNull = results.where((result) => result != null);
         return nonNull.isEmpty ? null : nonNull.toList();
       });
-    });
-  }
-
-  /// Parses and handles a JSON serialized request.
-  ///
-  /// This calls the appropriate method or methods for handling that request and
-  /// returns a JSON string, or `null` if no response should be sent.
-  Future<String> parseRequest(String request) {
-    return syncFuture(() {
-      var decodedRequest;
-      try {
-        decodedRequest = JSON.decode(request);
-      } on FormatException catch (error) {
-        return new RpcException(error_code.PARSE_ERROR, 'Invalid JSON: '
-            '${error.message}').serialize(request);
-      }
-
-      return handleRequest(decodedRequest);
-    }).then((response) {
-      if (response == null) return null;
-      return JSON.encode(response);
-    });
+    }).then(_responses.add);
   }
 
   /// Handles an individual parsed request.
diff --git a/lib/src/utils.dart b/lib/src/utils.dart
index 1eff004..a212f58 100644
--- a/lib/src/utils.dart
+++ b/lib/src/utils.dart
@@ -43,3 +43,24 @@
 /// [toString], so we remove that if it exists.
 String getErrorMessage(error) =>
     error.toString().replaceFirst(_exceptionPrefix, '');
+
+/// Returns a [StreamSink] that wraps [sink] and maps each event added using
+/// [callback].
+StreamSink mapStreamSink(StreamSink sink, callback(event)) =>
+    new _MappedStreamSink(sink, callback);
+
+/// A [StreamSink] wrapper that maps each event added to the sink.
+class _MappedStreamSink implements StreamSink {
+  final StreamSink _inner;
+  final Function _callback;
+
+  Future get done => _inner.done;
+
+  _MappedStreamSink(this._inner, this._callback);
+
+  void add(event) => _inner.add(_callback(event));
+  void addError(error, [StackTrace stackTrace]) =>
+      _inner.addError(error, stackTrace);
+  Future addStream(Stream stream) => _inner.addStream(stream.map(_callback));
+  Future close() => _inner.close();
+}
diff --git a/pubspec.yaml b/pubspec.yaml
index 2cf1bc8..df1a8a0 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,5 +1,5 @@
 name: json_rpc_2
-version: 0.0.2+3
+version: 0.1.0
 author: Dart Team <misc@dartlang.org>
 description: An implementation of the JSON-RPC 2.0 spec.
 homepage: http://www.dartlang.org
@@ -7,7 +7,7 @@
 dependencies:
   stack_trace: '>=0.9.1 <2.0.0'
 dev_dependencies:
-  unittest: ">=0.9.0 <0.10.0"
+  unittest: ">=0.9.0 <0.12.0"
 environment:
   sdk: ">=1.2.0 <2.0.0"
 
diff --git a/test/server/batch_test.dart b/test/server/batch_test.dart
index 441df58..7dda84f 100644
--- a/test/server/batch_test.dart
+++ b/test/server/batch_test.dart
@@ -13,16 +13,17 @@
 import 'utils.dart';
 
 void main() {
-  var server;
+  var controller;
   setUp(() {
-    server = new json_rpc.Server()
+    controller = new ServerController();
+    controller.server
         ..registerMethod('foo', () => 'foo')
         ..registerMethod('id', (params) => params.value)
         ..registerMethod('arg', (params) => params['arg'].value);
   });
 
   test('handles a batch of requests', () {
-    expect(server.handleRequest([
+    expect(controller.handleRequest([
       {'jsonrpc': '2.0', 'method': 'foo', 'id': 1},
       {'jsonrpc': '2.0', 'method': 'id', 'params': ['value'], 'id': 2},
       {'jsonrpc': '2.0', 'method': 'arg', 'params': {'arg': 'value'}, 'id': 3}
@@ -34,7 +35,7 @@
   });
 
   test('handles errors individually', () {
-    expect(server.handleRequest([
+    expect(controller.handleRequest([
       {'jsonrpc': '2.0', 'method': 'foo', 'id': 1},
       {'jsonrpc': '2.0', 'method': 'zap', 'id': 2},
       {'jsonrpc': '2.0', 'method': 'arg', 'params': {'arg': 'value'}, 'id': 3}
@@ -54,7 +55,7 @@
   });
 
   test('handles notifications individually', () {
-    expect(server.handleRequest([
+    expect(controller.handleRequest([
       {'jsonrpc': '2.0', 'method': 'foo', 'id': 1},
       {'jsonrpc': '2.0', 'method': 'id', 'params': ['value']},
       {'jsonrpc': '2.0', 'method': 'arg', 'params': {'arg': 'value'}, 'id': 3}
@@ -65,7 +66,7 @@
   });
 
   test('returns nothing if every request is a notification', () {
-    expect(server.handleRequest([
+    expect(controller.handleRequest([
       {'jsonrpc': '2.0', 'method': 'foo'},
       {'jsonrpc': '2.0', 'method': 'id', 'params': ['value']},
       {'jsonrpc': '2.0', 'method': 'arg', 'params': {'arg': 'value'}}
@@ -73,24 +74,12 @@
   });
 
   test('returns an error if the batch is empty', () {
-    expectErrorResponse(server, [], error_code.INVALID_REQUEST,
+    expectErrorResponse(controller, [], error_code.INVALID_REQUEST,
         'A batch must contain at least one request.');
   });
 
-  test('handles a batch of requests parsed from JSON', () {
-    expect(server.parseRequest(JSON.encode([
-      {'jsonrpc': '2.0', 'method': 'foo', 'id': 1},
-      {'jsonrpc': '2.0', 'method': 'id', 'params': ['value'], 'id': 2},
-      {'jsonrpc': '2.0', 'method': 'arg', 'params': {'arg': 'value'}, 'id': 3}
-    ])), completion(equals(JSON.encode([
-      {'jsonrpc': '2.0', 'result': 'foo', 'id': 1},
-      {'jsonrpc': '2.0', 'result': ['value'], 'id': 2},
-      {'jsonrpc': '2.0', 'result': 'value', 'id': 3}
-    ]))));
-  });
-
   test('disallows nested batches', () {
-    expect(server.handleRequest([
+    expect(controller.handleRequest([
       [{'jsonrpc': '2.0', 'method': 'foo', 'id': 1}]
     ]), completion(equals([{
       'jsonrpc': '2.0',
diff --git a/test/server/invalid_request_test.dart b/test/server/invalid_request_test.dart
index feeefea..74cf86f 100644
--- a/test/server/invalid_request_test.dart
+++ b/test/server/invalid_request_test.dart
@@ -13,23 +13,23 @@
 import 'utils.dart';
 
 void main() {
-  var server;
-  setUp(() => server = new json_rpc.Server());
+  var controller;
+  setUp(() => controller = new ServerController());
 
   test("a non-Array/Object request is invalid", () {
-    expectErrorResponse(server, 'foo', error_code.INVALID_REQUEST,
+    expectErrorResponse(controller, 'foo', error_code.INVALID_REQUEST,
         'Request must be an Array or an Object.');
   });
 
   test("requests must have a jsonrpc key", () {
-    expectErrorResponse(server, {
+    expectErrorResponse(controller, {
       'method': 'foo',
       'id': 1234
     }, error_code.INVALID_REQUEST, 'Request must contain a "jsonrpc" key.');
   });
 
   test("the jsonrpc version must be 2.0", () {
-    expectErrorResponse(server, {
+    expectErrorResponse(controller, {
       'jsonrpc': '1.0',
       'method': 'foo',
       'id': 1234
@@ -38,14 +38,14 @@
   });
 
   test("requests must have a method key", () {
-    expectErrorResponse(server, {
+    expectErrorResponse(controller, {
       'jsonrpc': '2.0',
       'id': 1234
     }, error_code.INVALID_REQUEST, 'Request must contain a "method" key.');
   });
 
   test("request method must be a string", () {
-    expectErrorResponse(server, {
+    expectErrorResponse(controller, {
       'jsonrpc': '2.0',
       'method': 1234,
       'id': 1234
@@ -54,7 +54,7 @@
   });
 
   test("request params must be an Array or Object", () {
-    expectErrorResponse(server, {
+    expectErrorResponse(controller, {
       'jsonrpc': '2.0',
       'method': 'foo',
       'params': 1234,
@@ -64,7 +64,7 @@
   });
 
   test("request id may not be an Array or Object", () {
-    expect(server.handleRequest({
+    expect(controller.handleRequest({
       'jsonrpc': '2.0',
       'method': 'foo',
       'id': {'bad': 'id'}
diff --git a/test/server/server_test.dart b/test/server/server_test.dart
index c18a8ca..a89a0d9 100644
--- a/test/server/server_test.dart
+++ b/test/server/server_test.dart
@@ -13,15 +13,15 @@
 import 'utils.dart';
 
 void main() {
-  var server;
-  setUp(() => server = new json_rpc.Server());
+  var controller;
+  setUp(() => controller = new ServerController());
 
   test("calls a registered method with the given name", () {
-    server.registerMethod('foo', (params) {
+    controller.server.registerMethod('foo', (params) {
       return {'params': params.value};
     });
 
-    expect(server.handleRequest({
+    expect(controller.handleRequest({
       'jsonrpc': '2.0',
       'method': 'foo',
       'params': {'param': 'value'},
@@ -34,9 +34,9 @@
   });
 
   test("calls a method that takes no parameters", () {
-    server.registerMethod('foo', () => 'foo');
+    controller.server.registerMethod('foo', () => 'foo');
 
-    expect(server.handleRequest({
+    expect(controller.handleRequest({
       'jsonrpc': '2.0',
       'method': 'foo',
       'id': 1234
@@ -48,9 +48,9 @@
   });
 
   test("a method that takes no parameters rejects parameters", () {
-    server.registerMethod('foo', () => 'foo');
+    controller.server.registerMethod('foo', () => 'foo');
 
-    expectErrorResponse(server, {
+    expectErrorResponse(controller, {
       'jsonrpc': '2.0',
       'method': 'foo',
       'params': {},
@@ -61,9 +61,9 @@
   });
 
   test("an unexpected error in a method is captured", () {
-    server.registerMethod('foo', () => throw new FormatException('bad format'));
+    controller.server.registerMethod('foo', () => throw new FormatException('bad format'));
 
-    expect(server.handleRequest({
+    expect(controller.handleRequest({
       'jsonrpc': '2.0',
       'method': 'foo',
       'id': 1234
@@ -83,9 +83,9 @@
   });
 
   test("doesn't return a result for a notification", () {
-    server.registerMethod('foo', (args) => 'result');
+    controller.server.registerMethod('foo', (args) => 'result');
 
-    expect(server.handleRequest({
+    expect(controller.handleRequest({
       'jsonrpc': '2.0',
       'method': 'foo',
       'params': {}
@@ -93,11 +93,11 @@
   });
 
   test("includes the error data in the response", () {
-    server.registerMethod('foo', (params) {
+    controller.server.registerMethod('foo', (params) {
       throw new json_rpc.RpcException(5, 'Error message.', data: 'data value');
     });
 
-    expectErrorResponse(server, {
+    expectErrorResponse(controller, {
       'jsonrpc': '2.0',
       'method': 'foo',
       'params': {},
@@ -108,58 +108,28 @@
         data: 'data value');
   });
 
-  group("JSON", () {
-    test("handles a request parsed from JSON", () {
-      server.registerMethod('foo', (params) {
-        return {'params': params.value};
-      });
-
-      expect(server.parseRequest(JSON.encode({
+  test("a JSON parse error is rejected", () {
+    return controller.handleJsonRequest('invalid json {').then((result) {
+      expect(JSON.decode(result), {
         'jsonrpc': '2.0',
-        'method': 'foo',
-        'params': {'param': 'value'},
-        'id': 1234
-      })), completion(equals(JSON.encode({
-        'jsonrpc': '2.0',
-        'result': {'params': {'param': 'value'}},
-        'id': 1234
-      }))));
-    });
-
-    test("handles a notification parsed from JSON", () {
-      server.registerMethod('foo', (params) {
-        return {'params': params};
-      });
-
-      expect(server.parseRequest(JSON.encode({
-        'jsonrpc': '2.0',
-        'method': 'foo',
-        'params': {'param': 'value'}
-      })), completion(isNull));
-    });
-
-    test("a JSON parse error is rejected", () {
-      return server.parseRequest('invalid json {').then((result) {
-        expect(JSON.decode(result), {
-          'jsonrpc': '2.0',
-          'error': {
-            'code': error_code.PARSE_ERROR,
-            'message': startsWith("Invalid JSON: "),
-            'data': {'request': 'invalid json {'}
-          },
-          'id': null
-        });
+        'error': {
+          'code': error_code.PARSE_ERROR,
+          'message': startsWith("Invalid JSON: "),
+          'data': {'request': 'invalid json {'}
+        },
+        'id': null
       });
     });
   });
 
   group("fallbacks", () {
     test("calls a fallback if no method matches", () {
-      server.registerMethod('foo', () => 'foo');
-      server.registerMethod('bar', () => 'foo');
-      server.registerFallback((params) => {'fallback': params.value});
+      controller.server
+          ..registerMethod('foo', () => 'foo')
+          ..registerMethod('bar', () => 'foo')
+          ..registerFallback((params) => {'fallback': params.value});
 
-      expect(server.handleRequest({
+      expect(controller.handleRequest({
         'jsonrpc': '2.0',
         'method': 'baz',
         'params': {'param': 'value'},
@@ -172,13 +142,13 @@
     });
 
     test("calls the first matching fallback", () {
-      server.registerFallback((params) =>
-          throw new json_rpc.RpcException.methodNotFound(params.method));
+      controller.server
+          ..registerFallback((params) =>
+              throw new json_rpc.RpcException.methodNotFound(params.method))
+          ..registerFallback((params) => 'fallback 2')
+          ..registerFallback((params) => 'fallback 3');
 
-      server.registerFallback((params) => 'fallback 2');
-      server.registerFallback((params) => 'fallback 3');
-
-      expect(server.handleRequest({
+      expect(controller.handleRequest({
         'jsonrpc': '2.0',
         'method': 'fallback 2',
         'id': 1234
@@ -190,9 +160,10 @@
     });
 
     test("an unexpected error in a fallback is captured", () {
-      server.registerFallback((_) => throw new FormatException('bad format'));
+      controller.server.registerFallback((_) =>
+          throw new FormatException('bad format'));
 
-      expect(server.handleRequest({
+      expect(controller.handleRequest({
         'jsonrpc': '2.0',
         'method': 'foo',
         'id': 1234
@@ -213,7 +184,8 @@
   });
 
   test("disallows multiple methods with the same name", () {
-    server.registerMethod('foo', () => null);
-    expect(() => server.registerMethod('foo', () => null), throwsArgumentError);
+    controller.server.registerMethod('foo', () => null);
+    expect(() => controller.server.registerMethod('foo', () => null),
+        throwsArgumentError);
   });
 }
diff --git a/test/server/stream_test.dart b/test/server/stream_test.dart
new file mode 100644
index 0000000..5459e3e
--- /dev/null
+++ b/test/server/stream_test.dart
@@ -0,0 +1,98 @@
+// Copyright (c) 2014, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+library json_rpc_2.test.server.stream_test;
+
+import 'dart:async';
+
+import 'package:unittest/unittest.dart';
+import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
+
+import 'utils.dart';
+
+void main() {
+  test(".withoutJson supports decoded stream and sink", () {
+    var requestController = new StreamController();
+    var responseController = new StreamController();
+    var server = new json_rpc.Server.withoutJson(
+        requestController.stream, responseController.sink);
+    server.listen();
+
+    server.registerMethod('foo', (params) {
+      return {'params': params.value};
+    });
+
+    requestController.add({
+      'jsonrpc': '2.0',
+      'method': 'foo',
+      'params': {'param': 'value'},
+      'id': 1234
+    });
+
+    expect(responseController.stream.first, completion(equals({
+      'jsonrpc': '2.0',
+      'result': {'params': {'param': 'value'}},
+      'id': 1234
+    })));
+  });
+
+  test(".listen returns when the controller is closed", () {
+    var requestController = new StreamController();
+    var responseController = new StreamController();
+    var server = new json_rpc.Server(
+        requestController.stream, responseController.sink);
+
+    var hasListenCompeted = false;
+    expect(server.listen().then((_) => hasListenCompeted = true), completes);
+
+    return pumpEventQueue().then((_) {
+      expect(hasListenCompeted, isFalse);
+
+      // This should cause listen to complete.
+      return requestController.close();
+    });
+  });
+
+  test(".listen returns a stream error", () {
+    var requestController = new StreamController();
+    var responseController = new StreamController();
+    var server = new json_rpc.Server(
+        requestController.stream, responseController.sink);
+
+    expect(server.listen(), throwsA('oh no'));
+    requestController.addError('oh no');
+  });
+
+  test(".listen can't be called twice", () {
+    var requestController = new StreamController();
+    var responseController = new StreamController();
+    var server = new json_rpc.Server(
+        requestController.stream, responseController.sink);
+    server.listen();
+
+    expect(() => server.listen(), throwsStateError);
+  });
+
+  test(".close cancels the stream subscription and closes the sink", () {
+    var requestController = new StreamController();
+    var responseController = new StreamController();
+    var server = new json_rpc.Server(
+        requestController.stream, responseController.sink);
+
+    expect(server.listen(), completes);
+    expect(server.close(), completes);
+
+    expect(() => requestController.stream.listen((_) {}), throwsStateError);
+    expect(responseController.isClosed, isTrue);
+  });
+
+  test(".close can't be called before .listen", () {
+    var requestController = new StreamController();
+    var responseController = new StreamController();
+    var server = new json_rpc.Server(
+        requestController.stream, responseController.sink);
+
+    expect(() => server.close(), throwsStateError);
+  });
+}
diff --git a/test/server/utils.dart b/test/server/utils.dart
index 6f92c0a..a4fd36a 100644
--- a/test/server/utils.dart
+++ b/test/server/utils.dart
@@ -4,16 +4,53 @@
 
 library json_rpc_2.test.server.util;
 
-import 'package:unittest/unittest.dart';
-import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
+import 'dart:async';
+import 'dart:convert';
 
-void expectErrorResponse(json_rpc.Server server, request, int errorCode,
+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;
+
+  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 @@
   })));
 }
 
+/// 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));
+}