-
Notifications
You must be signed in to change notification settings - Fork 30.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
PR-URL: #53073 Reviewed-By: Marco Ippolito <marcoippolito54@gmail.com> Reviewed-By: Matthew Aitken <maitken033380023@gmail.com> Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com> Reviewed-By: Rafael Gonzaga <rafael.nunu@hotmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
- Loading branch information
1 parent
494fa54
commit b941992
Showing
12 changed files
with
580 additions
and
135 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| 'use strict' | ||
|
|
||
| const { createInflateRaw, Z_DEFAULT_WINDOWBITS } = require('node:zlib') | ||
| const { isValidClientWindowBits } = require('./util') | ||
|
|
||
| const tail = Buffer.from([0x00, 0x00, 0xff, 0xff]) | ||
| const kBuffer = Symbol('kBuffer') | ||
| const kLength = Symbol('kLength') | ||
|
|
||
| class PerMessageDeflate { | ||
| /** @type {import('node:zlib').InflateRaw} */ | ||
| #inflate | ||
|
|
||
| #options = {} | ||
|
|
||
| constructor (extensions) { | ||
| this.#options.serverNoContextTakeover = extensions.has('server_no_context_takeover') | ||
| this.#options.serverMaxWindowBits = extensions.get('server_max_window_bits') | ||
| } | ||
|
|
||
| decompress (chunk, fin, callback) { | ||
| // An endpoint uses the following algorithm to decompress a message. | ||
| // 1. Append 4 octets of 0x00 0x00 0xff 0xff to the tail end of the | ||
| // payload of the message. | ||
| // 2. Decompress the resulting data using DEFLATE. | ||
|
|
||
| if (!this.#inflate) { | ||
| let windowBits = Z_DEFAULT_WINDOWBITS | ||
|
|
||
| if (this.#options.serverMaxWindowBits) { // empty values default to Z_DEFAULT_WINDOWBITS | ||
| if (!isValidClientWindowBits(this.#options.serverMaxWindowBits)) { | ||
| callback(new Error('Invalid server_max_window_bits')) | ||
| return | ||
| } | ||
|
|
||
| windowBits = Number.parseInt(this.#options.serverMaxWindowBits) | ||
| } | ||
|
|
||
| this.#inflate = createInflateRaw({ windowBits }) | ||
| this.#inflate[kBuffer] = [] | ||
| this.#inflate[kLength] = 0 | ||
|
|
||
| this.#inflate.on('data', (data) => { | ||
| this.#inflate[kBuffer].push(data) | ||
| this.#inflate[kLength] += data.length | ||
| }) | ||
|
|
||
| this.#inflate.on('error', (err) => { | ||
| this.#inflate = null | ||
| callback(err) | ||
| }) | ||
| } | ||
|
|
||
| this.#inflate.write(chunk) | ||
| if (fin) { | ||
| this.#inflate.write(tail) | ||
| } | ||
|
|
||
| this.#inflate.flush(() => { | ||
| const full = Buffer.concat(this.#inflate[kBuffer], this.#inflate[kLength]) | ||
|
|
||
| this.#inflate[kBuffer].length = 0 | ||
| this.#inflate[kLength] = 0 | ||
|
|
||
| callback(null, full) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| module.exports = { PerMessageDeflate } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| 'use strict' | ||
|
|
||
| const { WebsocketFrameSend } = require('./frame') | ||
| const { opcodes, sendHints } = require('./constants') | ||
|
|
||
| /** @type {Uint8Array} */ | ||
| const FastBuffer = Buffer[Symbol.species] | ||
|
|
||
| class SendQueue { | ||
| #queued = new Set() | ||
| #size = 0 | ||
|
|
||
| /** @type {import('net').Socket} */ | ||
| #socket | ||
|
|
||
| constructor (socket) { | ||
| this.#socket = socket | ||
| } | ||
|
|
||
| add (item, cb, hint) { | ||
| if (hint !== sendHints.blob) { | ||
| const data = clone(item, hint) | ||
|
|
||
| if (this.#size === 0) { | ||
| this.#dispatch(data, cb, hint) | ||
| } else { | ||
| this.#queued.add([data, cb, true, hint]) | ||
| this.#size++ | ||
|
|
||
| this.#run() | ||
| } | ||
|
|
||
| return | ||
| } | ||
|
|
||
| const promise = item.arrayBuffer() | ||
| const queue = [null, cb, false, hint] | ||
| promise.then((ab) => { | ||
| queue[0] = clone(ab, hint) | ||
| queue[2] = true | ||
|
|
||
| this.#run() | ||
| }) | ||
|
|
||
| this.#queued.add(queue) | ||
| this.#size++ | ||
| } | ||
|
|
||
| #run () { | ||
| for (const queued of this.#queued) { | ||
| const [data, cb, done, hint] = queued | ||
|
|
||
| if (!done) return | ||
|
|
||
| this.#queued.delete(queued) | ||
| this.#size-- | ||
|
|
||
| this.#dispatch(data, cb, hint) | ||
| } | ||
| } | ||
|
|
||
| #dispatch (data, cb, hint) { | ||
| const frame = new WebsocketFrameSend() | ||
| const opcode = hint === sendHints.string ? opcodes.TEXT : opcodes.BINARY | ||
|
|
||
| frame.frameData = data | ||
| const buffer = frame.createFrame(opcode) | ||
|
|
||
| this.#socket.write(buffer, cb) | ||
| } | ||
| } | ||
|
|
||
| function clone (data, hint) { | ||
| switch (hint) { | ||
| case sendHints.string: | ||
| return Buffer.from(data) | ||
| case sendHints.arrayBuffer: | ||
| case sendHints.blob: | ||
| return new FastBuffer(data) | ||
| case sendHints.typedArray: | ||
| return Buffer.copyBytesFrom(data) | ||
| } | ||
| } | ||
|
|
||
| module.exports = { SendQueue } |
Oops, something went wrong.