feat(json.stream): add utility function for streamed JSON parsing

Author: zkat

README.md

@@ -82,6 +82,27 @@ const res = await fetch.json('/-/ping')
 console.log(res) // Body parsed as JSON
 ```
 
+#### <a name="fetch-json-stream"></a> `> fetch.json.stream(url, jsonPath, [opts]) -> Stream`
+
+Performs a request to a given registry URL and parses the body of the response
+as JSON, with each entry being emitted through the stream.
+
+The `jsonPath` argument is a [`JSONStream.parse()`
+path](https://github.com/dominictarr/JSONStream#jsonstreamparsepath), and the
+returned stream (unlike default `JSONStream`s), has a valid
+`Symbol.asyncIterator` implementation.
+
+For available options, please see the section on [`fetch` options](#fetch-opts).
+
+##### Example
+
+```javascript
+console.log('https://npm.im/~zkat has access to the following packages:')
+for await (let {key, value} of fetch.json.stream('/-/user/zkat/package', '$*')) {
+  console.log(`https://npm.im/${key} (perms: ${value})`)
+}
+```
+
 #### <a name="fetch-opts"></a> `fetch` Options
 
 Fetch options are optional, and can be passed in as either a Map-like object

index.js

@@ -6,7 +6,9 @@ const checkResponse = require('./check-response.js')
 const config = require('./config.js')
 const getAuth = require('./auth.js')
 const fetch = require('make-fetch-happen')
+const JSONStream = require('JSONStream')
 const npa = require('npm-package-arg')
+const {PassThrough} = require('stream')
 const qs = require('querystring')
 const url = require('url')
 const zlib = require('zlib')
@@ -110,6 +112,18 @@ function fetchJSON (uri, opts) {
   return regFetch(uri, opts).then(res => res.json())
 }
 
+module.exports.json.stream = fetchJSONStream
+function fetchJSONStream (uri, jsonPath, opts) {
+  const parser = JSONStream.parse(jsonPath)
+  const pt = parser.pipe(new PassThrough({objectMode: true}))
+  parser.on('error', err => pt.emit('error', err))
+  regFetch(uri, opts).then(res => {
+    res.body.on('error', err => parser.emit('error', err))
+    res.body.pipe(parser)
+  }, err => pt.emit('error', err))
+  return pt
+}
+
 module.exports.pickRegistry = pickRegistry
 function pickRegistry (spec, opts) {
   spec = npa(spec)

package-lock.json

@@ -29,6 +29,7 @@
   },
   "license": "ISC",
   "dependencies": {
+    "JSONStream": "^1.3.4",
     "bluebird": "^3.5.1",
     "figgy-pudding": "^3.4.1",
     "lru-cache": "^4.1.3",
@@ -37,6 +38,7 @@
   },
   "devDependencies": {
     "cacache": "^11.0.2",
+    "get-stream": "^4.0.0",
     "mkdirp": "^0.5.1",
     "nock": "^9.4.3",
     "npmlog": "^4.1.2",

package.json

@@ -3,6 +3,7 @@
 const Buffer = require('safe-buffer').Buffer
 
 const config = require('../config.js')
+const getStream = require('get-stream')
 const npmlog = require('npmlog')
 const PassThrough = require('stream').PassThrough
 const silentLog = require('../silentlog.js')
@@ -237,6 +238,23 @@ test('json()', t => {
     .then(json => t.deepEqual(json, {hello: 'world'}, 'got json body'))
 })
 
+test('fetch.json.stream()', t => {
+  tnock(t, OPTS.registry).get('/hello').reply(200, {
+    a: 1,
+    b: 2,
+    c: 3
+  })
+  return getStream.array(
+    fetch.json.stream('/hello', '$*', OPTS)
+  ).then(data => {
+    t.deepEqual(data, [
+      {key: 'a', value: 1},
+      {key: 'b', value: 2},
+      {key: 'c', value: 3}
+    ], 'got a streamed JSON body')
+  })
+})
+
 test('opts.ignoreBody', t => {
   tnock(t, OPTS.registry)
     .get('/hello')