new ByteStreamClientChannel

pkg/analysis_server/lib/src/channel.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 channel;
 
 import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
@@ skipping 152 matching lines @@
     } else if (data is List<int>) {
       // TODO(brianwilkerson) Implement a more efficient protocol.
       sendResponse(new Response.invalidRequestFormat());
     } else {
       sendResponse(new Response.invalidRequestFormat());
     }
   }
 }
 
 /**
+ * Instances of the class [ByteStreamClientChannel] implement a
+ * [ClientCommunicationChannel] that uses a stream and a sink (typically,
+ * standard input and standard output) to communicate with servers.
+ */
+class ByteStreamClientChannel implements ClientCommunicationChannel {
+  final Stream input;
+  final IOSink output;
+
+  @override
+  Stream<Response> responseStream;
+
+  @override
+  Stream<Notification> notificationStream;
+
+  ByteStreamClientChannel(this.input, this.output) {
+    Stream jsonStream = input.transform((new Utf8Codec()).decoder)
+        .transform(new LineSplitter())
+        .transform(new JsonStreamDecoder())
+        .where((json) => json is Map)
+        .asBroadcastStream();
+    responseStream = jsonStream
+        .where((json) => json[Notification.EVENT] == null)
+        .transform(new ResponseConverter())
+        .asBroadcastStream();
+    notificationStream = jsonStream
+        .where((json) => json[Notification.EVENT] != null)
+        .transform(new NotificationConverter())
+        .asBroadcastStream();
+  }
+
+  @override
+  Future close() {
+    // TODO: implement close
+  }
+
+  @override
+  Future<Response> sendRequest(Request request) {
+    String id = request.id;
+    output.writeln(JSON.encode(request.toJson()));
+    return responseStream.firstWhere((Response response) => response.id == id);
+  }
+}
+
+/**
  * Instances of the class [ByteStreamServerChannel] implement a
- * [ClientCommunicationChannel] that uses a stream and a sink (typically,
- * standard input and standard output) to communicate with servers.
+ * [ServerCommunicationChannel] that uses a stream and a sink (typically,
+ * standard input and standard output) to communicate with clients.
  */
 class ByteStreamServerChannel implements ServerCommunicationChannel {
   final Stream input;
   final IOSink output;
 
   /**
    * Completer that will be signalled when the input stream is closed.
    */
   final Completer _closed = new Completer();
 
@@ skipping 116 matching lines @@
       }
     }
   }
 
   @override
   void close() {
     closed = true;
     sink.close();
   }
 }