JS/TSで擬似ブロック非同期キューを実装する方法は？

Question

ここには矛盾があります：javascript/typescriptで非同期ブロッキングキューを作成したいと思います（これは、typescriptなしで実装できます）。基本的には、JavaのBlockingQueueのようなものを実装したいのですが、実際にはブロックされているのではなく、非同期になり、デキューを待つことができます。ここで

は、私が実装するインタフェースです：

interface AsyncBlockingQueue<T> { 
    enqueue(t: T): void; 
    dequeue(): Promise<T>; 
} 

そして、私はそうのようにそれを使用したい：

// enqueue stuff somewhere else 

async function useBlockingQueue() { 
    // as soon as something is enqueued, the promise will be resolved: 
    const value = await asyncBlockingQueue.dequeue(); 
    // this will cause it to await for a second value 
    const secondValue = await asyncBlockingQueue.dequeue(); 
} 

任意のアイデア？

Answer 1

実際には非常に簡単ですが、dequeueは、enqueueが解決すると約束します。リゾルバをキューに入れておくだけでなく、デキューされる前に値がエンキューされ、すでに実行された約束をキューに保持するケースについても気にする必要があります。

class AsyncBlockingQueue { 
    constructor() { 
    // invariant: at least one of the arrays is empty 
    this.resolvers = []; 
    this.promises = []; 
    } 
    _add() { 
    this.promises.push(new Promise(resolve => { 
     this.resolvers.push(resolve); 
    }); 
    } 
    enqueue(t) { 
    // if (this.resolvers.length) this.resolvers.shift()(t); 
    // else this.promises.push(Promise.resolve(t)); 
    if (!this.resolvers.length) this._add(); 
    this.resolvers.shift()(t); 
    } 
    dequeue() { 
    if (!this.promises.length) this._add(); 
    return this.promises.shift(); 
    } 
    // now some utilities: 
    isEmpty() { // there are no values available 
    return !this.promises.length; // this.length == 0 
    } 
    isBlocked() { // it's waiting for values 
    return !!this.resolvers.length; // this.length < 0 
    } 
    get length() { 
    return this.promises.length - this.resolvers.length; 
    } 
} 

私はタイプスクリプトを知らないが、必要なタイプの注釈を追加するのは簡単だろう。

パフォーマンスを向上させるには、プレーン配列の代わりに循環バッファを使用するQueue実装を使用します。 this one。キューを1つだけ使用して、現在約束またはリゾルバを保存しているかどうかを覚えておくこともできます。

Answer 2

これは@Bergiの答えですが、typescript + genericsといくつかの変更を加えて、私のタイピング兄弟/姉妹の厳密なモードで動作させるようにしています。

class AsyncBlockingQueue<T> { 
    private _promises: Promise<T>[]; 
    private _resolvers: ((t: T) => void)[]; 

    constructor() { 
    this._resolvers = []; 
    this._promises = []; 
    } 

    private _add() { 
    this._promises.push(new Promise(resolve => { 
     this._resolvers.push(resolve); 
    })); 
    } 

    enqueue(t: T) { 
    if (!this._resolvers.length) this._add(); 
    const resolve = this._resolvers.shift(); 
    if (!resolve) { 
     // can never happen 
     throw new Error('resolve function was null or undefined when attempting to enqueue.') 
    }; 
    resolve(t); 
    } 

    dequeue() { 
    if (!this._promises.length) this._add(); 
    const promise = this._promises.shift(); 
    if (!promise) { 
     // can never happen 
     throw new Error('promise was null or undefined when attempting to dequeue.'); 
    } 
    return promise; 
    } 

    isEmpty() { 
    return !this._promises.length; 
    } 

    isBlocked() { 
    return !!this._resolvers.length; 
    } 

    get length() { 
    return this._promises.length - this._resolvers.length; 
    } 
}