问将请求流批处理为promises，按时间间隔分组

EN

{source1: [req1,req2,req3],
 source2: [req1,req2],
 .
 .
 sourceN: [req1,req2,...,reqm]}

var apiInterval = 10000;

function runner(){
  setTimeout(mv => { Promise.all(mv.map(reqs => Promise.all(reqs.map(req => apiCall(req)))))
                            .then(pss => pss.map(ps => ps.map(p => p.then(r => doSomethingWithEachApiCallResult(r)))));
                     clearMapValues(); // to be filled in the next 10 seconds
                     runner();
                   }, apiInterval, myMap.values.slice());
}

/**
 * A stream of requests come from various sources, can be transposed into a batch indexed 
 * by the source of the request.
 * 
 * The size of each batch is defined by a time interval. I.e. any request received within the
 * time interval is stored in a batch.
 */
export class BatchStream<K, V> {

    cache: Map<K, V[]>
    flushRate: number
    onBatch: (k: K, v: V[]) => Promise<void>
    debug: boolean

    constructor(onBatch: (k: K, v: V[]) => Promise<void>, flushRate = 5000, debug = false) {
        this.cache = new Map<K, V[]>()
        this.onBatch = onBatch
        this.debug = debug
        this.flushRate = flushRate
        this.flush = this.flush.bind(this)
    }

    push(k: K, v: V) {
        if (this.cache.has(k)) {
            let batch = this.cache.get(k)
            batch.push(v)
            this.cache.set(k, batch)
        } else {
            this.cache.set(k, [v])
            setTimeout(this.flush, this.flushRate, k)
        }
    }

    flush(k: K) {
        this.debug && console.log("Flush", k)
        let batch = this.cache.get(k)
        this.cache.delete(k)
        this.onBatch(k, batch)
        this.debug && console.log("Size", this.cache.size)
    }
}

it("BatchStream", (done) => {

        let sources = []
        let iterations = 10
        let jobs = []
        let jobsDone = 0
        let debug = true

        // Prepare sources
        for (let i = 97; i < 123; i++) {
            sources.push(String.fromCharCode(i))
        }

        // Prepare a stream of test data
        for (let k of sources) {
            for (let i = 0; i < iterations; i++) {
                jobs.push({ k, v: k + i.toString() })
            }
        }

        shuffle(jobs)

        // Batch handler
        let onBatch = (k: string, v: string[]) => {
            return new Promise<void>((resolve, reject) => {
                jobsDone += v.length
                debug && console.log(" --> " + k, v.length, v.join(","), jobsDone, sources.length * iterations)

                if (jobsDone == sources.length * iterations) {
                    done()
                }

                resolve()
            })
        }

        let batchStream = new BatchStream<string, string>(onBatch, 5000, debug)

        // Stream test data into batcher
        let delay = 0
        for (let j of jobs) {
            delay += 100
            setTimeout(() => {
                batchStream.push(j.k, j.v)
            }, delay)
        }
    })