search: first stab at streaming search

aredridel · zkat · commit cfd43b49aed3 · 2016-10-13T15:36:21.000-04:00
Credit: @aredridel PR-URL: #13746 Reviewed-By: @othiym23
diff --git a/lib/cache/update-index.js b/lib/cache/update-index.js
@@ -11,14 +11,18 @@ var cacheFile = require('npm-cache-filename')
 var getCacheStat = require('./get-stat.js')
 var mapToRegistry = require('../utils/map-to-registry.js')
 var pulseTillDone = require('../utils/pulse-till-done.js')
-var parseJSON = require('../utils/parse-json.js')
+var jsonstream = require('JSONStream')
+var asyncMap = require('slide').asyncMap
+var writeStreamAtomic = require('fs-write-stream-atomic')
+var once = require('once')
 
 /* /-/all is special.
  * It uses timestamp-based caching and partial updates,
  * because it is a monster.
  */
-function updateIndex (staleness, cb) {
-  assert(typeof cb === 'function', 'must pass callback to updateIndex')
+function updateIndex (staleness, args, notArgs, filter, cb) {
+  assert(typeof filter === 'function', 'must pass filter callback to updateIndex')
+  assert(typeof cb === 'function', 'must pass final callback to updateIndex')
 
   mapToRegistry('-/all', npm.config, function (er, uri, auth) {
     if (er) return cb(er)
@@ -27,79 +31,163 @@ function updateIndex (staleness, cb) {
       timeout: staleness,
       follow: true,
       staleOk: true,
-      auth: auth
+      auth: auth,
+      streaming: true
     }
-    var cacheBase = cacheFile(npm.config.get('cache'))(uri)
-    var cachePath = path.join(cacheBase, '.cache.json')
-    log.info('updateIndex', cachePath)
+    var cacheBase = path.join(cacheFile(npm.config.get('cache'))(uri), '_search')
+    log.info('updateIndex', cacheBase)
 
     getCacheStat(function (er, st) {
       if (er) return cb(er)
 
       mkdir(cacheBase, function (er, made) {
         if (er) return cb(er)
 
-        fs.readFile(cachePath, function (er, data) {
-          if (er) {
-            log.warn('', 'Building the local index for the first time, please be patient')
-            return updateIndex_(uri, params, {}, cachePath, cb)
-          }
+        chownr(made || cacheBase, st.uid, st.gid, function (er) {
+          if (er) return cb(er)
 
-          chownr(made || cachePath, st.uid, st.gid, function (er) {
+          fs.readdir(cacheBase, function (er, cacheFiles) {
             if (er) return cb(er)
 
-            data = parseJSON.noExceptions(data)
-            if (!data) {
-              fs.writeFile(cachePath, '{}', function (er) {
-                if (er) return cb(new Error('Broken cache.'))
-
-                log.warn('', 'Building the local index for the first time, please be patient')
-                return updateIndex_(uri, params, {}, cachePath, cb)
-              })
-            }
-
-            var t = +data._updated || 0
-            // use the cache and update in the background if it's not too old
-            if (Date.now() - t < 60000) {
-              cb(null, data)
-              cb = function () {}
-            }
-
-            if (t === 0) {
-              log.warn('', 'Building the local index for the first time, please be patient')
-            } else {
-              log.verbose('updateIndex', 'Cached search data present with timestamp', t)
-              uri += '/since?stale=update_after&startkey=' + t
-            }
-            updateIndex_(uri, params, data, cachePath, cb)
+            cacheFiles.sort()
+
+            var latest = 0
+            asyncMap(cacheFiles, function (file, cb) {
+              log.silly('search', 'reading cache ' + file)
+              cb = once(cb)
+              var m = /^(\d+)-(\d+)[.]json/.exec(file)
+              if (m) {
+                latest = Number(m[2])
+                var cacheFile = path.join(cacheBase, file)
+
+                fs.stat(cacheFile, function (er, stat) {
+                  if (er) return cb(er)
+                  var r = fs.createReadStream(cacheFile).pipe(log.newStream('readCache', stat.size))
+                  var f = r.pipe(collectResults(filter, args, notArgs, cb))
+                  f.once('error', cb)
+                })
+              } else {
+                cb(null, {})
+              }
+            }, function (err, data) {
+              if (err) return cb(err)
+
+              data = data.reduce(function (a, e) {
+                Object.keys(e).forEach(function (k) {
+                  a[k] = e[k]
+                })
+                return a
+              }, {})
+
+              // use the cache and make no request if it's not too old
+              if (Date.now() - latest < 60000) {
+                finish(data, cb)
+              } else {
+                if (latest === 0) {
+                  log.warn('', 'Building the local index for the first time, please be patient')
+                } else {
+                  log.verbose('updateIndex', 'Cached search data present with timestamp', latest)
+                  uri += '/since?stale=update_after&startkey=' + latest
+                }
+
+                updateIndex_(uri, params, latest, filter, args, notArgs, cacheBase, function (err, updated) {
+                  if (err) return cb(err)
+
+                  Object.keys(updated).forEach(function (k) {
+                    data[k] = updated[k]
+                  })
+
+                  finish(data, cb)
+                })
+              }
+            })
           })
         })
       })
     })
   })
 }
 
-function updateIndex_ (all, params, data, cachePath, cb) {
-  log.silly('update-index', 'fetching', all)
-  npm.registry.request(all, params, pulseTillDone('updateIndex', function (er, updates, _, res) {
-    if (er) return cb(er, data)
+function finish (data, cb) {
+  var keys = Object.keys(data)
+  keys.sort()
+  var results = keys.map(function (k) {
+    return data[k]
+  })
 
-    var headers = res.headers
-    var updated = updates._updated || Date.parse(headers.date)
+  cb(null, results)
+}
 
-    Object.keys(updates).forEach(function (p) { data[p] = updates[p] })
+function updateIndex_ (all, params, latest, filter, args, notArgs, cacheBase, cb) {
+  cb = once(cb)
+  log.silly('update-index', 'fetching', all)
+  npm.registry.request(all, params, pulseTillDone('updateIndex', function (er, res) {
+    if (er) return cb(er)
 
-    data._updated = updated
-    getCacheStat(function (er, st) {
-      if (er) return cb(er)
+    var results = null
+    var updated = null
+    var wroteUpdate = false
+
+    var trackerStream = log.newStream('updateIndex')
+
+    var tmpName = path.join(cacheBase, latest + '-next.json')
+    var writeStream = writeStreamAtomic(tmpName)
+    res.setMaxListeners(20) // node 0.8 has a lower margin
+    res.pipe(writeStream)
+    res.pipe(trackerStream)
+    writeStream.once('error', cb)
+    writeStream.once('close', function () {
+      wroteUpdate = true
+      maybeFinishUpdateIndex()
+    })
 
-      fs.writeFile(cachePath, JSON.stringify(data), function (er) {
-        delete data._updated
-        if (er) return cb(er)
-        chownr(cachePath, st.uid, st.gid, function (er) {
-          cb(er, data)
+    res.pipe(collectResults(filter, args, notArgs, function (err, results_, updated_) {
+      if (err) return cb(err)
+      results = results_
+      updated = updated_
+      maybeFinishUpdateIndex()
+    }))
+
+    function maybeFinishUpdateIndex () {
+      if (results && wroteUpdate) {
+        var finalName = path.join(cacheBase, latest + '-' + updated + '.json')
+        log.silly('update-index', 'moving final cache file into place', finalName)
+        fs.rename(tmpName, finalName, function (err) {
+          if (err) return cb(err)
+          cb(null, results)
         })
-      })
-    })
+      }
+    }
   }))
 }
+
+function collectResults (filter, args, notArgs, cb) {
+  cb = once(cb)
+
+  var results = {}
+  var updated = null
+  var stream = jsonstream.parse('*', function (pkg, key) {
+    if (key[0] === '_updated') {
+      updated = pkg
+      return
+    }
+    if (key[0][0] !== '_') {
+      if (filter(pkg, args, notArgs)) {
+        log.verbose('search', 'matched ' + pkg.name)
+        results[pkg.name] = pkg
+      } else {
+        log.silly('search', 'not matched ' + pkg.name)
+      }
+    } else {
+      log.silly('search', 'skipping ' + key)
+    }
+  })
+
+  stream.once('error', cb)
+
+  stream.once('end', function () {
+    cb(null, results, updated)
+  })
+
+  return stream
+}
diff --git a/lib/search.js b/lib/search.js
@@ -6,12 +6,16 @@ var columnify = require('columnify')
 var updateIndex = require('./cache/update-index.js')
 var usage = require('./utils/usage')
 var output = require('./utils/output.js')
+var log = require('npmlog')
 
 search.usage = usage(
   'search',
   'npm search [--long] [search terms ...]'
 )
 
+search.usage = 'npm search [--long] [search terms ...]' +
+               '\n\naliases: s, se'
+
 search.completion = function (opts, cb) {
   var compl = {}
   var partial = opts.partialWord
@@ -66,35 +70,24 @@ function search (args, silent, staleness, cb) {
     return s.toLowerCase()
   })
 
-  getFilteredData(staleness, opts, searchexclude, function (er, data) {
+  updateIndex(staleness, opts, searchexclude, filter, function (er, data) {
     // now data is the list of data that we want to show.
     // prettify and print it, and then provide the raw
     // data to the cb.
     if (er || silent) return cb(er, data)
+    log.clearProgress()
     output(prettify(data, args))
     cb(null, data)
   })
 }
 
-function getFilteredData (staleness, args, notArgs, cb) {
-  updateIndex(staleness, function (er, data) {
-    if (er) return cb(er)
-    return cb(null, filter(data, args, notArgs))
-  })
-}
-
 function filter (data, args, notArgs) {
-  // data={<name>:{package data}}
-  return Object.keys(data).map(function (d) {
-    return data[d]
-  }).filter(function (d) {
-    return typeof d === 'object'
-  }).map(stripData).map(getWords).filter(function (data) {
-    return filterWords(data, args, notArgs)
-  }).reduce(function (l, r) {
-    l[r.name] = r
-    return l
-  }, {})
+  if (typeof data !== 'object') return false
+
+  data = getWords(stripData(data))
+  if (filterWords(data, args, notArgs)) return true
+
+  return false
 }
 
 function stripData (data) {
@@ -163,7 +156,7 @@ function prettify (data, args) {
   var lines = Object.keys(data).map(function (d) {
     // strip keyname
     return data[d]
-  }).map(function (dat) {
+  }).map(stripData).map(function (dat) {
     dat.author = dat.maintainers
     delete dat.maintainers
     dat.date = dat.time
diff --git a/package.json b/package.json
@@ -131,6 +131,7 @@
     "inherits",
     "ini",
     "init-package-json",
+    "jsonstream",
     "lockfile",
     "lodash._baseindexof",
     "lodash._baseuniq",
diff --git a/test/tap/search.js b/test/tap/search.js
@@ -12,7 +12,7 @@ var common = require('../common-tap.js')
 var pkg = path.resolve(__dirname, 'search')
 var cache = path.resolve(pkg, 'cache')
 var registryCache = path.resolve(cache, 'localhost_1337', '-', 'all')
-var cacheJsonFile = path.resolve(registryCache, '.cache.json')
+var searchDir = path.resolve(registryCache, '_search')
 
 var timeMock = {
   epoch: 1411727900,
@@ -53,15 +53,18 @@ test('No previous cache, init cache triggered by first search', function (t) {
       'search', 'do not do extra search work on my behalf',
       '--registry', common.registry,
       '--cache', cache,
-      '--loglevel', 'silent',
-      '--color', 'always'
+      '--loglevel', 'error',
+      '--color', 'never'
     ],
     EXEC_OPTS,
-    function (err, code) {
+    function (err, code, stdout, stderr) {
       s.close()
+      t.equal(stderr, '', 'had no error output')
       t.equal(code, 0, 'search finished successfully')
       t.ifErr(err, 'search finished successfully')
 
+      var cacheJsonFile = path.resolve(searchDir, '0-' + timeMock.future + '.json')
+
       t.ok(
         fs.existsSync(cacheJsonFile),
         cacheJsonFile + ' expected to have been created'
@@ -102,15 +105,18 @@ test('previous cache, _updated set, should trigger since request', function (t)
       'search', 'do not do extra search work on my behalf',
       '--registry', common.registry,
       '--cache', cache,
-      '--loglevel', 'silly',
+      '--loglevel', 'error',
       '--color', 'always'
     ],
     EXEC_OPTS,
-    function (err, code) {
+    function (err, code, stdout, stderr) {
       s.close()
+      t.equal(stderr, '', 'no error output')
       t.equal(code, 0, 'search finished successfully')
       t.ifErr(err, 'search finished successfully')
 
+      var cacheJsonFile = path.resolve(searchDir, '0-' + timeMock.epoch + '.json')
+
       var cacheData = JSON.parse(fs.readFileSync(cacheJsonFile, 'utf8'))
       t.equal(
         cacheData._updated,
@@ -143,12 +149,14 @@ searches.forEach(function (search) {
         'search', search.term,
         '--registry', common.registry,
         '--cache', cache,
-        '--loglevel', 'silent',
+        '--loglevel', 'error',
         '--color', 'always'
       ],
       EXEC_OPTS,
-      function (err, code, stdout) {
+      function (err, code, stdout, stderr) {
         s.close()
+        t.equal(stderr, '', 'no error output')
+        t.notEqual(stdout, '', 'got output')
         t.equal(code, 0, 'search finished successfully')
         t.ifErr(err, 'search finished successfully')
         // \033 == \u001B
@@ -183,6 +191,8 @@ function setupCache () {
   cleanup()
   mkdirp.sync(cache)
   mkdirp.sync(registryCache)
+  mkdirp.sync(searchDir)
+  var cacheJsonFile = path.resolve(searchDir, '0-' + timeMock.epoch + '.json')
   var res = fs.writeFileSync(cacheJsonFile, stringifyUpdated(timeMock.epoch))
   if (res) throw new Error('Creating cache file failed')
 }
diff --git a/test/tap/update-index.js b/test/tap/update-index.js
