Server2server upload support also for remote merged schemas

readme.md

@@ -136,6 +136,82 @@ const files = ReactNativeFile.list([
 ])
 ```
 
+### Uploading from server-to-server
+
+The main difference between using apollo-upload-client in a browser vs server environment is that in the server version you need to provide a function that replaces `FormData` which is native to any browser. The `createUploadLink` function receives the param `serverFormData` to that end. We suggest you use the npm package `forma-data` as you can see in the example bellow.
+
+```js
+import { ApolloClient } from 'apollo-client'
+import { InMemoryCache } from 'apollo-cache-inmemory'
+import { createUploadLink } from 'apollo-upload-client'
+import gql from 'graphql-tag'
+import fs from 'fs'
+import fetch from 'node-fetch'
+import FormData from 'form-data'
+
+const client = new ApolloClient({
+  link: createUploadLink({
+    uri: 'https://example.server.com/graphql',
+    serverFormData: FormData,
+    fetch
+  }),
+  cache: new InMemoryCache()
+})
+
+const variables = {
+  file: fs.createReadStream('/path/to/file')
+}
+
+const mutation = gql`
+  mutation UploadFile($file: Upload!) {
+    uploadFile(file: $file) {
+      id
+    }
+  }
+`
+client.mutate({ mutation, variables })
+```
+
+### Redirecting uploads to remote merged schemas
+
+This will eventually be supported by `mergeSchemas` function, but at this point this is a solution.
+
+```js
+import { ApolloClient } from 'apollo-client'
+import { InMemoryCache } from 'apollo-cache-inmemory'
+import { createUploadLink } from 'apollo-upload-client'
+import { mergeSchemas } from 'graphql-tools'
+import gql from 'graphql-tag'
+import fetch from 'node-fetch'
+import FormData from 'form-data'
+
+mergeSchemas({
+  schemas: [bookSchema, authorSchema],
+  resolvers: mergeInfo => ({
+    Mutation: {
+      async uploadBook(parent, args, context, info){
+
+        const client = new ApolloClient({
+          link: createUploadLink({
+            uri: 'https://book.microservice.com/graphql',
+            serverFormData: FormData,
+            fetch
+          }),
+          cache: new InMemoryCache()
+        })
+
+        return await client.mutate({
+          // if you add 'request' in your context by default
+          // you can:
+          mutation: gql(context.request.body.query),
+          variables: args
+        })
+      }
+    }
+  }
+}
+```
+
 ## Support
 
 * Node.js v6.10+, see `package.json` `engines`.

src/extract-streams-files.js

@@ -0,0 +1,52 @@
+import extractFiles, { isObject } from 'extract-files'
+
+export function NoFormDataException(message) {
+  this.message = message
+  this.name = 'NoFormDataException'
+}
+
+export const isBrowserOrNative = (function() {
+  try {
+    if (FormData) return true
+  } catch (e) {
+    return false
+  }
+})()
+
+export const isStream = obj => {
+  return (
+    obj &&
+    typeof obj.pipe === 'function' &&
+    typeof obj._read === 'function' &&
+    typeof obj._readableState === 'object' &&
+    obj.readable !== false
+  )
+}
+
+export const extractFilesOrStreams = (tree, treePath) => {
+  if (isBrowserOrNative) return extractFiles(tree)
+  else {
+    if (treePath === void 0) treePath = ''
+    var files = []
+    var recurse = function recurse(node, nodePath) {
+      Object.keys(node).forEach(function(key) {
+        if (!isObject(node[key])) return
+        var path = '' + nodePath + key
+        // get streams and ajdust to busboy obj.stream format
+        if (isStream(node[key]) || node[key] instanceof Promise) {
+          files.push({
+            path: path,
+            file: node[key]
+          })
+          node[key] = null
+          return
+        } else if (node[key].length) node[key] = Array.prototype.slice.call(node[key])
+        recurse(node[key], path + '.')
+      })
+    }
+
+    if (isObject(tree))
+      recurse(tree, treePath === '' ? treePath : treePath + '.')
+    return files
+  }
+}

src/index.js

@@ -7,7 +7,12 @@ import {
   createSignalIfSupported,
   parseAndCheckHttpResponse
 } from 'apollo-link-http-common'
-import extractFiles from 'extract-files'
+import {
+  extractFilesOrStreams,
+  isStream,
+  isBrowserOrNative,
+  NoFormDataException
+} from './extract-streams-files'
 
 export { ReactNativeFile } from 'extract-files'
 
@@ -17,7 +22,8 @@ export const createUploadLink = ({
   fetchOptions,
   credentials,
   headers,
-  includeExtensions
+  includeExtensions,
+  serverFormData
 } = {}) => {
   const linkConfig = {
     http: { includeExtensions },
@@ -43,16 +49,26 @@ export const createUploadLink = ({
       contextConfig
     )
 
-    const files = extractFiles(body)
+    const files = extractFilesOrStreams(body)
     const payload = serializeFetchParameter(body, 'Payload')
 
+    // hold files promises (will only have items for server2server uploads)
+    const promises = []
     if (files.length) {
       // Automatically set by fetch when the body is a FormData instance.
       delete options.headers['content-type']
 
       // GraphQL multipart request spec:
       // https://github.com/jaydenseric/graphql-multipart-request-spec
-      options.body = new FormData()
+      if (isBrowserOrNative) options.body = new FormData()
+      else if (serverFormData)
+        // on the server - expecting to receive a FormData object following the same
+        // specs as browser's FormData - tested with 'form-data' npm package only
+        options.body = new serverFormData()
+      else
+        throw new NoFormDataException(`FormData function doesn't exist on this server version. \
+We suggest you installing 'form-data' via npm and pass it as \
+as an argument in 'createUploadLink' function : '{ serverFormData: FormData }'`)
       options.body.append('operations', payload)
       options.body.append(
         'map',
@@ -63,37 +79,72 @@ export const createUploadLink = ({
           }, {})
         )
       )
-      files.forEach(({ file }, index) =>
-        options.body.append(index, file, file.name)
-      )
+      files.forEach(({ file }, index) => {
+        if (isStream(file))
+          // stream from a 'fs.createReadStream' call
+          options.body.append(index, file)
+        else if (file instanceof Promise)
+          // cover the apollo-upload-server files wrapped in Promises
+          promises.push(
+            new Promise((resolve, reject) => {
+              file
+                .then(file => {
+                  const { filename, mimetype: contentType } = file
+                  const bufs = []
+                  file.stream.on('data', function(buf) {
+                    bufs.push(buf)
+                  })
+                  file.stream.on('end', function() {
+                    const buffer = Buffer.concat(bufs)
+                    const knownLength = buffer.byteLength
+                    options.body.append(index, buffer, {
+                      filename: filename,
+                      contentType,
+                      knownLength
+                    })
+                    resolve()
+                  })
+                  file.stream.on('error', reject)
+                })
+                .catch(reject)
+            })
+          )
+        else options.body.append(index, file, file.name)
+      })
     } else options.body = payload
-
     return new Observable(observer => {
       // Allow aborting fetch, if supported.
       const { controller, signal } = createSignalIfSupported()
       if (controller) options.signal = signal
 
-      linkFetch(uri, options)
-        .then(response => {
-          // Forward the response on the context.
-          operation.setContext({ response })
-          return response
-        })
-        .then(parseAndCheckHttpResponse(operation))
-        .then(result => {
-          observer.next(result)
-          observer.complete()
-        })
-        .catch(error => {
-          if (error.name === 'AbortError')
-            // Fetch was aborted.
-            return
+      // process all FileStream (if any) and submit request
+      Promise.all(promises)
+        .then(() => {
+          linkFetch(uri, options)
+            .then(response => {
+              // Forward the response on the context.
+              operation.setContext({ response })
+              return response
+            })
+            .then(parseAndCheckHttpResponse(operation))
+            .then(result => {
+              observer.next(result)
+              observer.complete()
+            })
+            .catch(error => {
+              if (error.name === 'AbortError')
+                // Fetch was aborted.
+                return
 
-          if (error.result && error.result.errors && error.result.data)
-            // There is a GraphQL result to forward.
-            observer.next(error.result)
+              if (error.result && error.result.errors && error.result.data)
+                // There is a GraphQL result to forward.
+                observer.next(error.result)
 
-          observer.error(error)
+              observer.error(error)
+            })
+        })
+        .catch(e => {
+          throw { message: 'Error while draining stream.', error: e }
         })
 
       // Cleanup function.