* Wait for the next frame matching `predicate`. Drains queued frames first; * the first matching frame is consumed and returned. Times out cleanly.
(predicate: FrameWaiter, timeoutMs: number)
| 122 | * the first matching frame is consumed and returned. Times out cleanly. |
| 123 | */ |
| 124 | waitForFrame(predicate: FrameWaiter, timeoutMs: number): Promise<AnyFrame> { |
| 125 | return new Promise((resolve, reject) => { |
| 126 | // Drain queue. |
| 127 | for (let i = 0; i < this._queue.length; i++) { |
| 128 | const frame = this._queue[i]; |
| 129 | if (frame === undefined) continue; |
| 130 | if (predicate(frame)) { |
| 131 | this._queue.splice(i, 1); |
| 132 | resolve(frame); |
| 133 | return; |
| 134 | } |
| 135 | } |
| 136 | if (this._closed) { |
| 137 | reject(new Error(`ws closed before matching frame arrived (code=${this._closeReason?.code})`)); |
| 138 | return; |
| 139 | } |
| 140 | const waiter: PendingWaiter = { |
| 141 | match: predicate, |
| 142 | resolve: (f) => { |
| 143 | if (waiter.timer) clearTimeout(waiter.timer); |
| 144 | resolve(f); |
| 145 | }, |
| 146 | reject: (e) => { |
| 147 | if (waiter.timer) clearTimeout(waiter.timer); |
| 148 | reject(e); |
| 149 | }, |
| 150 | }; |
| 151 | waiter.timer = setTimeout(() => { |
| 152 | const idx = this._waiters.indexOf(waiter); |
| 153 | if (idx >= 0) this._waiters.splice(idx, 1); |
| 154 | reject(new Error(`waitForFrame timed out after ${timeoutMs}ms`)); |
| 155 | }, timeoutMs); |
| 156 | waiter.timer.unref?.(); |
| 157 | this._waiters.push(waiter); |
| 158 | }); |
| 159 | } |
| 160 | |
| 161 | /** Send a control message and wait for its `ack` (matched by `id`). */ |
| 162 | async sendAndAwaitAck(frame: { type: string; id: string; payload: unknown }, timeoutMs: number): Promise<AnyFrame> { |