| OLD | NEW |
|---|
| (Empty) |
| 1 // Copyright (c) 2014, the Dart project authors. Please see the AUTHORS file | |
| 2 // for details. All rights reserved. Use of this source code is governed by a | |
| 3 // BSD-style license that can be found in the LICENSE file. | |
| 4 | |
| 5 library channel; | |
| 6 | |
| 7 import 'dart:async'; | |
| 8 import 'dart:convert'; | |
| 9 import 'dart:io'; | |
| 10 | |
| 11 import 'package:analysis_server/src/protocol.dart'; | |
| 12 | |
| 13 /** | |
| 14 * The abstract class [ClientCommunicationChannel] defines the behavior of | |
| 15 * objects that allow a client to send [Request]s to an [AnalysisServer] and to | |
| 16 * receive both [Response]s and [Notification]s. | |
| 17 */ | |
| 18 abstract class ClientCommunicationChannel { | |
| 19 /** | |
| 20 * The stream of notifications from the server. | |
| 21 */ | |
| 22 Stream<Notification> notificationStream; | |
| 23 | |
| 24 /** | |
| 25 * The stream of responses from the server. | |
| 26 */ | |
| 27 Stream<Response> responseStream; | |
| 28 | |
| 29 /** | |
| 30 * Send the given [request] to the server | |
| 31 * and return a future with the associated [Response]. | |
| 32 */ | |
| 33 Future<Response> sendRequest(Request request); | |
| 34 | |
| 35 /** | |
| 36 * Close the channel to the server. Once called, all future communication | |
| 37 * with the server via [sendRequest] will silently be ignored. | |
| 38 */ | |
| 39 Future close(); | |
| 40 } | |
| 41 | |
| 42 /** | |
| 43 * The abstract class [ServerCommunicationChannel] defines the behavior of | |
| 44 * objects that allow an [AnalysisServer] to receive [Request]s and to return | |
| 45 * both [Response]s and [Notification]s. | |
| 46 */ | |
| 47 abstract class ServerCommunicationChannel { | |
| 48 /** | |
| 49 * Listen to the channel for requests. If a request is received, invoke the | |
| 50 * [onRequest] function. If an error is encountered while trying to read from | |
| 51 * the socket, invoke the [onError] function. If the socket is closed by the | |
| 52 * client, invoke the [onDone] function. | |
| 53 * Only one listener is allowed per channel. | |
| 54 */ | |
| 55 void listen(void onRequest(Request request), {Function onError, void onDone()}
); | |
| 56 | |
| 57 /** | |
| 58 * Send the given [notification] to the client. | |
| 59 */ | |
| 60 void sendNotification(Notification notification); | |
| 61 | |
| 62 /** | |
| 63 * Send the given [response] to the client. | |
| 64 */ | |
| 65 void sendResponse(Response response); | |
| 66 | |
| 67 /** | |
| 68 * Close the communication channel. | |
| 69 */ | |
| 70 void close(); | |
| 71 } | |
| 72 | |
| 73 /** | |
| 74 * Instances of the class [WebSocketClientChannel] implement a | |
| 75 * [ClientCommunicationChannel] that uses a [WebSocket] to communicate with | |
| 76 * servers. | |
| 77 */ | |
| 78 class WebSocketClientChannel implements ClientCommunicationChannel { | |
| 79 /** | |
| 80 * The socket being wrapped. | |
| 81 */ | |
| 82 final WebSocket socket; | |
| 83 | |
| 84 @override | |
| 85 Stream<Response> responseStream; | |
| 86 | |
| 87 @override | |
| 88 Stream<Notification> notificationStream; | |
| 89 | |
| 90 /** | |
| 91 * Initialize a new [WebSocket] wrapper for the given [socket]. | |
| 92 */ | |
| 93 WebSocketClientChannel(this.socket) { | |
| 94 Stream jsonStream = socket | |
| 95 .where((data) => data is String) | |
| 96 .transform(new JsonStreamDecoder()) | |
| 97 .where((json) => json is Map) | |
| 98 .asBroadcastStream(); | |
| 99 responseStream = jsonStream | |
| 100 .where((json) => json[Notification.EVENT] == null) | |
| 101 .transform(new ResponseConverter()) | |
| 102 .asBroadcastStream(); | |
| 103 notificationStream = jsonStream | |
| 104 .where((json) => json[Notification.EVENT] != null) | |
| 105 .transform(new NotificationConverter()) | |
| 106 .asBroadcastStream(); | |
| 107 } | |
| 108 | |
| 109 @override | |
| 110 Future<Response> sendRequest(Request request) { | |
| 111 String id = request.id; | |
| 112 socket.add(JSON.encode(request.toJson())); | |
| 113 return responseStream.firstWhere((Response response) => response.id == id); | |
| 114 } | |
| 115 | |
| 116 @override | |
| 117 Future close() { | |
| 118 return socket.close(); | |
| 119 } | |
| 120 } | |
| 121 | |
| 122 /** | |
| 123 * Instances of the class [WebSocketServerChannel] implement a | |
| 124 * [ServerCommunicationChannel] that uses a [WebSocket] to communicate with | |
| 125 * clients. | |
| 126 */ | |
| 127 class WebSocketServerChannel implements ServerCommunicationChannel { | |
| 128 /** | |
| 129 * The socket being wrapped. | |
| 130 */ | |
| 131 final WebSocket socket; | |
| 132 | |
| 133 /** | |
| 134 * Initialize a newly create [WebSocket] wrapper to wrap the given [socket]. | |
| 135 */ | |
| 136 WebSocketServerChannel(this.socket); | |
| 137 | |
| 138 @override | |
| 139 void listen(void onRequest(Request request), {void onError(), void onDone()})
{ | |
| 140 socket.listen((data) => readRequest(data, onRequest), onError: onError, | |
| 141 onDone: onDone); | |
| 142 } | |
| 143 | |
| 144 @override | |
| 145 void sendNotification(Notification notification) { | |
| 146 socket.add(JSON.encode(notification.toJson())); | |
| 147 } | |
| 148 | |
| 149 @override | |
| 150 void sendResponse(Response response) { | |
| 151 socket.add(JSON.encode(response.toJson())); | |
| 152 } | |
| 153 | |
| 154 /** | |
| 155 * Read a request from the given [data] and use the given function to handle | |
| 156 * the request. | |
| 157 */ | |
| 158 void readRequest(Object data, void onRequest(Request request)) { | |
| 159 if (data is String) { | |
| 160 // Parse the string as a JSON descriptor and process the resulting | |
| 161 // structure as a request. | |
| 162 Request request = new Request.fromString(data); | |
| 163 if (request == null) { | |
| 164 sendResponse(new Response.invalidRequestFormat()); | |
| 165 return; | |
| 166 } | |
| 167 onRequest(request); | |
| 168 } else if (data is List<int>) { | |
| 169 // TODO(brianwilkerson) Implement a more efficient protocol. | |
| 170 sendResponse(new Response.invalidRequestFormat()); | |
| 171 } else { | |
| 172 sendResponse(new Response.invalidRequestFormat()); | |
| 173 } | |
| 174 } | |
| 175 | |
| 176 @override | |
| 177 void close() { | |
| 178 socket.close(WebSocketStatus.NORMAL_CLOSURE); | |
| 179 } | |
| 180 } | |
| 181 | |
| 182 /** | |
| 183 * Instances of the class [ByteStreamClientChannel] implement a | |
| 184 * [ClientCommunicationChannel] that uses a stream and a sink (typically, | |
| 185 * standard input and standard output) to communicate with servers. | |
| 186 */ | |
| 187 class ByteStreamClientChannel implements ClientCommunicationChannel { | |
| 188 final Stream input; | |
| 189 final IOSink output; | |
| 190 | |
| 191 @override | |
| 192 Stream<Response> responseStream; | |
| 193 | |
| 194 @override | |
| 195 Stream<Notification> notificationStream; | |
| 196 | |
| 197 ByteStreamClientChannel(this.input, this.output) { | |
| 198 Stream jsonStream = input.transform((new Utf8Codec()).decoder) | |
| 199 .transform(new LineSplitter()) | |
| 200 .transform(new JsonStreamDecoder()) | |
| 201 .where((json) => json is Map) | |
| 202 .asBroadcastStream(); | |
| 203 responseStream = jsonStream | |
| 204 .where((json) => json[Notification.EVENT] == null) | |
| 205 .transform(new ResponseConverter()) | |
| 206 .asBroadcastStream(); | |
| 207 notificationStream = jsonStream | |
| 208 .where((json) => json[Notification.EVENT] != null) | |
| 209 .transform(new NotificationConverter()) | |
| 210 .asBroadcastStream(); | |
| 211 } | |
| 212 | |
| 213 @override | |
| 214 Future close() { | |
| 215 return output.close(); | |
| 216 } | |
| 217 | |
| 218 @override | |
| 219 Future<Response> sendRequest(Request request) { | |
| 220 String id = request.id; | |
| 221 output.writeln(JSON.encode(request.toJson())); | |
| 222 return responseStream.firstWhere((Response response) => response.id == id); | |
| 223 } | |
| 224 } | |
| 225 | |
| 226 /** | |
| 227 * Instances of the class [ByteStreamServerChannel] implement a | |
| 228 * [ServerCommunicationChannel] that uses a stream and a sink (typically, | |
| 229 * standard input and standard output) to communicate with clients. | |
| 230 */ | |
| 231 class ByteStreamServerChannel implements ServerCommunicationChannel { | |
| 232 final Stream input; | |
| 233 final IOSink output; | |
| 234 | |
| 235 /** | |
| 236 * Completer that will be signalled when the input stream is closed. | |
| 237 */ | |
| 238 final Completer _closed = new Completer(); | |
| 239 | |
| 240 ByteStreamServerChannel(this.input, this.output); | |
| 241 | |
| 242 /** | |
| 243 * Future that will be completed when the input stream is closed. | |
| 244 */ | |
| 245 Future get closed { | |
| 246 return _closed.future; | |
| 247 } | |
| 248 | |
| 249 @override | |
| 250 void listen(void onRequest(Request request), {Function onError, void | |
| 251 onDone()}) { | |
| 252 input.transform((new Utf8Codec()).decoder).transform(new LineSplitter() | |
| 253 ).listen((String data) => _readRequest(data, onRequest), onError: onErro
r, | |
| 254 onDone: () { | |
| 255 close(); | |
| 256 onDone(); | |
| 257 }); | |
| 258 } | |
| 259 | |
| 260 @override | |
| 261 void sendNotification(Notification notification) { | |
| 262 // Don't send any further notifications after the communication channel is | |
| 263 // closed. | |
| 264 if (_closed.isCompleted) { | |
| 265 return; | |
| 266 } | |
| 267 output.writeln(JSON.encode(notification.toJson())); | |
| 268 } | |
| 269 | |
| 270 @override | |
| 271 void sendResponse(Response response) { | |
| 272 // Don't send any further responses after the communication channel is | |
| 273 // closed. | |
| 274 if (_closed.isCompleted) { | |
| 275 return; | |
| 276 } | |
| 277 output.writeln(JSON.encode(response.toJson())); | |
| 278 } | |
| 279 | |
| 280 /** | |
| 281 * Read a request from the given [data] and use the given function to handle | |
| 282 * the request. | |
| 283 */ | |
| 284 void _readRequest(Object data, void onRequest(Request request)) { | |
| 285 // Ignore any further requests after the communication channel is closed. | |
| 286 if (_closed.isCompleted) { | |
| 287 return; | |
| 288 } | |
| 289 // Parse the string as a JSON descriptor and process the resulting | |
| 290 // structure as a request. | |
| 291 Request request = new Request.fromString(data); | |
| 292 if (request == null) { | |
| 293 sendResponse(new Response.invalidRequestFormat()); | |
| 294 return; | |
| 295 } | |
| 296 onRequest(request); | |
| 297 } | |
| 298 | |
| 299 @override | |
| 300 void close() { | |
| 301 if (!_closed.isCompleted) { | |
| 302 _closed.complete(); | |
| 303 } | |
| 304 } | |
| 305 } | |
| 306 | |
| 307 /** | |
| 308 * Instances of the class [JsonStreamDecoder] convert JSON strings to JSON | |
| 309 * maps. | |
| 310 */ | |
| 311 class JsonStreamDecoder extends Converter<String, Map> { | |
| 312 @override | |
| 313 Map convert(String text) => JSON.decode(text); | |
| 314 | |
| 315 @override | |
| 316 ChunkedConversionSink startChunkedConversion(Sink sink) => | |
| 317 new ChannelChunkSink<String, Map>(this, sink); | |
| 318 } | |
| 319 | |
| 320 /** | |
| 321 * Instances of the class [ResponseConverter] convert JSON maps to [Response]s. | |
| 322 */ | |
| 323 class ResponseConverter extends Converter<Map, Response> { | |
| 324 @override | |
| 325 Response convert(Map json) => new Response.fromJson(json); | |
| 326 | |
| 327 @override | |
| 328 ChunkedConversionSink startChunkedConversion(Sink sink) => | |
| 329 new ChannelChunkSink<Map, Response>(this, sink); | |
| 330 } | |
| 331 | |
| 332 /** | |
| 333 * Instances of the class [NotificationConverter] convert JSON maps to | |
| 334 * [Notification]s. | |
| 335 */ | |
| 336 class NotificationConverter extends Converter<Map, Notification> { | |
| 337 @override | |
| 338 Notification convert(Map json) => new Notification.fromJson(json); | |
| 339 | |
| 340 @override | |
| 341 ChunkedConversionSink startChunkedConversion(Sink sink) => | |
| 342 new ChannelChunkSink<Map, Notification>(this, sink); | |
| 343 } | |
| 344 | |
| 345 /** | |
| 346 * Instances of the class [ChannelChunkSink] uses a [Converter] to translate | |
| 347 * chunks. | |
| 348 */ | |
| 349 class ChannelChunkSink<S, T> extends ChunkedConversionSink<S> { | |
| 350 /** | |
| 351 * The converter used to translate chunks. | |
| 352 */ | |
| 353 final Converter<S, T> converter; | |
| 354 | |
| 355 /** | |
| 356 * The sink to which the converted chunks are added. | |
| 357 */ | |
| 358 final Sink sink; | |
| 359 | |
| 360 /** | |
| 361 * A flag indicating whether the sink has been closed. | |
| 362 */ | |
| 363 bool closed = false; | |
| 364 | |
| 365 /** | |
| 366 * Initialize a newly create sink to use the given [converter] to convert | |
| 367 * chunks before adding them to the given [sink]. | |
| 368 */ | |
| 369 ChannelChunkSink(this.converter, this.sink); | |
| 370 | |
| 371 @override | |
| 372 void add(S chunk) { | |
| 373 if (!closed) { | |
| 374 T convertedChunk = converter.convert(chunk); | |
| 375 if (convertedChunk != null) { | |
| 376 sink.add(convertedChunk); | |
| 377 } | |
| 378 } | |
| 379 } | |
| 380 | |
| 381 @override | |
| 382 void close() { | |
| 383 closed = true; | |
| 384 sink.close(); | |
| 385 } | |
| 386 } | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 544693002:
Split channels library. (Closed)
Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart