fix(match): Rewrite the conditional stream logic (#25)

cache.js

@@ -7,6 +7,7 @@ const ssri = require('ssri')
 const through = require('mississippi').through
 const to = require('mississippi').to
 const url = require('url')
+const stream = require('stream')
 
 const MAX_MEM_SIZE = 5 * 1024 * 1024 // 5MB
 
@@ -57,51 +58,45 @@ module.exports = class Cache {
             status: 200
           })
         }
-        return Promise.resolve().then(() => {
-          const cachePath = this._path
-          let disturbed = false
-          // avoid opening cache file handles until a user actually tries to
-          // read from it.
-          const body = through((chunk, enc, cb) => {
-            if (disturbed) {
-              cb(null, chunk, enc)
-            } else {
-              disturbed = true
-              if (opts.memoize !== false && info.size > MAX_MEM_SIZE) {
-                pipe(
-                  cacache.get.stream.byDigest(cachePath, info.integrity, {
-                    memoize: opts.memoize
-                  }),
-                  body,
-                  () => {}
-                )
-              } else {
-                // cacache is much faster at bulk reads
-                cacache.get.byDigest(cachePath, info.integrity, {
-                  memoize: opts.memoize
-                }).then(data => {
-                  body.write(data, () => {
-                    body.end()
-                  })
-                }, err => body.emit('error', err))
-              }
-              cb() // throw away dummy data
+        let body
+        const cachePath = this._path
+        // avoid opening cache file handles until a user actually tries to
+        // read from it.
+        if (opts.memoize !== false && info.size > MAX_MEM_SIZE) {
+          body = new stream.PassThrough()
+          const realRead = body._read
+          body._read = function (size) {
+            body._read = realRead
+            pipe(
+              cacache.get.stream.byDigest(cachePath, info.integrity, {
+                memoize: opts.memoize
+              }),
+              body,
+              err => body.emit(err))
+            return realRead.call(this, size)
+          }
+        } else {
+          let readOnce = false
+          // cacache is much faster at bulk reads
+          body = new stream.Readable({
+            read () {
+              if (readOnce) return this.push(null)
+              readOnce = true
+              cacache.get.byDigest(cachePath, info.integrity, {
+                memoize: opts.memoize
+              }).then(data => {
+                this.push(data)
+                this.push(null)
+              }, err => this.emit('error', err))
             }
           })
-          body.write('dummy')
-          return new fetch.Response(body, {
-            url: req.url,
-            headers: resHeaders,
-            status: 200,
-            size: info.size
-          })
-        }).catch(err => {
-          if (err.code === 'ENOENT') {
-            return null
-          } else {
-            throw err
-          }
-        })
+        }
+        return Promise.resolve(new fetch.Response(body, {
+          url: req.url,
+          headers: resHeaders,
+          status: 200,
+          size: info.size
+        }))
       }
     })
   }