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 cc4f3de commit 0e415b2
Showing 1 changed file with 12 additions and 30 deletions.
42 changes: 12 additions & 30 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
* @typedef {object} QueueNode
* @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 @@ -43,24 +39,19 @@ class SendQueue {
// fast-path
this.#socket.write(frame, cb)
} else {
/** @type {SendQueueNode} */
/** @type {QueueNode} */
const node = {
next: null,
promise: null,
callback: cb,
frame
}
if (this.#tail !== null) {
this.#tail.next = node
}
this.#tail = node
(this.#queue ??= new FixedQueue()).push(node)
}
return
}

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

if (this.#tail === null) {
this.#tail = node
}

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

if (!this.#running) {
this.#run()
Expand All @@ -84,9 +69,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 @@ -95,11 +81,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 0e415b2

Please sign in to comment.