1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
var iterate = require('stream-iterate')
var from = require('from2')
var defaultKey = function (val) {
return val.key || val
}
var union = function (streamA, streamB, toKey) {
var readA = iterate(streamA)
var readB = iterate(streamB)
if (!toKey) toKey = defaultKey
var stream = from.obj(function loop (size, cb) {
readA(function (err, dataA, nextA) {
if (err) return cb(err)
readB(function (err, dataB, nextB) {
if (err) return cb(err)
if (!dataA && !dataB) return cb(null, null)
if (!dataA) {
nextB()
return cb(null, dataB)
}
if (!dataB) {
nextA()
return cb(null, dataA)
}
var keyA = toKey(dataA)
var keyB = toKey(dataB)
if (keyA === keyB) {
nextB()
return loop(size, cb)
}
if (keyA < keyB) {
nextA()
return cb(null, dataA)
}
nextB()
cb(null, dataB)
})
})
})
stream.on('close', function () {
if (streamA.destroy) streamA.destroy()
if (streamB.destroy) streamB.destroy()
})
return stream
}
module.exports = union
|
