Skip to content

Commit

Permalink
use FixedQueue
Browse files Browse the repository at this point in the history
  • Loading branch information
tsctx committed May 21, 2024
1 parent aacd26c commit a88052b
Showing 1 changed file with 13 additions and 26 deletions.
39 changes: 13 additions & 26 deletions lib/web/websocket/sender.js
Original file line number Diff line number Diff line change
Expand Up @@ -2,27 +2,23 @@

const { WebsocketFrameSend } = require('./frame')
const { opcodes, sendHints } = require('./constants')
const FixedQueue = require('../../dispatcher/fixed-queue')

/** @type {typeof Uint8Array} */
const FastBuffer = Buffer[Symbol.species]

/**
* @typedef {object} SendQueueNode
* @property {SendQueueNode | null} next
* @property {Promise<void> | null} promise
* @property {((...args: any[]) => any)} callback
* @property {Buffer | null} frame
*/

class SendQueue {
/**
* @type {SendQueueNode | null}
* @type {FixedQueue | null}
*/
#head = null
/**
* @type {SendQueueNode | null}
*/
#tail = null
#queue = null

/**
* @type {boolean}
Expand All @@ -45,22 +41,20 @@ class SendQueue {
} else {
/** @type {SendQueueNode} */
const node = {
next: null,
promise: null,
callback: cb,
frame
}
if (this.#tail !== null) {
this.#tail.next = node
if (this.#queue === null) {
this.#queue = new FixedQueue()
}
this.#tail = node
this.#queue.push(node)
}
return
}

/** @type {SendQueueNode} */
const node = {
next: null,
promise: item.arrayBuffer().then((ab) => {
node.promise = null
node.frame = createFrame(ab, hint)
Expand All @@ -69,15 +63,11 @@ class SendQueue {
frame: null
}

if (this.#head === null) {
this.#head = node
if (this.#queue === null) {
this.#queue = new FixedQueue()
}

if (this.#tail === null) {
this.#tail = node
} else {
this.#tail.next = node
}
this.#queue.push(node)

if (!this.#running) {
this.#run()
Expand All @@ -86,9 +76,10 @@ class SendQueue {

async #run () {
this.#running = true
/** @type {SendQueueNode | null} */
let node = this.#head
while (node !== null) {
/** @type {FixedQueue} */
const queue = this.#queue
while (!queue.isEmpty()) {
const node = queue.shift()
// wait pending promise
if (node.promise !== null) {
await node.promise
Expand All @@ -97,11 +88,7 @@ class SendQueue {
this.#socket.write(node.frame, node.callback)
// cleanup
node.callback = node.frame = null
// set next
node = node.next
}
this.#head = null
this.#tail = null
this.#running = false
}
}
Expand Down

0 comments on commit a88052b

Please sign in to comment.