diff --git a/lib/web/websocket/stream/websocketstream.js b/lib/web/websocket/stream/websocketstream.js index 383fd0bf7aa..8f135ba3517 100644 --- a/lib/web/websocket/stream/websocketstream.js +++ b/lib/web/websocket/stream/websocketstream.js @@ -47,7 +47,7 @@ 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(), + onParserDrain: () => this.#resumeSocketIfNeeded(), onSocketData: (chunk) => { if (!this.#parser.write(chunk)) { this.#handler.socket.pause() @@ -117,7 +117,7 @@ class WebSocketStream { this.#closedPromise = Promise.withResolvers() // 7. Apply backpressure to the WebSocket. - // TODO + // This is done when the socket becomes available. // 8. If options [" signal "] exists , if (options.signal != null) { @@ -195,6 +195,22 @@ class WebSocketStream { // 3. Close the WebSocket with this , code , and reason . closeWebSocketConnection(this.#handler, code, reason, true) + this.#resumeSocketIfNeeded() + } + + #resumeSocketIfNeeded () { + const socket = this.#handler.socket + + if (socket == null) { + return + } + + if ( + this.#handler.readyState === states.CLOSING || + (this.#readableStreamController?.desiredSize > 0 && !this.#parser.writableNeedDrain) + ) { + socket.resume() + } } #write (chunk) { @@ -257,6 +273,7 @@ class WebSocketStream { /** @type {import('../websocket').Handler['onConnectionEstablished']} */ #onConnectionEstablished (response, parsedExtensions) { this.#handler.socket = response.socket + this.#handler.socket.pause() // Get options from dispatcher options const maxFragments = this.#handler.controller.dispatcher?.webSocketOptions?.maxFragments @@ -291,6 +308,7 @@ class WebSocketStream { start: (controller) => { this.#readableStreamController = controller }, + pull: () => this.#resumeSocketIfNeeded(), cancel: (reason) => this.#cancel(reason) }) @@ -301,7 +319,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.#resumeSocketIfNeeded() + }, abort: (reason) => this.#closeUsingReason(reason) }) @@ -350,6 +371,9 @@ class WebSocketStream { this.#readableStreamController.enqueue(chunk) // 4. Apply backpressure to the WebSocket. + if (this.#readableStreamController.desiredSize <= 0) { + this.#handler.socket.pause() + } } /** @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.#resumeSocketIfNeeded() } // To cancel a WebSocketStream stream given reason , close using reason giving stream and reason . 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 } ] },