restructure client api to use streams

pkg/analysis_server/test/mocks.dart

Index: pkg/analysis_server/test/mocks.dart
diff --git a/pkg/analysis_server/test/mocks.dart b/pkg/analysis_server/test/mocks.dart
index 1c87b0e331bd0b7d75ef2cd5c316f3ae46c82633..c8b8078ccc431f9b3400795c841c096330133f90 100644
--- a/pkg/analysis_server/test/mocks.dart
+++ b/pkg/analysis_server/test/mocks.dart
@@ -4,35 +4,50 @@
 
 library mocks;
 
-import 'dart:io';
 import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
 
 /**
- * A mock [WebSocket] that immediately passes data to the listener.
+ * A mock [WebSocket] for testing.
  */
 class MockSocket<T> implements WebSocket {
-  var onData;
-  StreamSubscription<T> listen(void onData(T event),
-                     {Function onError, void onDone(), bool cancelOnError}) {
-    this.onData = onData;
-    return null;
+
+  final JsonEncoder jsonEncoder = const JsonEncoder(null);
+
+  final JsonDecoder jsonDecoder = const JsonDecoder(null);
+
+  StreamController controller = new StreamController();
+  MockSocket twin;
+  Stream stream;
+
+  factory MockSocket.pair() {
+    MockSocket socket1 = new MockSocket();
+    MockSocket socket2 = new MockSocket();
+    socket1.twin = socket2;
+    socket2.twin = socket1;
+    socket1.stream = socket2.controller.stream;
+    socket2.stream = socket1.controller.stream;
+    return socket1;
   }
-  void add(T event) => onData(event);
-  noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
-}
 
-/**
- * A mock [WebSocket] for sending invalid JSON data and counting responses.
- */
-class InvalidJsonMockSocket<T> implements WebSocket {
-  int responseCount = 0;
-  var onData;
-  StreamSubscription<T> listen(void onData(T event),
-                     {Function onError, void onDone(), bool cancelOnError}) {
-    this.onData = onData;
-    return null;
+  MockSocket();
+
+  void add(T text) => controller.add(text);
+
+  void allowMultipleListeners() {
+    stream = stream.asBroadcastStream();
   }
-  void addInvalid(T event) => onData(event);
-  void add(T event) { responseCount++; }
+
+  Future close([int code, String reason]) => controller.close()
+      .then((_) => twin.controller.close());
+
+  StreamSubscription<T> listen(void onData(T event),
+                     { Function onError, void onDone(), bool cancelOnError}) =>
+    stream.listen(onData, onError: onError, onDone: onDone,
+        cancelOnError: cancelOnError);
+
+  Stream<T> where(bool test(T)) => stream.where(test);
+
   noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
 }