Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(640)

Issues Search
    My Issues | Starred     Open | Closed | All

Unified Diff: sdk/lib/io/websocket_impl.dart

Issue 14696012: Make web-socket pass AutobanhTestSuite. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Created 7 years, 7 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« sdk/lib/io/websocket.dart ('K') | « sdk/lib/io/websocket.dart ('k') | sdk/lib/utf/utf_stream.dart » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: sdk/lib/io/websocket_impl.dart
diff --git a/sdk/lib/io/websocket_impl.dart b/sdk/lib/io/websocket_impl.dart
index 52587ed0306ac45ac6785fc993e6dd343bbc512b..3ca46cd51e3e048bb29496c3970f2f9af0b9f2c5 100644
--- a/sdk/lib/io/websocket_impl.dart
+++ b/sdk/lib/io/websocket_impl.dart
@@ -76,6 +76,9 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
switch (_state) {
case START:
_fin = (byte & 0x80) != 0;
Søren Gjesse 2013/05/08 12:15:40 Add comment // The RSV1, RSV2 bits RSV3 most be a
Anders Johnsen 2013/06/11 12:23:47 Done.
+ if ((byte & 0x70) != 0) {
+ throw new WebSocketException("Protocol error");
+ }
_opcode = (byte & 0xF);
switch (_opcode) {
case _WebSocketOpcode.CONTINUATION:
@@ -89,7 +92,15 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
throw new WebSocketException("Protocol error");
}
_currentMessageType = _WebSocketMessageType.TEXT;
- _buffer = new StringBuffer();
+ _controller = new StreamController();
+ _controller.stream
+ .transform(new Utf8DecoderTransformer(null))
+ .fold(new StringBuffer(), (buffer, str) => buffer..write(str))
+ .then((buffer) {
+ sink.add(buffer.toString());
+ }, onError: (error) {
+ sink.addError(error);
+ });
break;
case _WebSocketOpcode.BINARY:
@@ -97,7 +108,14 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
throw new WebSocketException("Protocol error");
}
_currentMessageType = _WebSocketMessageType.BINARY;
- _buffer = new _BufferList();
+ _controller = new StreamController();
+ _controller.stream
+ .fold(new _BufferList(), (buffer, data) => buffer..add(data))
+ .then((buffer) {
+ sink.add(buffer.readBytes());
+ }, onError: (error) {
+ sink.addError(error);
+ });
break;
case _WebSocketOpcode.CLOSE:
@@ -116,7 +134,7 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
case LEN_FIRST:
_masked = (byte & 0x80) != 0;
_len = byte & 0x7F;
- if (_isControlFrame() && _len > 126) {
+ if (_isControlFrame() && _len > 125) {
throw new WebSocketException("Protocol error");
}
if (_len < 126) {
@@ -183,30 +201,15 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
_controlFrameEnd(sink);
}
} else {
- switch (_currentMessageType) {
- case _WebSocketMessageType.NONE:
- throw new WebSocketException("Protocol error");
-
- case _WebSocketMessageType.TEXT:
- _buffer.write(_decodeString(
- buffer.sublist(index, index + payload)));
- index += payload;
- if (_remainingPayloadBytes == 0) {
- _messageFrameEnd(sink);
- }
- break;
-
- case _WebSocketMessageType.BINARY:
- _buffer.write(buffer.sublist(index, index + payload));
- index += payload;
- if (_remainingPayloadBytes == 0) {
- _messageFrameEnd(sink);
- }
- break;
-
- default:
+ if (_currentMessageType != _WebSocketMessageType.TEXT &&
+ _currentMessageType != _WebSocketMessageType.BINARY) {
throw new WebSocketException("Protocol error");
}
+ _controller.add(new Uint8List.view(buffer.buffer, index, payload));
+ index += payload;
+ if (_remainingPayloadBytes == 0) {
+ _messageFrameEnd(sink);
+ }
}
// Hack - as we always do index++ below.
@@ -255,10 +258,10 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
sink.close();
break;
case _WebSocketOpcode.PING:
- // TODO(ajohnsen): Handle ping.
+ sink.add(new _WebSocketPing());
break;
case _WebSocketOpcode.PONG:
- // TODO(ajohnsen): Handle pong.
+ sink.add(new _WebSocketPong());
break;
}
_prepareForNextFrame();
@@ -274,13 +277,13 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
if (_fin) {
switch (_currentMessageType) {
case _WebSocketMessageType.TEXT:
- sink.add(_buffer.toString());
+ _controller.close();
break;
case _WebSocketMessageType.BINARY:
- sink.add(_buffer.readBytes());
+ _controller.close();
break;
}
- _buffer = null;
+ _controller = null;
_currentMessageType = _WebSocketMessageType.NONE;
}
_prepareForNextFrame();
@@ -299,8 +302,7 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
throw new WebSocketException("Protocol error");
}
if (_controlPayload.length > 2) {
- closeReason = _decodeString(
- _controlPayload.sublist(2));
+ closeReason = _decodeUtf8Strict(_controlPayload.sublist(2));
}
}
_state = CLOSED;
@@ -308,11 +310,11 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
break;
case _WebSocketOpcode.PING:
- // TODO(ajohnsen): Handle ping.
+ sink.add(new _WebSocketPing(_controlPayload));
break;
case _WebSocketOpcode.PONG:
- // TODO(ajohnsen): Handle pong.
+ sink.add(new _WebSocketPong(_controlPayload));
break;
}
_prepareForNextFrame();
@@ -351,13 +353,25 @@ class _WebSocketProtocolTransformer extends StreamEventTransformer {
int _currentMessageType;
List<int> _controlPayload;
- var _buffer; // Either StringBuffer or _BufferList.
+ StreamController _controller;
int closeCode = WebSocketStatus.NO_STATUS_RECEIVED;
String closeReason = "";
}
+class _WebSocketPing {
+ final List<int> payload;
+ _WebSocketPing([this.payload = null]);
+}
+
+
+class _WebSocketPong {
+ final List<int> payload;
+ _WebSocketPong([this.payload = null]);
+}
+
+
class _WebSocketTransformerImpl implements WebSocketTransformer {
final StreamController<WebSocket> _controller =
new StreamController<WebSocket>();
@@ -434,6 +448,14 @@ class _WebSocketOutgoingTransformer extends StreamEventTransformer {
_WebSocketOutgoingTransformer(_WebSocketImpl this.webSocket);
void handleData(message, EventSink<List<int>> sink) {
+ if (message is _WebSocketPong) {
+ addFrame(_WebSocketOpcode.PONG, message.payload, sink);
+ return;
+ }
+ if (message is _WebSocketPing) {
+ addFrame(_WebSocketOpcode.PONG, message.payload, sink);
+ return;
+ }
List<int> data;
int opcode;
if (message != null) {
@@ -470,7 +492,11 @@ class _WebSocketOutgoingTransformer extends StreamEventTransformer {
}
void addFrame(int opcode, List<int> data, EventSink<List<int>> sink) {
- bool mask = !webSocket._serverSide; // Masking not implemented for server.
+ createFrame(opcode, data, webSocket._serverSide).forEach(sink.add);
+ }
+
+ static Iterator createFrame(int opcode, List<int> data, bool serverSide) {
+ bool mask = !serverSide; // Masking not implemented for server.
int dataLength = data == null ? 0 : data.length;
// Determine the header size.
int headerSize = (mask) ? 6 : 2;
@@ -511,9 +537,10 @@ class _WebSocketOutgoingTransformer extends StreamEventTransformer {
}
}
assert(index == headerSize);
- sink.add(header);
- if (data != null) {
- sink.add(data);
+ if (data == null) {
+ return [header];
+ } else {
+ return [header, data];
}
}
}
@@ -584,13 +611,18 @@ class _WebSocketConsumer implements StreamConsumer {
}
Future close() {
+ _ensureController();
Future closeSocket() {
return socket.close().then((_) => webSocket);
}
- if (_controller == null) return closeSocket();
_controller.close();
return _closeCompleter.future.then((_) => closeSocket());
}
+
+ void add(data) {
+ _ensureController();
+ _controller.add(data);
+ }
}
@@ -682,22 +714,34 @@ class _WebSocketImpl extends Stream implements WebSocket {
_WebSocketImpl._fromSocket(Socket this._socket,
[bool this._serverSide = false]) {
- _sink = new _StreamSinkImpl(new _WebSocketConsumer(this, _socket));
+ var consumer = new _WebSocketConsumer(this, _socket);
+ _sink = new _StreamSinkImpl(consumer);
_readyState = WebSocket.OPEN;
var transformer = new _WebSocketProtocolTransformer(_serverSide);
_socket.transform(transformer).listen(
(data) {
- _controller.add(data);
+ if (data is _WebSocketPing) {
+ consumer.add(new _WebSocketPong(data.payload));
+ } else if (data is _WebSocketPong) {
+ // TODO(ajohnsen): Notify pong?
+ } else {
+ _controller.add(data);
+ }
},
onError: (error) {
+ if (error is ArgumentError) {
+ close(WebSocketStatus.INVALID_FRAME_PAYLOAD_DATA);
+ } else {
+ close(WebSocketStatus.PROTOCOL_ERROR);
+ }
_controller.addError(error);
_controller.close();
},
onDone: () {
if (_readyState == WebSocket.OPEN) {
_readyState = WebSocket.CLOSING;
- if (transformer.closeCode != WebSocketStatus.NO_STATUS_RECEIVED) {
+ if (!_isReservedStatusCode(transformer.closeCode)) {
close(transformer.closeCode);
} else {
close();
@@ -735,9 +779,7 @@ class _WebSocketImpl extends Stream implements WebSocket {
Future close([int code, String reason]) {
if (!_writeClosed) {
- if (code == WebSocketStatus.RESERVED_1004 ||
- code == WebSocketStatus.NO_STATUS_RECEIVED ||
- code == WebSocketStatus.RESERVED_1015) {
+ if (_isReservedStatusCode(code)) {
throw new WebSocketException("Reserved status code $code");
}
_outCloseCode = code;
@@ -746,4 +788,16 @@ class _WebSocketImpl extends Stream implements WebSocket {
}
return _sink.close();
}
+
+ static bool _isReservedStatusCode(int code) {
+ return code != null &&
+ (code < WebSocketStatus.NORMAL_CLOSURE ||
+ code == WebSocketStatus.RESERVED_1004 ||
+ code == WebSocketStatus.NO_STATUS_RECEIVED ||
+ code == WebSocketStatus.ABNORMAL_CLOSURE ||
+ (code > WebSocketStatus.INTERNAL_SERVER_ERROR &&
+ code < WebSocketStatus.RESERVED_1015) ||
+ (code >= WebSocketStatus.RESERVED_1015 &&
+ code < 3000));
+ }
}
« sdk/lib/io/websocket.dart ('K') | « sdk/lib/io/websocket.dart ('k') | sdk/lib/utf/utf_stream.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698