Rewrite Companion providers to use streams to allow simultaneous upload/download without saving to disk

examples/custom-provider/server/customprovider.js

@@ -32,11 +32,14 @@ function adaptData (res) {
  * an example of a custom provider module. It implements @uppy/companion's Provider interface
  */
 class MyCustomProvider {
+  static version = 2
+
   constructor () {
     this.authProvider = 'myunsplash'
   }
 
-  list ({ token, directory }, done) {
+  // eslint-disable-next-line class-methods-use-this
+  async list ({ token, directory }) {
     const path = directory ? `/${directory}/photos` : ''
     const options = {
       url: `${BASE_URL}/collections${path}`,
@@ -47,18 +50,20 @@ class MyCustomProvider {
       },
     }
 
-    request(options, (err, resp, body) => {
-      if (err) {
-        console.log(err)
-        done(err)
-        return
-      }
+    return new Promise((resolve, reject) => (
+      request(options, (err, resp, body) => {
+        if (err) {
+          console.log(err)
+          reject(err)
+          return
+        }
 
-      done(null, adaptData(body))
-    })
+        resolve(adaptData(body))
+      })))
   }
 
-  download ({ id, token }, onData) {
+  // eslint-disable-next-line class-methods-use-this
+  async download ({ id, token }) {
     const options = {
       url: `${BASE_URL}/photos/${id}`,
       method: 'GET',
@@ -68,21 +73,30 @@ class MyCustomProvider {
       },
     }
 
-    request(options, (err, resp, body) => {
-      if (err) {
-        console.log(err)
-        return
-      }
-
-      const url = body.links.download
-      request.get(url)
-        .on('data', (chunk) => onData(null, chunk))
-        .on('end', () => onData(null, null))
-        .on('error', (err) => console.log(err))
+    const resp = await new Promise((resolve, reject) => {
+      const req = request(options)
+        .on('response', (response) => {
+          // Don't allow any more data to flow yet.
+          // https://github.com/request/request/issues/1990#issuecomment-184712275
+          response.pause()
+
+          if (resp.statusCode !== 200) {
+            req.abort() // Or we will leak memory
+            reject(new Error(`HTTP response ${resp.statusCode}`))
+            return
+          }
+
+          resolve(response)
+        })
+        .on('error', reject)
     })
+
+    // The returned stream will be consumed and uploaded from the current position
+    return { stream: resp }
   }
 
-  size ({ id, token }, done) {
+  // eslint-disable-next-line class-methods-use-this
+  async size ({ id, token }) {
     const options = {
       url: `${BASE_URL}/photos/${id}`,
       method: 'GET',
@@ -92,15 +106,16 @@ class MyCustomProvider {
       },
     }
 
-    request(options, (err, resp, body) => {
-      if (err) {
-        console.log(err)
-        done(err)
-        return
-      }
+    return new Promise((resolve, reject) => (
+      request(options, (err, resp, body) => {
+        if (err) {
+          console.log(err)
+          reject(err)
+          return
+        }
 
-      done(null, body.width * body.height)
-    })
+        resolve(body.size)
+      })))
   }
 }
 

examples/custom-provider/server/index.js

@@ -4,6 +4,7 @@ const express = require('express')
 const bodyParser = require('body-parser')
 const session = require('express-session')
 const uppy = require('../../../packages/@uppy/companion')
+const MyCustomProvider = require('./customprovider')
 
 const app = express()
 
@@ -42,8 +43,8 @@ const uppyOptions = {
         key: 'your unsplash key here',
         secret: 'your unsplash secret here',
       },
-      // you provider module
-      module: require('./customprovider'),
+      // you provider class/module:
+      module: MyCustomProvider,
     },
   },
   server: {

examples/dev/Dashboard.js

@@ -29,6 +29,8 @@ const DropTarget = require('@uppy/drop-target/src')
 const UPLOADER = 'tus'
 // const UPLOADER = 's3'
 // const UPLOADER = 's3-multipart'
+// xhr will use protocol 'multipart' in companion, if used with a remote service, e.g. google drive.
+// If local upload will use browser XHR
 // const UPLOADER = 'xhr'
 // const UPLOADER = 'transloadit'
 // const UPLOADER = 'transloadit-s3'
@@ -44,6 +46,7 @@ const XHR_ENDPOINT = 'https://xhr-server.herokuapp.com/upload'
 
 const TRANSLOADIT_KEY = '...'
 const TRANSLOADIT_TEMPLATE = '...'
+const TRANSLOADIT_SERVICE_URL = 'https://api2.transloadit.com'
 
 // DEV CONFIG: enable or disable Golden Retriever
 
@@ -109,6 +112,7 @@ module.exports = () => {
       break
     case 'transloadit':
       uppyDashboard.use(Transloadit, {
+        service: TRANSLOADIT_SERVICE_URL,
         waitForEncoding: true,
         params: {
           auth: { key: TRANSLOADIT_KEY },
@@ -141,6 +145,7 @@ module.exports = () => {
         bundle: true,
       })
       break
+    default:
   }
 
   if (RESTORE) {

packages/@uppy/aws-s3-multipart/src/index.js

@@ -4,7 +4,7 @@ const EventTracker = require('@uppy/utils/lib/EventTracker')
 const emitSocketProgress = require('@uppy/utils/lib/emitSocketProgress')
 const getSocketHost = require('@uppy/utils/lib/getSocketHost')
 const { RateLimitedQueue } = require('@uppy/utils/lib/RateLimitedQueue')
-const Uploader = require('./MultipartUploader')
+const MultipartUploader = require('./MultipartUploader')
 
 function assertServerError (res) {
   if (res && res.error) {
@@ -187,7 +187,7 @@ module.exports = class AwsS3Multipart extends BasePlugin {
         this.uppy.emit('s3-multipart:part-uploaded', cFile, part)
       }
 
-      const upload = new Uploader(file.data, {
+      const upload = new MultipartUploader(file.data, {
         // .bind to pass the file object to each handler.
         createMultipartUpload: this.opts.createMultipartUpload.bind(this, file),
         listParts: this.opts.listParts.bind(this, file),
@@ -320,7 +320,7 @@ module.exports = class AwsS3Multipart extends BasePlugin {
 
       this.onFileRemove(file.id, () => {
         queuedRequest.abort()
-        socket.send('pause', {})
+        socket.send('cancel', {})
         this.resetUploaderReferences(file.id, { abort: true })
         resolve(`upload ${file.id} was removed`)
       })
@@ -348,7 +348,7 @@ module.exports = class AwsS3Multipart extends BasePlugin {
 
       this.onCancelAll(file.id, () => {
         queuedRequest.abort()
-        socket.send('pause', {})
+        socket.send('cancel', {})
         this.resetUploaderReferences(file.id)
         resolve(`upload ${file.id} was canceled`)
       })

packages/@uppy/companion/KUBERNETES.md

@@ -25,6 +25,7 @@ data:
   COMPANION_DOMAIN: "YOUR SERVER DOMAIN"
   COMPANION_DOMAINS: "sub1.domain.com,sub2.domain.com,sub3.domain.com"
   COMPANION_PROTOCOL: "YOUR SERVER PROTOCOL"
+  COMPANION_STREAMING_UPLOAD: true
   COMPANION_REDIS_URL: redis://:[email protected]:6379
   COMPANION_SECRET: "shh!Issa Secret!"
   COMPANION_DROPBOX_KEY: "YOUR DROPBOX KEY"

packages/@uppy/companion/env.test.sh

@@ -5,6 +5,8 @@ export COMPANION_SELF_ENDPOINT="localhost:3020"
 export COMPANION_HIDE_METRICS="false"
 export COMPANION_HIDE_WELCOME="false"
 
+export COMPANION_STREAMING_UPLOAD="true"
+
 export COMPANION_PROTOCOL="http"
 export COMPANION_DATADIR="./test/output"
 export COMPANION_SECRET="secret"

packages/@uppy/companion/env_example

@@ -4,6 +4,7 @@ COMPANION_DOMAIN=uppy.xxxx.com
 COMPANION_SELF_ENDPOINT=uppy.xxxx.com
 COMPANION_HIDE_METRICS=false
 COMPANION_HIDE_WELCOME=false
+COMPANION_STREAMING_UPLOAD=true
 
 COMPANION_PROTOCOL=https
 COMPANION_DATADIR=/mnt/uppy-server-data

packages/@uppy/companion/package.json

@@ -85,6 +85,8 @@
     "@types/uuid": "3.4.7",
     "@types/webpack": "^5.28.0",
     "@types/ws": "6.0.4",
+    "into-stream": "^6.0.0",
+    "nock": "^13.1.3",
     "supertest": "3.4.2",
     "typescript": "~4.3"
   },

packages/@uppy/companion/src/companion.js

@@ -39,6 +39,7 @@ const defaultOptions = {
   },
   debug: true,
   logClientVersion: true,
+  streamingUpload: false,
 }
 
 // make the errors available publicly for custom providers