diff --git a/lib/web/websocket/connection.js b/lib/web/websocket/connection.js index 4ecc8a195fc..d6f22bc2c6b 100644 --- a/lib/web/websocket/connection.js +++ b/lib/web/websocket/connection.js @@ -206,7 +206,7 @@ function establishWebSocketConnection (url, protocols, client, handler, options) } } - response.socket.on('data', handler.onSocketData) + response.socket.on('readable', handler.onSocketReadable) response.socket.on('close', handler.onSocketClose) response.socket.on('error', handler.onSocketError) diff --git a/lib/web/websocket/stream/websocketstream.js b/lib/web/websocket/stream/websocketstream.js index 383fd0bf7aa..5063a5b2f25 100644 --- a/lib/web/websocket/stream/websocketstream.js +++ b/lib/web/websocket/stream/websocketstream.js @@ -47,12 +47,8 @@ class WebSocketStream { onConnectionEstablished: (response, extensions) => this.#onConnectionEstablished(response, extensions), onMessage: (opcode, data) => this.#onMessage(opcode, data), onParserError: (err) => failWebsocketConnection(this.#handler, null, err.message), - onParserDrain: () => this.#handler.socket.resume(), - onSocketData: (chunk) => { - if (!this.#parser.write(chunk)) { - this.#handler.socket.pause() - } - }, + onParserDrain: () => this.#readFromSocket(), + onSocketReadable: () => this.#readFromSocket(), onSocketError: (err) => { this.#handler.readyState = states.CLOSING @@ -117,7 +113,7 @@ class WebSocketStream { this.#closedPromise = Promise.withResolvers() // 7. Apply backpressure to the WebSocket. - // TODO + // Bytes are not read until the readable stream has demand. // 8. If options [" signal "] exists , if (options.signal != null) { @@ -195,6 +191,29 @@ class WebSocketStream { // 3. Close the WebSocket with this , code , and reason . closeWebSocketConnection(this.#handler, code, reason, true) + this.#readFromSocket() + } + + #readFromSocket () { + const socket = this.#handler.socket + + if (socket == null || this.#parser == null || this.#parser.writableNeedDrain) { + return + } + + while ( + this.#handler.readyState === states.CLOSING || + ( + this.#handler.readyState === states.OPEN && + this.#readableStreamController?.desiredSize > 0 + ) + ) { + const chunk = socket.read() + + if (chunk === null || !this.#parser.write(chunk)) { + return + } + } } #write (chunk) { @@ -291,6 +310,7 @@ class WebSocketStream { start: (controller) => { this.#readableStreamController = controller }, + pull: () => this.#readFromSocket(), cancel: (reason) => this.#cancel(reason) }) @@ -301,7 +321,10 @@ class WebSocketStream { // 13. Set up writable with writeAlgorithm , closeAlgorithm , and abortAlgorithm . const writable = new WritableStream({ write: (chunk) => this.#write(chunk), - close: () => closeWebSocketConnection(this.#handler, null, null), + close: () => { + closeWebSocketConnection(this.#handler, null, null) + this.#readFromSocket() + }, abort: (reason) => this.#closeUsingReason(reason) }) @@ -350,6 +373,7 @@ class WebSocketStream { this.#readableStreamController.enqueue(chunk) // 4. Apply backpressure to the WebSocket. + // The next socket read is gated on the controller's desired size. } /** @type {import('../websocket').Handler['onSocketClose']} */ @@ -441,6 +465,7 @@ class WebSocketStream { // 4. Close the WebSocket with stream , code , and reasonString . If this throws an exception, // discard code and reasonString and close the WebSocket with stream . closeWebSocketConnection(this.#handler, code, reasonString) + this.#readFromSocket() } // To cancel a WebSocketStream stream given reason , close using reason giving stream and reason . diff --git a/lib/web/websocket/websocket.js b/lib/web/websocket/websocket.js index e473a1bc491..dce2e1b8c6e 100644 --- a/lib/web/websocket/websocket.js +++ b/lib/web/websocket/websocket.js @@ -43,7 +43,7 @@ function getSocketAddress (socket) { * @property {(opcode: number, data: Buffer) => void} onMessage * @property {(error: Error) => void} onParserError * @property {() => void} onParserDrain - * @property {(chunk: Buffer) => void} onSocketData + * @property {() => void} onSocketReadable * @property {(err: Error) => void} onSocketError * @property {() => void} onSocketClose * @property {(body: Buffer) => void} onPing @@ -77,12 +77,8 @@ class WebSocket extends EventTarget { onConnectionEstablished: (response, extensions) => this.#onConnectionEstablished(response, extensions), onMessage: (opcode, data) => this.#onMessage(opcode, data), onParserError: (err) => failWebsocketConnection(this.#handler, null, err.message), - onParserDrain: () => this.#onParserDrain(), - onSocketData: (chunk) => { - if (!this.#parser.write(chunk)) { - this.#handler.socket.pause() - } - }, + onParserDrain: () => this.#readFromSocket(), + onSocketReadable: () => this.#readFromSocket(), onSocketError: (err) => { this.#handler.readyState = states.CLOSING @@ -521,6 +517,8 @@ class WebSocket extends EventTarget { } }) } + + this.#readFromSocket() } #onMessage (type, data) { @@ -564,8 +562,19 @@ class WebSocket extends EventTarget { }) } - #onParserDrain () { - this.#handler.socket.resume() + #readFromSocket () { + const socket = this.#handler.socket + + if (socket == null || this.#parser == null || this.#parser.writableNeedDrain) { + return + } + + let chunk + while ((chunk = socket.read()) !== null) { + if (!this.#parser.write(chunk)) { + return + } + } } /** diff --git a/test/web-platform-tests/expectation.json b/test/web-platform-tests/expectation.json index 001ed3ffe9e..5fe462501ff 100644 --- a/test/web-platform-tests/expectation.json +++ b/test/web-platform-tests/expectation.json @@ -39163,8 +39163,7 @@ "cases": [ { "name": "backpressure should be applied to received messages", - "success": false, - "message": "assert_greater_than_equal: data send should have taken at least 2 seconds expected a number greater than or equal to 1.8 but got 0.0826730728149414" + "success": true } ] },