RxJs tampon jusqu'à ce que la base de données d'insertion (promesse)

Question

J'ai un flux de données, avec des données rapidement entrantes. Je veux les insérer dans une base de données en gardant l'ordre. J'ai une base de données, qui renvoie une promesse, qui est résolue quand une insertion est réussie.

Je souhaite faire un flux Rx, qui tamponne les nouvelles données, jusqu'à ce que les données en mémoire tampon est insérée.

Comment puis-je faire cela?

Answer 1

Je crois d'obtenir exactement ce que vous désirez, vous devez créer votre propre opérateur. Rompant RxJS un peu, vous pouvez obtenir quelque chose comme (avertissement, n'ont pas testé) ...

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); 
    } 
    } 

} 

qui peut ensuite être utilisé comme

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

Il est pas exactement purement réactif et si quelqu'un peut être capable de trouver quelque chose de mieux.