async version of streamjs(https://github.com/winterbe/streamjs)
npm install async-streamjsnpm i async-streamjsjs
import AsyncStream from 'async-stream'
import _ from 'lodash'test('lazy query', async () => {
let seq = []
const totalCount = 12
const pageSize = 5
async function fakeQueryDB(pageIdx) {
seq.push(
query page ${pageIdx})
let offset = pageIdx * pageSize
return Array.from({length: pageSize}, (v, i) => i + offset)
.filter(v => v < totalCount)
}
let done = false
await AsyncStream.range()
.map(async pageIdx => {
let arr = done ? [] : await fakeQueryDB(pageIdx)
done = arr.length < pageSize
return arr
})
.takeWhile(arr => 0 < arr.length)
.flatMap(_.identity)
.forEach(v => seq.push(value ${v})) expect(seq).toEqual([
"query page 0",
"value 0",
"value 1",
"value 2",
"value 3",
"value 4",
"query page 1",
"value 5",
"value 6",
"value 7",
"value 8",
"value 9",
"query page 2",
"value 10",
"value 11",
])
});1
`$3
js
import AsyncStream from 'async-stream'
import _ from 'lodash'test('fifo queue', async () => {
let targetArr = await AsyncStream.fromAsyncCallback(resolve => {
let srcArr0 = _.range(0, 5)
let srcArr1 = _.range(5, 10)
for (let i of srcArr0) {
resolve(i)
}
setTimeout(() => {
for (let i of srcArr1) {
resolve(i)
}
resolve(null)
}, 500)
setTimeout(() => {
resolve(null)
}, 1000)
}).toArray()
expect(targetArr).toEqual(_.range(0, 10))
});
`
$3
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
* asyncCallback:
asyncPredicate: async val => boolean
asyncMapper: async val => any
asyncReducer: async (accumulate, val) => nextAccumulate
bufferedExecutor: resolveOnce => void, resolveOnce can call multiple times, resolveOnce(null) means end of data empty stream constant:
import {EMPTY_STREAM} from 'async-stream'`