Convert json_rpc.Server to take a Stream and StreamSink.

pkg/json_rpc_2/test/server/stream_test.dart

+// 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);
+  });
+}