use FixedQueue

tsctx · web-flow · commit 0e415b29ce4e · 2024-05-21T22:59:26.000+09:00
diff --git a/lib/web/websocket/sender.js b/lib/web/websocket/sender.js
@@ -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}
@@ -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)
@@ -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()
@@ -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
@@ -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
   }
 }
