File size: 1,556 Bytes
4e23b01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 | export class AsyncEventQueue<T> implements AsyncIterable<T>, AsyncIterator<T> {
private readonly values: T[] = [];
private readonly waiters: Array<{
resolve: (result: IteratorResult<T>) => void;
reject: (reason?: unknown) => void;
}> = [];
private error: unknown;
private failed = false;
private ended = false;
push(value: T): void {
if (this.failed || this.ended) return;
const waiter = this.waiters.shift();
if (waiter !== undefined) {
waiter.resolve({ done: false, value });
return;
}
this.values.push(value);
}
end(): void {
if (this.failed || this.ended) return;
this.ended = true;
for (const waiter of this.waiters.splice(0)) {
waiter.resolve({ done: true, value: undefined });
}
}
fail(error: unknown): void {
if (this.failed || this.ended) return;
this.error = error;
this.failed = true;
if (this.values.length > 0) return;
for (const waiter of this.waiters.splice(0)) {
waiter.reject(error);
}
}
next(): Promise<IteratorResult<T>> {
if (this.values.length > 0) {
const value = this.values.shift()!;
return Promise.resolve({ done: false, value });
}
if (this.failed) {
return Promise.reject(this.error);
}
if (this.ended) {
return Promise.resolve({ done: true, value: undefined });
}
return new Promise<IteratorResult<T>>((resolve, reject) => {
this.waiters.push({ resolve, reject });
});
}
[Symbol.asyncIterator](): AsyncIterator<T> {
return this;
}
}
|