|
| 1 | +var hash = require('hyperdb/lib/hash') |
| 2 | +var Readable = require('readable-stream').Readable |
| 3 | +var LRU = require('lru') |
| 4 | + |
| 5 | +module.exports = { attachCreateReadStream } |
| 6 | + |
| 7 | +function attachCreateReadStream (DB) { |
| 8 | + DB.prototype.createReadStream = createReadStream |
| 9 | +} |
| 10 | + |
| 11 | +function createReadStream (key, opts) { |
| 12 | + if (!opts) opts = {} |
| 13 | + var self = this |
| 14 | + var path = hash(key, true) |
| 15 | + var cacheMax = opts.cacheSize || 128 |
| 16 | + var keyCache = new LRU(cacheMax) |
| 17 | + var streamQueue |
| 18 | + var queueNeedsSorting = true |
| 19 | + var stream = new Readable({ objectMode: true }) |
| 20 | + stream._read = read |
| 21 | + |
| 22 | + return stream |
| 23 | + |
| 24 | + function read () { |
| 25 | + if (stream.destroyed) return |
| 26 | + // if no heads - get heads and process first tries |
| 27 | + if (!streamQueue) { |
| 28 | + self.heads(function (err, heads) { |
| 29 | + if (err) stream.emit('error', err) |
| 30 | + if (!heads.length) { |
| 31 | + stream.push(null) |
| 32 | + return |
| 33 | + } |
| 34 | + streamQueue = heads.map(h => ({ node: h, index: 0 })) |
| 35 | + next() |
| 36 | + }) |
| 37 | + return |
| 38 | + } |
| 39 | + next() |
| 40 | + } |
| 41 | + |
| 42 | + function next () { |
| 43 | + if (!streamQueue.length) { |
| 44 | + stream.push(null) |
| 45 | + return |
| 46 | + } |
| 47 | + // sort stream queue first to ensure that you always get the latest node |
| 48 | + // this requires offsetting feeds sequences based on when it started in relation to others |
| 49 | + if (queueNeedsSorting) streamQueue.sort(sortQueueByClockAndSeq) |
| 50 | + var data = streamQueue.pop() |
| 51 | + var node = data.node |
| 52 | + readNext(node, data.index, (err, match) => { |
| 53 | + if (err) { |
| 54 | + return stream.emit('error', err) |
| 55 | + } |
| 56 | + if (!match) return next() |
| 57 | + // check if really a match and not encountered before |
| 58 | + check(node, (matchingNode) => { |
| 59 | + if (!matchingNode) next() |
| 60 | + else { |
| 61 | + keyCache.set(node.key, true) |
| 62 | + stream.push(matchingNode) |
| 63 | + } |
| 64 | + }) |
| 65 | + }) |
| 66 | + } |
| 67 | + |
| 68 | + function sortQueueByClockAndSeq (a, b) { |
| 69 | + a = a.node |
| 70 | + b = b.node |
| 71 | + var sortValue = sortNodesByClock(a, b) |
| 72 | + if (sortValue !== 0) return sortValue |
| 73 | + // same time, so use sequence to order |
| 74 | + if (a.feed === b.feed) return a.seq - b.seq |
| 75 | + var bOffset = b.clock.reduce((p, v) => p + v, b.seq) |
| 76 | + var aOffset = a.clock.reduce((p, v) => p + v, a.seq) |
| 77 | + // if real sequence is the same then return sort on feed |
| 78 | + if (bOffset === aOffset) return b.feed - a.feed |
| 79 | + return aOffset - bOffset |
| 80 | + } |
| 81 | + |
| 82 | + function check (node, cb) { |
| 83 | + // is it actually a match and not a collision |
| 84 | + if (!(node && node.key && node.key.indexOf(key) === 0)) return cb() |
| 85 | + // have we encountered this node before |
| 86 | + if (keyCache.get(node.key)) return cb() |
| 87 | + // it is not in the cache but might still be a duplicate if cache is full |
| 88 | + // if (keyCache.length === cacheMax) { |
| 89 | + // so check if this is the first instance of the node |
| 90 | + // TODO: Atm this is a bit of a hack to get conflicting values |
| 91 | + // ideally this should not need to retraverse the trie. |
| 92 | + // Potential issue here when db is updated after stream was created! |
| 93 | + return self._get(node.key, false, [], noop, (err, latest) => { |
| 94 | + if (err) return stream.emit('error', err) |
| 95 | + if (sortNodesByClock(node, Array.isArray(latest) ? latest[0] : latest) >= 0) { |
| 96 | + cb(latest) |
| 97 | + } else { |
| 98 | + cb() |
| 99 | + } |
| 100 | + }) |
| 101 | + } |
| 102 | + |
| 103 | + function readNext (node, i, cb) { |
| 104 | + var writers = self._writers |
| 105 | + var trie |
| 106 | + var missing = 0 |
| 107 | + var error |
| 108 | + var vals |
| 109 | + for (; i < path.length - 1; i++) { |
| 110 | + if (node.path[i] === path[i]) continue |
| 111 | + // check trie |
| 112 | + trie = node.trie[i] |
| 113 | + if (!trie) { |
| 114 | + return cb(null) |
| 115 | + } |
| 116 | + vals = trie[path[i]] |
| 117 | + // not found |
| 118 | + if (!vals || !vals.length) { |
| 119 | + return cb(null) |
| 120 | + } |
| 121 | + |
| 122 | + missing = vals.length |
| 123 | + error = null |
| 124 | + for (var j = 0; j < vals.length; j++) { |
| 125 | + // fetch potential |
| 126 | + writers[vals[j].feed].get(vals[j].seq, (err, val) => { |
| 127 | + if (err) { |
| 128 | + error = err |
| 129 | + } else { |
| 130 | + pushToQueue({ node: val, index: i }) |
| 131 | + } |
| 132 | + missing-- |
| 133 | + if (!missing) { |
| 134 | + cb(error) |
| 135 | + } |
| 136 | + }) |
| 137 | + } |
| 138 | + return |
| 139 | + } |
| 140 | + |
| 141 | + // Traverse the rest of the node's trie, recursively, |
| 142 | + // hunting for more nodes with the desired prefix. |
| 143 | + for (; i < node.trie.length; i++) { |
| 144 | + trie = node.trie[i] || [] |
| 145 | + for (j = 0; j < trie.length; j++) { |
| 146 | + var entrySet = trie[j] || [] |
| 147 | + for (var el = 0; el < entrySet.length; el++) { |
| 148 | + var entry = entrySet[el] |
| 149 | + missing++ |
| 150 | + writers[entry.feed].get(entry.seq, (err, val) => { |
| 151 | + if (err) { |
| 152 | + error = err |
| 153 | + } else if (val.key && val.value) { |
| 154 | + pushToQueue({ node: val, index: i + 1 }) |
| 155 | + } |
| 156 | + missing-- |
| 157 | + if (!missing) { |
| 158 | + if (i < node.trie.length) { |
| 159 | + pushToQueue({ node: node, index: i + 1 }) |
| 160 | + cb(error, false) |
| 161 | + } else { |
| 162 | + cb(error, true) |
| 163 | + } |
| 164 | + } |
| 165 | + }) |
| 166 | + } |
| 167 | + } |
| 168 | + if (missing > 0) return |
| 169 | + } |
| 170 | + return cb(null, true) |
| 171 | + } |
| 172 | + |
| 173 | + function pushToQueue (item) { |
| 174 | + queueNeedsSorting = streamQueue.length > 0 && (sortQueueByClockAndSeq(item, streamQueue[streamQueue.length - 1]) < 0) |
| 175 | + streamQueue.push(item) |
| 176 | + } |
| 177 | +} |
| 178 | + |
| 179 | +function sortNodesByClock (a, b) { |
| 180 | + var isGreater = false |
| 181 | + var isLess = false |
| 182 | + var length = a.clock.length |
| 183 | + if (b.clock.length > length) length = b.clock.length |
| 184 | + for (var i = 0; i < length; i++) { |
| 185 | + var diff = (a.clock[i] || 0) - (b.clock[i] || 0) |
| 186 | + if (diff > 0) isGreater = true |
| 187 | + if (diff < 0) isLess = true |
| 188 | + } |
| 189 | + if (isGreater && isLess) return 0 |
| 190 | + if (isLess) return -1 |
| 191 | + if (isGreater) return 1 |
| 192 | + return 0 |
| 193 | +} |
| 194 | + |
| 195 | +function noop () {} |
0 commit comments