RxJs Puffer bis Datenbank einfügen (Versprechen)

Question

Ich habe einen Datenstrom, mit schnell eingehenden Daten. Ich möchte sie in eine Datenbank einfügen, indem ich Ordnung halte. Ich habe eine Datenbank, die eine Zusage zurückgibt, die gelöst wird, wenn eine Einfügung erfolgreich ist.

Ich möchte einen Rx-Stream erstellen, der die neuen Daten puffert, bis die gepufferten Daten eingefügt werden.

Wie kann ich das tun?

Answer 1

Ich glaube, um genau das zu bekommen, was Sie wünschen, würden Sie Ihren eigenen Betreiber erstellen müssen. Brechen von RxJS bekommen Sie können leicht so etwas wie (Warnung, nicht getestet) ...

export class BusyBuffer<T> { 
    private itemQueue = new Subject<T>(); 
    private bufferTrigger = new Subject<{}>(); 
    private busy = false; 

    constructor(consumerCallback: (items: T[]) => Promise<void>) { 
    this.itemQueue.buffer(this.bufferTrigger).subscribe(items => { 
     this.busy = true; 
     consumerCallback(items).then(() => { 
     this.busy = false; 
     this.bufferTrigger.next(null); 
     }); 
    }); 
    } 

    submitItem(item: T) { 
    this.itemQueue.next(item); 
    if(!busy) { 
     this.bufferTrigger.next(null); 
    } 
    } 

} 

, die dann verwendet werden können, wie

let busyBuffer = new BusyBuffer<T>(items => { 
    return database.insertRecords(items); 
}); 
items.subscribe(item => busyBuffer.submitItem(item)); 

Es ist nicht genau, obwohl rein reaktiv ist und jemand kann in der Lage sein, sich etwas besseres auszudenken.