| 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 http_parser.web_socket; | |
| 6 | |
| 7 import 'dart:async'; | |
| 8 import 'dart:convert'; | |
| 9 import 'dart:math'; | |
| 10 import 'dart:typed_data'; | |
| 11 | |
| 12 import 'package:crypto/crypto.dart'; | |
| 13 | |
| 14 import 'bytes_builder.dart'; | |
| 15 | |
| 16 /// An implementation of the WebSocket protocol that's not specific to "dart:io" | |
| 17 /// or to any particular HTTP API. | |
| 18 /// | |
| 19 /// Because this is HTTP-API-agnostic, it doesn't handle the initial [WebSocket | |
| 20 /// handshake][]. This needs to be handled manually by the user of the code. | |
| 21 /// Once that's been done, [new CompatibleWebSocket] can be called with the | |
| 22 /// underlying socket and it will handle the remainder of the protocol. | |
| 23 /// | |
| 24 /// [WebSocket handshake]: https://tools.ietf.org/html/rfc6455#section-4 | |
| 25 abstract class CompatibleWebSocket implements Stream, StreamSink { | |
| 26 /// The interval for sending ping signals. | |
| 27 /// | |
| 28 /// If a ping message is not answered by a pong message from the peer, the | |
| 29 /// `WebSocket` is assumed disconnected and the connection is closed with a | |
| 30 /// [WebSocketStatus.GOING_AWAY] close code. When a ping signal is sent, the | |
| 31 /// pong message must be received within [pingInterval]. | |
| 32 /// | |
| 33 /// There are never two outstanding pings at any given time, and the next ping | |
| 34 /// timer starts when the pong is received. | |
| 35 /// | |
| 36 /// By default, the [pingInterval] is `null`, indicating that ping messages | |
| 37 /// are disabled. | |
| 38 Duration pingInterval; | |
| 39 | |
| 40 /// The [close code][] set when the WebSocket connection is closed. | |
| 41 /// | |
| 42 /// [close code]: https://tools.ietf.org/html/rfc6455#section-7.1.5 | |
| 43 /// | |
| 44 /// Before the connection has been closed, this will be `null`. | |
| 45 int get closeCode; | |
| 46 | |
| 47 /// The [close reason][] set when the WebSocket connection is closed. | |
| 48 /// | |
| 49 /// [close reason]: https://tools.ietf.org/html/rfc6455#section-7.1.6 | |
| 50 /// | |
| 51 /// Before the connection has been closed, this will be `null`. | |
| 52 String get closeReason; | |
| 53 | |
| 54 /// Signs a `Sec-WebSocket-Key` header sent by a WebSocket client as part of | |
| 55 /// the [initial handshake]. | |
| 56 /// | |
| 57 /// The return value should be sent back to the client in a | |
| 58 /// `Sec-WebSocket-Accept` header. | |
| 59 /// | |
| 60 /// [initial handshake]: https://tools.ietf.org/html/rfc6455#section-4.2.2 | |
| 61 static String signKey(String key) { | |
| 62 var hash = new SHA1(); | |
| 63 // We use [codeUnits] here rather than UTF-8-decoding the string because | |
| 64 // [key] is expected to be base64 encoded, and so will be pure ASCII. | |
| 65 hash.add((key + _webSocketGUID).codeUnits); | |
| 66 return CryptoUtils.bytesToBase64(hash.close()); | |
| 67 } | |
| 68 | |
| 69 /// Creates a new WebSocket handling messaging across an existing socket. | |
| 70 /// | |
| 71 /// Because this is HTTP-API-agnostic, the initial [WebSocket handshake][] | |
| 72 /// must have already been completed on the socket before this is called. | |
| 73 /// | |
| 74 /// If [stream] is also a [StreamSink] (for example, if it's a "dart:io" | |
| 75 /// `Socket`), it will be used for both sending and receiving data. Otherwise, | |
| 76 /// it will be used for receiving data and [sink] will be used for sending it. | |
| 77 /// | |
| 78 /// If this is a WebSocket server, [serverSide] should be `true` (the | |
| 79 /// default); if it's a client, [serverSide] should be `false`. | |
| 80 /// | |
| 81 /// [WebSocket handshake]: https://tools.ietf.org/html/rfc6455#section-4 | |
| 82 factory CompatibleWebSocket(Stream<List<int>> stream, | |
| 83 {StreamSink<List<int>> sink, bool serverSide: true}) { | |
| 84 if (sink == null) { | |
| 85 if (stream is! StreamSink) { | |
| 86 throw new ArgumentError("If stream isn't also a StreamSink, sink must " | |
| 87 "be passed explicitly."); | |
| 88 } | |
| 89 sink = stream as StreamSink; | |
| 90 } | |
| 91 | |
| 92 return new _WebSocketImpl._fromSocket(stream, sink, serverSide); | |
| 93 } | |
| 94 | |
| 95 /// Closes the web socket connection. | |
| 96 /// | |
| 97 /// [closeCode] and [closeReason] are the [close code][] and [reason][] sent | |
| 98 /// to the remote peer, respectively. If they are omitted, the peer will see | |
| 99 /// a "no status received" code with no reason. | |
| 100 /// | |
| 101 /// [close code]: https://tools.ietf.org/html/rfc6455#section-7.1.5 | |
| 102 /// [reason]: https://tools.ietf.org/html/rfc6455#section-7.1.6 | |
| 103 Future close([int closeCode, String closeReason]); | |
| 104 } | |
| 105 | |
| 106 /// An exception thrown by [CompatibleWebSocket]. | |
| 107 class CompatibleWebSocketException implements Exception { | |
| 108 final String message; | |
| 109 | |
| 110 CompatibleWebSocketException([this.message]); | |
| 111 | |
| 112 String toString() => message == null | |
| 113 ? "CompatibleWebSocketException" : | |
| 114 "CompatibleWebSocketException: $message"; | |
| 115 } | |
| 116 | |
| 117 // The following code is copied from sdk/lib/io/websocket_impl.dart. The | |
| 118 // "dart:io" implementation isn't used directly both to support non-"dart:io" | |
| 119 // applications, and because it's incompatible with non-"dart:io" HTTP requests | |
| 120 // (issue 18172). | |
| 121 // | |
| 122 // Because it's copied directly, only modifications necessary to support the | |
| 123 // desired public API and to remove "dart:io" dependencies have been made. | |
| 124 | |
| 125 /** | |
| 126 * Web socket status codes used when closing a web socket connection. | |
| 127 */ | |
| 128 abstract class _WebSocketStatus { | |
| 129 static const int NORMAL_CLOSURE = 1000; | |
| 130 static const int GOING_AWAY = 1001; | |
| 131 static const int PROTOCOL_ERROR = 1002; | |
| 132 static const int UNSUPPORTED_DATA = 1003; | |
| 133 static const int RESERVED_1004 = 1004; | |
| 134 static const int NO_STATUS_RECEIVED = 1005; | |
| 135 static const int ABNORMAL_CLOSURE = 1006; | |
| 136 static const int INVALID_FRAME_PAYLOAD_DATA = 1007; | |
| 137 static const int POLICY_VIOLATION = 1008; | |
| 138 static const int MESSAGE_TOO_BIG = 1009; | |
| 139 static const int MISSING_MANDATORY_EXTENSION = 1010; | |
| 140 static const int INTERNAL_SERVER_ERROR = 1011; | |
| 141 static const int RESERVED_1015 = 1015; | |
| 142 } | |
| 143 | |
| 144 abstract class _WebSocketState { | |
| 145 static const int CONNECTING = 0; | |
| 146 static const int OPEN = 1; | |
| 147 static const int CLOSING = 2; | |
| 148 static const int CLOSED = 3; | |
| 149 } | |
| 150 | |
| 151 const String _webSocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; | |
| 152 | |
| 153 final _random = new Random(); | |
| 154 | |
| 155 // Matches _WebSocketOpcode. | |
| 156 class _WebSocketMessageType { | |
| 157 static const int NONE = 0; | |
| 158 static const int TEXT = 1; | |
| 159 static const int BINARY = 2; | |
| 160 } | |
| 161 | |
| 162 | |
| 163 class _WebSocketOpcode { | |
| 164 static const int CONTINUATION = 0; | |
| 165 static const int TEXT = 1; | |
| 166 static const int BINARY = 2; | |
| 167 static const int RESERVED_3 = 3; | |
| 168 static const int RESERVED_4 = 4; | |
| 169 static const int RESERVED_5 = 5; | |
| 170 static const int RESERVED_6 = 6; | |
| 171 static const int RESERVED_7 = 7; | |
| 172 static const int CLOSE = 8; | |
| 173 static const int PING = 9; | |
| 174 static const int PONG = 10; | |
| 175 static const int RESERVED_B = 11; | |
| 176 static const int RESERVED_C = 12; | |
| 177 static const int RESERVED_D = 13; | |
| 178 static const int RESERVED_E = 14; | |
| 179 static const int RESERVED_F = 15; | |
| 180 } | |
| 181 | |
| 182 /** | |
| 183 * The web socket protocol transformer handles the protocol byte stream | |
| 184 * which is supplied through the [:handleData:]. As the protocol is processed, | |
| 185 * it'll output frame data as either a List<int> or String. | |
| 186 * | |
| 187 * Important infomation about usage: Be sure you use cancelOnError, so the | |
| 188 * socket will be closed when the processer encounter an error. Not using it | |
| 189 * will lead to undefined behaviour. | |
| 190 */ | |
| 191 // TODO(ajohnsen): make this transformer reusable? | |
| 192 class _WebSocketProtocolTransformer implements StreamTransformer, EventSink { | |
| 193 static const int START = 0; | |
| 194 static const int LEN_FIRST = 1; | |
| 195 static const int LEN_REST = 2; | |
| 196 static const int MASK = 3; | |
| 197 static const int PAYLOAD = 4; | |
| 198 static const int CLOSED = 5; | |
| 199 static const int FAILURE = 6; | |
| 200 | |
| 201 int _state = START; | |
| 202 bool _fin = false; | |
| 203 int _opcode = -1; | |
| 204 int _len = -1; | |
| 205 bool _masked = false; | |
| 206 int _remainingLenBytes = -1; | |
| 207 int _remainingMaskingKeyBytes = 4; | |
| 208 int _remainingPayloadBytes = -1; | |
| 209 int _unmaskingIndex = 0; | |
| 210 int _currentMessageType = _WebSocketMessageType.NONE; | |
| 211 int closeCode = _WebSocketStatus.NO_STATUS_RECEIVED; | |
| 212 String closeReason = ""; | |
| 213 | |
| 214 EventSink _eventSink; | |
| 215 | |
| 216 final bool _serverSide; | |
| 217 final List _maskingBytes = new List(4); | |
| 218 final BytesBuilder _payload = new BytesBuilder(copy: false); | |
| 219 | |
| 220 _WebSocketProtocolTransformer([this._serverSide = false]); | |
| 221 | |
| 222 Stream bind(Stream stream) { | |
| 223 return new Stream.eventTransformed( | |
| 224 stream, | |
| 225 (EventSink eventSink) { | |
| 226 if (_eventSink != null) { | |
| 227 throw new StateError("WebSocket transformer already used."); | |
| 228 } | |
| 229 _eventSink = eventSink; | |
| 230 return this; | |
| 231 }); | |
| 232 } | |
| 233 | |
| 234 void addError(Object error, [StackTrace stackTrace]) => | |
| 235 _eventSink.addError(error, stackTrace); | |
| 236 | |
| 237 void close() => _eventSink.close(); | |
| 238 | |
| 239 /** | |
| 240 * Process data received from the underlying communication channel. | |
| 241 */ | |
| 242 void add(Uint8List buffer) { | |
| 243 int count = buffer.length; | |
| 244 int index = 0; | |
| 245 int lastIndex = count; | |
| 246 if (_state == CLOSED) { | |
| 247 throw new CompatibleWebSocketException("Data on closed connection"); | |
| 248 } | |
| 249 if (_state == FAILURE) { | |
| 250 throw new CompatibleWebSocketException("Data on failed connection"); | |
| 251 } | |
| 252 while ((index < lastIndex) && _state != CLOSED && _state != FAILURE) { | |
| 253 int byte = buffer[index]; | |
| 254 if (_state <= LEN_REST) { | |
| 255 if (_state == START) { | |
| 256 _fin = (byte & 0x80) != 0; | |
| 257 if ((byte & 0x70) != 0) { | |
| 258 // The RSV1, RSV2 bits RSV3 must be all zero. | |
| 259 throw new CompatibleWebSocketException("Protocol error"); | |
| 260 } | |
| 261 _opcode = (byte & 0xF); | |
| 262 if (_opcode <= _WebSocketOpcode.BINARY) { | |
| 263 if (_opcode == _WebSocketOpcode.CONTINUATION) { | |
| 264 if (_currentMessageType == _WebSocketMessageType.NONE) { | |
| 265 throw new CompatibleWebSocketException("Protocol error"); | |
| 266 } | |
| 267 } else { | |
| 268 assert(_opcode == _WebSocketOpcode.TEXT || | |
| 269 _opcode == _WebSocketOpcode.BINARY); | |
| 270 if (_currentMessageType != _WebSocketMessageType.NONE) { | |
| 271 throw new CompatibleWebSocketException("Protocol error"); | |
| 272 } | |
| 273 _currentMessageType = _opcode; | |
| 274 } | |
| 275 } else if (_opcode >= _WebSocketOpcode.CLOSE && | |
| 276 _opcode <= _WebSocketOpcode.PONG) { | |
| 277 // Control frames cannot be fragmented. | |
| 278 if (!_fin) throw new CompatibleWebSocketException("Protocol error"); | |
| 279 } else { | |
| 280 throw new CompatibleWebSocketException("Protocol error"); | |
| 281 } | |
| 282 _state = LEN_FIRST; | |
| 283 } else if (_state == LEN_FIRST) { | |
| 284 _masked = (byte & 0x80) != 0; | |
| 285 _len = byte & 0x7F; | |
| 286 if (_isControlFrame() && _len > 125) { | |
| 287 throw new CompatibleWebSocketException("Protocol error"); | |
| 288 } | |
| 289 if (_len == 126) { | |
| 290 _len = 0; | |
| 291 _remainingLenBytes = 2; | |
| 292 _state = LEN_REST; | |
| 293 } else if (_len == 127) { | |
| 294 _len = 0; | |
| 295 _remainingLenBytes = 8; | |
| 296 _state = LEN_REST; | |
| 297 } else { | |
| 298 assert(_len < 126); | |
| 299 _lengthDone(); | |
| 300 } | |
| 301 } else { | |
| 302 assert(_state == LEN_REST); | |
| 303 _len = _len << 8 | byte; | |
| 304 _remainingLenBytes--; | |
| 305 if (_remainingLenBytes == 0) { | |
| 306 _lengthDone(); | |
| 307 } | |
| 308 } | |
| 309 } else { | |
| 310 if (_state == MASK) { | |
| 311 _maskingBytes[4 - _remainingMaskingKeyBytes--] = byte; | |
| 312 if (_remainingMaskingKeyBytes == 0) { | |
| 313 _maskDone(); | |
| 314 } | |
| 315 } else { | |
| 316 assert(_state == PAYLOAD); | |
| 317 // The payload is not handled one byte at a time but in blocks. | |
| 318 int payloadLength = min(lastIndex - index, _remainingPayloadBytes); | |
| 319 _remainingPayloadBytes -= payloadLength; | |
| 320 // Unmask payload if masked. | |
| 321 if (_masked) { | |
| 322 _unmask(index, payloadLength, buffer); | |
| 323 } | |
| 324 // Control frame and data frame share _payloads. | |
| 325 _payload.add( | |
| 326 new Uint8List.view(buffer.buffer, index, payloadLength)); | |
| 327 index += payloadLength; | |
| 328 if (_isControlFrame()) { | |
| 329 if (_remainingPayloadBytes == 0) _controlFrameEnd(); | |
| 330 } else { | |
| 331 if (_currentMessageType != _WebSocketMessageType.TEXT && | |
| 332 _currentMessageType != _WebSocketMessageType.BINARY) { | |
| 333 throw new CompatibleWebSocketException("Protocol error"); | |
| 334 } | |
| 335 if (_remainingPayloadBytes == 0) _messageFrameEnd(); | |
| 336 } | |
| 337 | |
| 338 // Hack - as we always do index++ below. | |
| 339 index--; | |
| 340 } | |
| 341 } | |
| 342 | |
| 343 // Move to the next byte. | |
| 344 index++; | |
| 345 } | |
| 346 } | |
| 347 | |
| 348 void _unmask(int index, int length, Uint8List buffer) { | |
| 349 const int BLOCK_SIZE = 16; | |
| 350 // Skip Int32x4-version if message is small. | |
| 351 if (length >= BLOCK_SIZE) { | |
| 352 // Start by aligning to 16 bytes. | |
| 353 final int startOffset = BLOCK_SIZE - (index & 15); | |
| 354 final int end = index + startOffset; | |
| 355 for (int i = index; i < end; i++) { | |
| 356 buffer[i] ^= _maskingBytes[_unmaskingIndex++ & 3]; | |
| 357 } | |
| 358 index += startOffset; | |
| 359 length -= startOffset; | |
| 360 final int blockCount = length ~/ BLOCK_SIZE; | |
| 361 if (blockCount > 0) { | |
| 362 // Create mask block. | |
| 363 int mask = 0; | |
| 364 for (int i = 3; i >= 0; i--) { | |
| 365 mask = (mask << 8) | _maskingBytes[(_unmaskingIndex + i) & 3]; | |
| 366 } | |
| 367 Int32x4 blockMask = new Int32x4(mask, mask, mask, mask); | |
| 368 Int32x4List blockBuffer = new Int32x4List.view( | |
| 369 buffer.buffer, index, blockCount); | |
| 370 for (int i = 0; i < blockBuffer.length; i++) { | |
| 371 blockBuffer[i] ^= blockMask; | |
| 372 } | |
| 373 final int bytes = blockCount * BLOCK_SIZE; | |
| 374 index += bytes; | |
| 375 length -= bytes; | |
| 376 } | |
| 377 } | |
| 378 // Handle end. | |
| 379 final int end = index + length; | |
| 380 for (int i = index; i < end; i++) { | |
| 381 buffer[i] ^= _maskingBytes[_unmaskingIndex++ & 3]; | |
| 382 } | |
| 383 } | |
| 384 | |
| 385 void _lengthDone() { | |
| 386 if (_masked) { | |
| 387 if (!_serverSide) { | |
| 388 throw new CompatibleWebSocketException( | |
| 389 "Received masked frame from server"); | |
| 390 } | |
| 391 _state = MASK; | |
| 392 } else { | |
| 393 if (_serverSide) { | |
| 394 throw new CompatibleWebSocketException( | |
| 395 "Received unmasked frame from client"); | |
| 396 } | |
| 397 _remainingPayloadBytes = _len; | |
| 398 _startPayload(); | |
| 399 } | |
| 400 } | |
| 401 | |
| 402 void _maskDone() { | |
| 403 _remainingPayloadBytes = _len; | |
| 404 _startPayload(); | |
| 405 } | |
| 406 | |
| 407 void _startPayload() { | |
| 408 // If there is no actual payload perform perform callbacks without | |
| 409 // going through the PAYLOAD state. | |
| 410 if (_remainingPayloadBytes == 0) { | |
| 411 if (_isControlFrame()) { | |
| 412 switch (_opcode) { | |
| 413 case _WebSocketOpcode.CLOSE: | |
| 414 _state = CLOSED; | |
| 415 _eventSink.close(); | |
| 416 break; | |
| 417 case _WebSocketOpcode.PING: | |
| 418 _eventSink.add(new _WebSocketPing()); | |
| 419 break; | |
| 420 case _WebSocketOpcode.PONG: | |
| 421 _eventSink.add(new _WebSocketPong()); | |
| 422 break; | |
| 423 } | |
| 424 _prepareForNextFrame(); | |
| 425 } else { | |
| 426 _messageFrameEnd(); | |
| 427 } | |
| 428 } else { | |
| 429 _state = PAYLOAD; | |
| 430 } | |
| 431 } | |
| 432 | |
| 433 void _messageFrameEnd() { | |
| 434 if (_fin) { | |
| 435 switch (_currentMessageType) { | |
| 436 case _WebSocketMessageType.TEXT: | |
| 437 _eventSink.add(UTF8.decode(_payload.takeBytes())); | |
| 438 break; | |
| 439 case _WebSocketMessageType.BINARY: | |
| 440 _eventSink.add(_payload.takeBytes()); | |
| 441 break; | |
| 442 } | |
| 443 _currentMessageType = _WebSocketMessageType.NONE; | |
| 444 } | |
| 445 _prepareForNextFrame(); | |
| 446 } | |
| 447 | |
| 448 void _controlFrameEnd() { | |
| 449 switch (_opcode) { | |
| 450 case _WebSocketOpcode.CLOSE: | |
| 451 closeCode = _WebSocketStatus.NO_STATUS_RECEIVED; | |
| 452 var payload = _payload.takeBytes(); | |
| 453 if (payload.length > 0) { | |
| 454 if (payload.length == 1) { | |
| 455 throw new CompatibleWebSocketException("Protocol error"); | |
| 456 } | |
| 457 closeCode = payload[0] << 8 | payload[1]; | |
| 458 if (closeCode == _WebSocketStatus.NO_STATUS_RECEIVED) { | |
| 459 throw new CompatibleWebSocketException("Protocol error"); | |
| 460 } | |
| 461 if (payload.length > 2) { | |
| 462 closeReason = UTF8.decode(payload.sublist(2)); | |
| 463 } | |
| 464 } | |
| 465 _state = CLOSED; | |
| 466 _eventSink.close(); | |
| 467 break; | |
| 468 | |
| 469 case _WebSocketOpcode.PING: | |
| 470 _eventSink.add(new _WebSocketPing(_payload.takeBytes())); | |
| 471 break; | |
| 472 | |
| 473 case _WebSocketOpcode.PONG: | |
| 474 _eventSink.add(new _WebSocketPong(_payload.takeBytes())); | |
| 475 break; | |
| 476 } | |
| 477 _prepareForNextFrame(); | |
| 478 } | |
| 479 | |
| 480 bool _isControlFrame() { | |
| 481 return _opcode == _WebSocketOpcode.CLOSE || | |
| 482 _opcode == _WebSocketOpcode.PING || | |
| 483 _opcode == _WebSocketOpcode.PONG; | |
| 484 } | |
| 485 | |
| 486 void _prepareForNextFrame() { | |
| 487 if (_state != CLOSED && _state != FAILURE) _state = START; | |
| 488 _fin = false; | |
| 489 _opcode = -1; | |
| 490 _len = -1; | |
| 491 _remainingLenBytes = -1; | |
| 492 _remainingMaskingKeyBytes = 4; | |
| 493 _remainingPayloadBytes = -1; | |
| 494 _unmaskingIndex = 0; | |
| 495 } | |
| 496 } | |
| 497 | |
| 498 | |
| 499 class _WebSocketPing { | |
| 500 final List<int> payload; | |
| 501 _WebSocketPing([this.payload = null]); | |
| 502 } | |
| 503 | |
| 504 | |
| 505 class _WebSocketPong { | |
| 506 final List<int> payload; | |
| 507 _WebSocketPong([this.payload = null]); | |
| 508 } | |
| 509 | |
| 510 // TODO(ajohnsen): Make this transformer reusable. | |
| 511 class _WebSocketOutgoingTransformer implements StreamTransformer, EventSink { | |
| 512 final _WebSocketImpl webSocket; | |
| 513 EventSink _eventSink; | |
| 514 | |
| 515 _WebSocketOutgoingTransformer(this.webSocket); | |
| 516 | |
| 517 Stream bind(Stream stream) { | |
| 518 return new Stream.eventTransformed( | |
| 519 stream, | |
| 520 (EventSink eventSink) { | |
| 521 if (_eventSink != null) { | |
| 522 throw new StateError("WebSocket transformer already used"); | |
| 523 } | |
| 524 _eventSink = eventSink; | |
| 525 return this; | |
| 526 }); | |
| 527 } | |
| 528 | |
| 529 void add(message) { | |
| 530 if (message is _WebSocketPong) { | |
| 531 addFrame(_WebSocketOpcode.PONG, message.payload); | |
| 532 return; | |
| 533 } | |
| 534 if (message is _WebSocketPing) { | |
| 535 addFrame(_WebSocketOpcode.PING, message.payload); | |
| 536 return; | |
| 537 } | |
| 538 List<int> data; | |
| 539 int opcode; | |
| 540 if (message != null) { | |
| 541 if (message is String) { | |
| 542 opcode = _WebSocketOpcode.TEXT; | |
| 543 data = UTF8.encode(message); | |
| 544 } else { | |
| 545 if (message is !List<int>) { | |
| 546 throw new ArgumentError(message); | |
| 547 } | |
| 548 opcode = _WebSocketOpcode.BINARY; | |
| 549 data = message; | |
| 550 } | |
| 551 } else { | |
| 552 opcode = _WebSocketOpcode.TEXT; | |
| 553 } | |
| 554 addFrame(opcode, data); | |
| 555 } | |
| 556 | |
| 557 void addError(Object error, [StackTrace stackTrace]) => | |
| 558 _eventSink.addError(error, stackTrace); | |
| 559 | |
| 560 void close() { | |
| 561 int code = webSocket._outCloseCode; | |
| 562 String reason = webSocket._outCloseReason; | |
| 563 List<int> data; | |
| 564 if (code != null) { | |
| 565 data = new List<int>(); | |
| 566 data.add((code >> 8) & 0xFF); | |
| 567 data.add(code & 0xFF); | |
| 568 if (reason != null) { | |
| 569 data.addAll(UTF8.encode(reason)); | |
| 570 } | |
| 571 } | |
| 572 addFrame(_WebSocketOpcode.CLOSE, data); | |
| 573 _eventSink.close(); | |
| 574 } | |
| 575 | |
| 576 void addFrame(int opcode, List<int> data) => | |
| 577 createFrame(opcode, data, webSocket._serverSide).forEach(_eventSink.add); | |
| 578 | |
| 579 static Iterable createFrame(int opcode, List<int> data, bool serverSide) { | |
| 580 bool mask = !serverSide; // Masking not implemented for server. | |
| 581 int dataLength = data == null ? 0 : data.length; | |
| 582 // Determine the header size. | |
| 583 int headerSize = (mask) ? 6 : 2; | |
| 584 if (dataLength > 65535) { | |
| 585 headerSize += 8; | |
| 586 } else if (dataLength > 125) { | |
| 587 headerSize += 2; | |
| 588 } | |
| 589 Uint8List header = new Uint8List(headerSize); | |
| 590 int index = 0; | |
| 591 // Set FIN and opcode. | |
| 592 header[index++] = 0x80 | opcode; | |
| 593 // Determine size and position of length field. | |
| 594 int lengthBytes = 1; | |
| 595 int firstLengthByte = 1; | |
| 596 if (dataLength > 65535) { | |
| 597 header[index++] = 127; | |
| 598 lengthBytes = 8; | |
| 599 } else if (dataLength > 125) { | |
| 600 header[index++] = 126; | |
| 601 lengthBytes = 2; | |
| 602 } | |
| 603 // Write the length in network byte order into the header. | |
| 604 for (int i = 0; i < lengthBytes; i++) { | |
| 605 header[index++] = dataLength >> (((lengthBytes - 1) - i) * 8) & 0xFF; | |
| 606 } | |
| 607 if (mask) { | |
| 608 header[1] |= 1 << 7; | |
| 609 var maskBytes = [_random.nextInt(256), _random.nextInt(256), | |
| 610 _random.nextInt(256), _random.nextInt(256)]; | |
| 611 header.setRange(index, index + 4, maskBytes); | |
| 612 index += 4; | |
| 613 if (data != null) { | |
| 614 Uint8List list; | |
| 615 // If this is a text message just do the masking inside the | |
| 616 // encoded data. | |
| 617 if (opcode == _WebSocketOpcode.TEXT && data is Uint8List) { | |
| 618 list = data; | |
| 619 } else { | |
| 620 if (data is Uint8List) { | |
| 621 list = new Uint8List.fromList(data); | |
| 622 } else { | |
| 623 list = new Uint8List(data.length); | |
| 624 for (int i = 0; i < data.length; i++) { | |
| 625 if (data[i] < 0 || 255 < data[i]) { | |
| 626 throw new ArgumentError( | |
| 627 "List element is not a byte value " | |
| 628 "(value ${data[i]} at index $i)"); | |
| 629 } | |
| 630 list[i] = data[i]; | |
| 631 } | |
| 632 } | |
| 633 } | |
| 634 const int BLOCK_SIZE = 16; | |
| 635 int blockCount = list.length ~/ BLOCK_SIZE; | |
| 636 if (blockCount > 0) { | |
| 637 // Create mask block. | |
| 638 int mask = 0; | |
| 639 for (int i = 3; i >= 0; i--) { | |
| 640 mask = (mask << 8) | maskBytes[i]; | |
| 641 } | |
| 642 Int32x4 blockMask = new Int32x4(mask, mask, mask, mask); | |
| 643 Int32x4List blockBuffer = new Int32x4List.view( | |
| 644 list.buffer, 0, blockCount); | |
| 645 for (int i = 0; i < blockBuffer.length; i++) { | |
| 646 blockBuffer[i] ^= blockMask; | |
| 647 } | |
| 648 } | |
| 649 // Handle end. | |
| 650 for (int i = blockCount * BLOCK_SIZE; i < list.length; i++) { | |
| 651 list[i] ^= maskBytes[i & 3]; | |
| 652 } | |
| 653 data = list; | |
| 654 } | |
| 655 } | |
| 656 assert(index == headerSize); | |
| 657 if (data == null) { | |
| 658 return [header]; | |
| 659 } else { | |
| 660 return [header, data]; | |
| 661 } | |
| 662 } | |
| 663 } | |
| 664 | |
| 665 | |
| 666 class _WebSocketConsumer implements StreamConsumer { | |
| 667 final _WebSocketImpl webSocket; | |
| 668 final StreamSink<List<int>> sink; | |
| 669 StreamController _controller; | |
| 670 StreamSubscription _subscription; | |
| 671 bool _issuedPause = false; | |
| 672 bool _closed = false; | |
| 673 Completer _closeCompleter = new Completer(); | |
| 674 Completer _completer; | |
| 675 | |
| 676 _WebSocketConsumer(this.webSocket, this.sink); | |
| 677 | |
| 678 void _onListen() { | |
| 679 if (_subscription != null) { | |
| 680 _subscription.cancel(); | |
| 681 } | |
| 682 } | |
| 683 | |
| 684 void _onPause() { | |
| 685 if (_subscription != null) { | |
| 686 _subscription.pause(); | |
| 687 } else { | |
| 688 _issuedPause = true; | |
| 689 } | |
| 690 } | |
| 691 | |
| 692 void _onResume() { | |
| 693 if (_subscription != null) { | |
| 694 _subscription.resume(); | |
| 695 } else { | |
| 696 _issuedPause = false; | |
| 697 } | |
| 698 } | |
| 699 | |
| 700 void _cancel() { | |
| 701 if (_subscription != null) { | |
| 702 var subscription = _subscription; | |
| 703 _subscription = null; | |
| 704 subscription.cancel(); | |
| 705 } | |
| 706 } | |
| 707 | |
| 708 _ensureController() { | |
| 709 if (_controller != null) return; | |
| 710 _controller = new StreamController(sync: true, | |
| 711 onPause: _onPause, | |
| 712 onResume: _onResume, | |
| 713 onCancel: _onListen); | |
| 714 var stream = _controller.stream.transform( | |
| 715 new _WebSocketOutgoingTransformer(webSocket)); | |
| 716 sink.addStream(stream) | |
| 717 .then((_) { | |
| 718 _done(); | |
| 719 _closeCompleter.complete(webSocket); | |
| 720 }, onError: (error, StackTrace stackTrace) { | |
| 721 _closed = true; | |
| 722 _cancel(); | |
| 723 if (error is ArgumentError) { | |
| 724 if (!_done(error, stackTrace)) { | |
| 725 _closeCompleter.completeError(error, stackTrace); | |
| 726 } | |
| 727 } else { | |
| 728 _done(); | |
| 729 _closeCompleter.complete(webSocket); | |
| 730 } | |
| 731 }); | |
| 732 } | |
| 733 | |
| 734 bool _done([error, StackTrace stackTrace]) { | |
| 735 if (_completer == null) return false; | |
| 736 if (error != null) { | |
| 737 _completer.completeError(error, stackTrace); | |
| 738 } else { | |
| 739 _completer.complete(webSocket); | |
| 740 } | |
| 741 _completer = null; | |
| 742 return true; | |
| 743 } | |
| 744 | |
| 745 Future addStream(var stream) { | |
| 746 if (_closed) { | |
| 747 stream.listen(null).cancel(); | |
| 748 return new Future.value(webSocket); | |
| 749 } | |
| 750 _ensureController(); | |
| 751 _completer = new Completer(); | |
| 752 _subscription = stream.listen( | |
| 753 (data) { | |
| 754 _controller.add(data); | |
| 755 }, | |
| 756 onDone: _done, | |
| 757 onError: _done, | |
| 758 cancelOnError: true); | |
| 759 if (_issuedPause) { | |
| 760 _subscription.pause(); | |
| 761 _issuedPause = false; | |
| 762 } | |
| 763 return _completer.future; | |
| 764 } | |
| 765 | |
| 766 Future close() { | |
| 767 _ensureController(); | |
| 768 Future closeSocket() { | |
| 769 return sink.close().catchError((_) {}).then((_) => webSocket); | |
| 770 } | |
| 771 _controller.close(); | |
| 772 return _closeCompleter.future.then((_) => closeSocket()); | |
| 773 } | |
| 774 | |
| 775 void add(data) { | |
| 776 if (_closed) return; | |
| 777 _ensureController(); | |
| 778 _controller.add(data); | |
| 779 } | |
| 780 | |
| 781 void closeSocket() { | |
| 782 _closed = true; | |
| 783 _cancel(); | |
| 784 close(); | |
| 785 } | |
| 786 } | |
| 787 | |
| 788 | |
| 789 class _WebSocketImpl extends Stream implements CompatibleWebSocket { | |
| 790 StreamController _controller; | |
| 791 StreamSubscription _subscription; | |
| 792 StreamController _sink; | |
| 793 | |
| 794 final bool _serverSide; | |
| 795 int _readyState = _WebSocketState.CONNECTING; | |
| 796 bool _writeClosed = false; | |
| 797 int _closeCode; | |
| 798 String _closeReason; | |
| 799 Duration _pingInterval; | |
| 800 Timer _pingTimer; | |
| 801 _WebSocketConsumer _consumer; | |
| 802 | |
| 803 int _outCloseCode; | |
| 804 String _outCloseReason; | |
| 805 Timer _closeTimer; | |
| 806 | |
| 807 _WebSocketImpl._fromSocket(Stream<List<int>> stream, | |
| 808 StreamSink<List<int>> sink, [this._serverSide = false]) { | |
| 809 _consumer = new _WebSocketConsumer(this, sink); | |
| 810 _sink = new StreamController(); | |
| 811 _sink.stream.pipe(_consumer); | |
| 812 _readyState = _WebSocketState.OPEN; | |
| 813 | |
| 814 var transformer = new _WebSocketProtocolTransformer(_serverSide); | |
| 815 _subscription = stream.transform(transformer).listen( | |
| 816 (data) { | |
| 817 if (data is _WebSocketPing) { | |
| 818 if (!_writeClosed) _consumer.add(new _WebSocketPong(data.payload)); | |
| 819 } else if (data is _WebSocketPong) { | |
| 820 // Simply set pingInterval, as it'll cancel any timers. | |
| 821 pingInterval = _pingInterval; | |
| 822 } else { | |
| 823 _controller.add(data); | |
| 824 } | |
| 825 }, | |
| 826 onError: (error) { | |
| 827 if (_closeTimer != null) _closeTimer.cancel(); | |
| 828 if (error is FormatException) { | |
| 829 _close(_WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA); | |
| 830 } else { | |
| 831 _close(_WebSocketStatus.PROTOCOL_ERROR); | |
| 832 } | |
| 833 _controller.close(); | |
| 834 }, | |
| 835 onDone: () { | |
| 836 if (_closeTimer != null) _closeTimer.cancel(); | |
| 837 if (_readyState == _WebSocketState.OPEN) { | |
| 838 _readyState = _WebSocketState.CLOSING; | |
| 839 if (!_isReservedStatusCode(transformer.closeCode)) { | |
| 840 _close(transformer.closeCode); | |
| 841 } else { | |
| 842 _close(); | |
| 843 } | |
| 844 _readyState = _WebSocketState.CLOSED; | |
| 845 } | |
| 846 _closeCode = transformer.closeCode; | |
| 847 _closeReason = transformer.closeReason; | |
| 848 _controller.close(); | |
| 849 }, | |
| 850 cancelOnError: true); | |
| 851 _subscription.pause(); | |
| 852 _controller = new StreamController(sync: true, | |
| 853 onListen: _subscription.resume, | |
| 854 onPause: _subscription.pause, | |
| 855 onResume: _subscription.resume); | |
| 856 } | |
| 857 | |
| 858 StreamSubscription listen(void onData(message), | |
| 859 {Function onError, | |
| 860 void onDone(), | |
| 861 bool cancelOnError}) { | |
| 862 return _controller.stream.listen(onData, | |
| 863 onError: onError, | |
| 864 onDone: onDone, | |
| 865 cancelOnError: cancelOnError); | |
| 866 } | |
| 867 | |
| 868 Duration get pingInterval => _pingInterval; | |
| 869 | |
| 870 void set pingInterval(Duration interval) { | |
| 871 if (_writeClosed) return; | |
| 872 if (_pingTimer != null) _pingTimer.cancel(); | |
| 873 _pingInterval = interval; | |
| 874 | |
| 875 if (_pingInterval == null) return; | |
| 876 | |
| 877 _pingTimer = new Timer(_pingInterval, () { | |
| 878 if (_writeClosed) return; | |
| 879 _consumer.add(new _WebSocketPing()); | |
| 880 _pingTimer = new Timer(_pingInterval, () { | |
| 881 // No pong received. | |
| 882 _close(_WebSocketStatus.GOING_AWAY); | |
| 883 }); | |
| 884 }); | |
| 885 } | |
| 886 | |
| 887 int get closeCode => _closeCode; | |
| 888 String get closeReason => _closeReason; | |
| 889 | |
| 890 void add(data) => _sink.add(data); | |
| 891 void addError(error, [StackTrace stackTrace]) => | |
| 892 _sink.addError(error, stackTrace); | |
| 893 Future addStream(Stream stream) => _sink.addStream(stream); | |
| 894 Future get done => _sink.done; | |
| 895 | |
| 896 Future close([int code, String reason]) { | |
| 897 if (_isReservedStatusCode(code)) { | |
| 898 throw new CompatibleWebSocketException("Reserved status code $code"); | |
| 899 } | |
| 900 if (_outCloseCode == null) { | |
| 901 _outCloseCode = code; | |
| 902 _outCloseReason = reason; | |
| 903 } | |
| 904 if (_closeTimer == null && !_controller.isClosed) { | |
| 905 // When closing the web-socket, we no longer accept data. | |
| 906 _closeTimer = new Timer(const Duration(seconds: 5), () { | |
| 907 _subscription.cancel(); | |
| 908 _controller.close(); | |
| 909 }); | |
| 910 } | |
| 911 return _sink.close(); | |
| 912 } | |
| 913 | |
| 914 void _close([int code, String reason]) { | |
| 915 if (_writeClosed) return; | |
| 916 if (_outCloseCode == null) { | |
| 917 _outCloseCode = code; | |
| 918 _outCloseReason = reason; | |
| 919 } | |
| 920 _writeClosed = true; | |
| 921 _consumer.closeSocket(); | |
| 922 } | |
| 923 | |
| 924 static bool _isReservedStatusCode(int code) { | |
| 925 return code != null && | |
| 926 (code < _WebSocketStatus.NORMAL_CLOSURE || | |
| 927 code == _WebSocketStatus.RESERVED_1004 || | |
| 928 code == _WebSocketStatus.NO_STATUS_RECEIVED || | |
| 929 code == _WebSocketStatus.ABNORMAL_CLOSURE || | |
| 930 (code > _WebSocketStatus.INTERNAL_SERVER_ERROR && | |
| 931 code < _WebSocketStatus.RESERVED_1015) || | |
| 932 (code >= _WebSocketStatus.RESERVED_1015 && | |
| 933 code < 3000)); | |
| 934 } | |
| 935 } | |
| 936 | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 248463004:
Revert "Add a non-dart:io WebSocket implementation to http_parser." (Closed)
Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart