restructure client api to use streams

pkg/analysis_server/test/mocks.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 mocks;
 
+import 'dart:async';
+import 'dart:convert';
 import 'dart:io';
-import 'dart:async';
 
 /**
- * A mock [WebSocket] that immediately passes data to the listener.
+ * A mock [WebSocket] for testing.
  */
 class MockSocket<T> implements WebSocket {
-  var onData;
+
+  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;
+  }
+
+  MockSocket();
+
+  void add(T text) => controller.add(text);
+
+  void allowMultipleListeners() {
+    stream = stream.asBroadcastStream();
+  }
+
+  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}) {
-    this.onData = onData;
-    return null;
-  }
-  void add(T event) => onData(event);
-  noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
-}
+                     { Function onError, void onDone(), bool cancelOnError}) =>
+    stream.listen(onData, onError: onError, onDone: onDone,
+        cancelOnError: cancelOnError);
 
-/**
- * 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;
-  }
-  void addInvalid(T event) => onData(event);
-  void add(T event) { responseCount++; }
+  Stream<T> where(bool test(T)) => stream.where(test);
+
   noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
 }