Allow the same data to be shared by multiple Upload scalars

changelog.md

@@ -2,6 +2,15 @@
 
 ## 6.0.0-alpha.1
 
+### Major
+
+- API change: `processRequest` now requires a [`http.ServerResponse`](https://nodejs.org/api/http.html#http_class_http_serverresponse) as its second argument
+- API change: upload promises now resolve with a `createReadStream` method instead of a `stream` property
+- An `Upload` variable can now be used by multiple resolvers
+- Multile `Upload` variables can now use the same multipart data
+
+## 6.0.0-alpha.1
+
 Big thanks to new collaborator [@mike-marcacci](https://github.com/mike-marcacci) for his help solving tricky bugs and edge-cases!
 
 ### Major

package.json

@@ -37,7 +37,7 @@
   },
   "dependencies": {
     "busboy": "^0.2.14",
-    "fs-capacitor": "^0.0.3",
+    "fs-capacitor": "^1.0.0",
     "object-path": "^0.11.4"
   },
   "devDependencies": {

readme.md

@@ -79,7 +79,7 @@ app.use(
 
 ### Custom middleware
 
-Middleware wraps the async function `processRequest` which accepts a Node.js request and an optional [options object](#options) as arguments. It returns a promise that resolves an operations object for a GraphQL server to consume (usually as the request body). Import it to create custom middleware:
+Middleware wraps the async function `processRequest` which accepts a required [http.IncomingMessage](https://nodejs.org/api/http.html#http_class_http_incomingmessage), a required [http.ServerResponse](https://nodejs.org/api/http.html#http_class_http_serverresponse), and an optional [options object](#options) as arguments. It returns a promise that resolves an operations object for a GraphQL server to consume (usually as the request body). Import it to create custom middleware:
 
 ```js
 import { processRequest } from 'apollo-upload-server'
@@ -89,10 +89,10 @@ import { processRequest } from 'apollo-upload-server'
 
 A file upload promise that resolves an object containing:
 
-- `stream`
-- `filename`
-- `mimetype`
-- `encoding`
+- `filename`: `string` - the filename of the upload
+- `mimetype`: `string` - the mimetype of the upload
+- `encoding`: `string` - the encoding of the upload
+- `createReadStream`: `() => ReadStream` - calling this method returns a readable stream of the upload's contents. Calling `createReadStream` multiple times will create multiple streams which can be read indipendantly of each other. Note that `createReadStream` will throw if called after all resolvers have finished, or after an error has interrupted the request.
 
 It must be added to your types and resolvers:
 

src/errors.mjs

@@ -23,5 +23,4 @@ export class MaxFilesUploadError extends UploadError {}
 export class MapBeforeOperationsUploadError extends UploadError {}
 export class FilesBeforeMapUploadError extends UploadError {}
 export class FileMissingUploadError extends UploadError {}
-export class UploadPromiseDisconnectUploadError extends UploadError {}
-export class FileStreamDisconnectUploadError extends UploadError {}
+export class DisconnectUploadError extends UploadError {}

src/middleware.mjs

@@ -1,15 +1,14 @@
 import Busboy from 'busboy'
 import objectPath from 'object-path'
-import Capacitor from 'fs-capacitor'
+import WriteStream from 'fs-capacitor'
 import {
   SPEC_URL,
   MaxFileSizeUploadError,
   MaxFilesUploadError,
   MapBeforeOperationsUploadError,
   FilesBeforeMapUploadError,
   FileMissingUploadError,
-  UploadPromiseDisconnectUploadError,
-  FileStreamDisconnectUploadError
+  DisconnectUploadError
 } from './errors'
 
 class Upload {
@@ -30,9 +29,16 @@ class Upload {
 
 export const processRequest = (
   request,
+  response,
   { maxFieldSize, maxFileSize, maxFiles } = {}
 ) =>
   new Promise((resolve, reject) => {
+    let operations
+    let operationsPath
+    let map
+    let currentStream
+    let error
+
     const parser = new Busboy({
       headers: request.headers,
       limits: {
@@ -43,16 +49,36 @@ export const processRequest = (
       }
     })
 
-    let operations
-    let operationsPath
-    let map
-    let currentStream
+    const exit = exitError => {
+      if (error) return
+      error = exitError
 
-    const exit = error => {
       reject(error)
-      parser.destroy(error)
+
+      parser.destroy()
+
+      if (currentStream) currentStream.destroy(error)
+
+      if (map)
+        for (const upload of map.values())
+          if (!upload.file) upload.reject(error)
+
+      request.unpipe(parser)
+      request.resume()
     }
 
+    let isReleased = false
+    const release = () => {
+      if (isReleased) return
+      isReleased = true
+
+      if (map)
+        for (const upload of map.values())
+          if (upload.file) upload.file.capacitor.destroy()
+    }
+
+    // Parser Events
+    // -------------
     parser.on('field', (fieldName, value) => {
       switch (fieldName) {
         case 'operations':
@@ -93,8 +119,6 @@ export const processRequest = (
           for (const [fieldName, paths] of mapEntries) {
             map.set(fieldName, new Upload())
 
-            // Repopulate operations with the promise wherever the file occurred
-            // for use by the Upload scalar.
             for (const path of paths)
               operationsPath.set(path, map.get(fieldName).promise)
           }
@@ -108,7 +132,6 @@ export const processRequest = (
       if (!map) {
         // Prevent an unhandled error from crashing the process.
         stream.on('error', () => {})
-
         stream.resume()
 
         return exit(
@@ -120,56 +143,57 @@ export const processRequest = (
       }
 
       currentStream = stream
-
       stream.on('end', () => {
         if (currentStream === stream) currentStream = null
       })
 
-      if (map.has(fieldName)) {
-        const capacitor = new Capacitor()
+      const upload = map.get(fieldName)
+      if (upload) {
+        const capacitor = new WriteStream()
 
         capacitor.on('error', () => {
           stream.unpipe()
           stream.resume()
         })
 
-        stream.on('limit', () =>
+        stream.on('limit', () => {
+          if (currentStream === stream) currentStream = null
+          stream.unpipe()
           capacitor.destroy(
             new MaxFileSizeUploadError(
               'File truncated as it exceeds the size limit.',
               413
             )
           )
-        )
-
-        stream.on('error', error => {
-          if (capacitor.finished || capacitor.destroyed) return
-
-          // A terminated connection may cause the request to emit a 'close'
-          // event either before or after the parser encounters an error,
-          // depending on the Node.js version and the state of stream buffers.
+        })
 
-          if (
-            error.message ===
-              // https://github.com/mscdex/dicer/blob/v0.2.5/lib/Dicer.js#L62
-              'Unexpected end of multipart data' ||
-            error.message ===
-              // https://github.com/mscdex/dicer/blob/v0.2.5/lib/Dicer.js#L65
-              'Part terminated early due to unexpected end of multipart data'
-          )
-            error = new FileStreamDisconnectUploadError(error.message)
+        stream.on('error', streamError => {
+          if (currentStream === stream) currentStream = null
 
-          capacitor.destroy(error)
+          stream.unpipe()
+          capacitor.destroy(error || streamError)
         })
 
         stream.pipe(capacitor)
 
-        map.get(fieldName).resolve({
-          stream: capacitor,
-          filename,
-          mimetype,
-          encoding
-        })
+        upload.resolve(
+          Object.create(null, {
+            capacitor: { value: capacitor, enumerable: false },
+            createReadStream: {
+              value() {
+                const createReadStreamError =
+                  capacitor.error || (isReleased ? error : null)
+
+                if (createReadStreamError) throw createReadStreamError
+                return capacitor.createReadStream()
+              },
+              enumerable: true
+            },
+            encoding: { value: encoding, enumerable: true },
+            filename: { value: filename, enumerable: true },
+            mimetype: { value: mimetype, enumerable: true }
+          })
+        )
       }
       // Discard the unexpected file.
       else {
@@ -196,30 +220,26 @@ export const processRequest = (
             )
     })
 
-    parser.on('error', error => {
-      request.unpipe(parser)
-      request.resume()
+    parser.once('error', error => {
+      exit(error)
+    })
 
-      if (map)
-        for (const upload of map.values())
-          if (!upload.file) upload.reject(error)
+    // Response Events
+    // ---------------
+    response.once('finish', release)
+    response.once('close', release)
 
-      if (currentStream) currentStream.destroy(error)
+    // Request Events
+    // --------------
+    let requestEnded = false
+    request.once('end', () => {
+      requestEnded = true
     })
 
-    request.on('close', () => {
-      if (map)
-        for (const upload of map.values())
-          if (!upload.file)
-            upload.reject(
-              new UploadPromiseDisconnectUploadError(
-                'Request disconnected before file upload stream parsing.'
-              )
-            )
-
-      if (!parser._finished)
-        parser.destroy(
-          new FileStreamDisconnectUploadError(
+    request.once('close', () => {
+      if (!requestEnded)
+        exit(
+          new DisconnectUploadError(
             'Request disconnected during file upload stream parsing.'
           )
         )
@@ -234,7 +254,7 @@ export const apolloUploadKoa = options => async (ctx, next) => {
   const finished = new Promise(resolve => ctx.req.on('end', resolve))
 
   try {
-    ctx.request.body = await processRequest(ctx.req, options)
+    ctx.request.body = await processRequest(ctx.req, ctx.res, options)
     await next()
   } finally {
     await finished
@@ -245,16 +265,16 @@ export const apolloUploadExpress = options => (request, response, next) => {
   if (!request.is('multipart/form-data')) return next()
 
   const finished = new Promise(resolve => request.on('end', resolve))
-  const { send } = response
 
+  const { send } = response
   response.send = (...args) => {
     finished.then(() => {
       response.send = send
       response.send(...args)
     })
   }
 
-  processRequest(request, options)
+  processRequest(request, response, options)
     .then(body => {
       request.body = body
       next()