Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869import {Semaphore} from './semaphore'
/** * simple blocking queue, for turning streams into async pulls. * cribbed mostly from {@link https://github.com/ComFreek/async-playground} */export class BlockingQueue<T> { #sema: Semaphore #maxsize?: number #items: T[]
constructor(maxsize = 1000) { this.#sema = new Semaphore() this.#maxsize = maxsize ? maxsize : undefined this.#items = [] }
/** @returns the depth of the queue */ get depth(): number { return this.#items.length }
/** place one or more items on the queue, to be picked up by awaiters. */ prequeue(...elements: T[]) { for (const el of elements.reverse()) { if (this.#maxsize && this.#items.length >= this.#maxsize) { throw Error('out of room') }
this.#items.unshift(el) this.#sema.free() } }
/** place one or more items on the queue, to be picked up by awaiters. */ enqueue(...elements: T[]) { for (const el of elements) { if (this.#maxsize && this.#items.length >= this.#maxsize) { throw Error('out of room') }
this.#items.push(el) this.#sema.free() } }
/** * block while waiting for an item off the queue. * * @param signal - a signal to use for aborting the block. * @returns the item off the queue; rejects if aborted. */ async dequeue(signal?: AbortSignal): Promise<T> { if (await this.#sema.take(signal)) { return this.#poll() }
signal?.throwIfAborted() throw Error('canceled dequeue') }
#poll() { const item = this.#items.length > 0 && this.#items.shift() if (item) return item
throw Error('no elements') }}