Rewrite Companion providers to use streams to allow simultaneous uplo…

…ad/download without saving to disk (#3159) * rewrite to async/await * Only fetch size (HEAD) if needed #3034 * Update packages/@uppy/companion/src/server/controllers/url.js Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com> * Change HEAD to GET in getURLMeta and abort request immediately upon response headers received #3034 (comment) * fix lint * fix lint * cut off length of file names or else we get "MetadataTooLarge: Your metadata headers exceed the maximum allowed metadata size" in tus / S3 * try to fix flaky test * remove iife and cleanup code a bit * fix lint by reordering code * rename Uploader to MultipartUploader * Rewrite Uploader to use fs-capacitor #3098 This allows for upload to start almost immediately without having to first download the file. And it allows for uploading bigger files, because transloadit assembly will not timeout, as it will get upload progress events all the time. No longer need for illusive progress. Also fix eslint warnings and simplify logic Still TODO: TUS pause/resume has a bug: tus/tus-js-client#275 * add comment in dev Dashboard and pull out variable * fix a bug where remote xhr upload would ignore progress events in the UI * fix bug where s3 multipart cancel wasn't working * fix also cancel for xhr * Rewrite providers to use streams This removes the need for disk space as data will be buffered in memory and backpressure will be respected #3098 (comment) All providers "download" methods will now return a { stream } which can be consumed by uploader. Also: - Remove capacitor (no longer needed) - Change Provider/SearchProvider API to async (Breaking change for custom companion providers) - Fix the case with unknown length streams (zoom / google drive). Need to be downloaded first - rewrite controllers deauth-callback, thumbnail, list, logout to async - getURLMeta: make sure size is never NaN (NaN gets converted to null in JSON.stringify when sent to client but not when used in backend) - fix purest mock (it wasn't returning statusCode on('response')) - add missing http mock for "request" for THUMBNAIL_URL and http://url.myendpoint.com/file (these request errors were never caught by tests previously) - "upload functions with tus protocol" test: move filename checking to new test where size is null. Fix broken expects - fix some lint * Implement streamingUpload flag COMPANION_STREAMING_UPLOAD Default to false due to backward compatibility If set to true, will start to upload files at the same time as dowlnoading them, by piping the streams - Also implement progress for downloading too - and fix progress duplication logic - fix test that assumed file was fully downloaded after first progress event * rearrange validation logic * add COMPANION_STREAMING_UPLOAD to env.test.sh too * implement maxFileSize option in companion for both unknown length and known length downloads * fix bug * fix memory leak when non 200 status streams were being kept * fix lint * Add backward-compatibility for companion providers Implement a new static field "version" on providers, which when not set to 2, will cause a compatibility layer to be added for supporting old callback style provider api also fix some eslint and rename some vars * document new provider API * remove static as it doesn't work on node 10 * try to fix build issue * degrade to node 14 in github actions due to hitting this error: nodejs/node#40030 https://github.com/transloadit/uppy/pull/3159/checks?check_run_id=3544858518 * pull out duplicated logic into reusable function * fix lint * make methods private * re-add unsplash download_location request got lost in merge * add try/catch as suggested #3159 (comment) * Only set default chunkSize if needed for being more compliant with previous behavior when streamingUpload = false * Improve flaky test Trying to fix this error: FAIL packages/@uppy/utils/src/delay.test.js ● delay › should reject when signal is aborted expect(received).toBeLessThan(expected) Expected: < 70 Received: 107 32 | const time = Date.now() - start 33 | expect(time).toBeGreaterThanOrEqual(30) > 34 | expect(time).toBeLessThan(70) | ^ 35 | }) 36 | }) 37 | at Object.<anonymous> (packages/@uppy/utils/src/delay.test.js:34:18) https://github.com/transloadit/uppy/runs/3984613454?check_suite_focus=true * Apply suggestions from code review Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com> * fix review feedback & lint * Apply suggestions from code review Co-authored-by: Merlijn Vos <merlijn@soverin.net> * remove unneeded ts-ignore * Update packages/@uppy/companion/src/server/controllers/url.js Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com> * Update packages/@uppy/companion/src/server/Uploader.js Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com> * reduce nesting * fix lint * optimize promisify #3159 (comment) * Update packages/@uppy/companion/test/__tests__/uploader.js Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com> Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com> Co-authored-by: Merlijn Vos <merlijn@soverin.net>
transloadit · Nov 1, 2021 · 56339fc · 56339fc
1 parent 94e686d
commit 56339fc
Show file tree

Hide file tree

Showing 45 changed files with 1,248 additions and 889 deletions.
diff --git a/examples/custom-provider/server/customprovider.js b/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)
+      })))
   }
 }
 

diff --git a/examples/custom-provider/server/index.js b/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: {

diff --git a/examples/dev/Dashboard.js b/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) {

diff --git a/packages/@uppy/aws-s3-multipart/src/index.js b/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`)
       })

diff --git a/packages/@uppy/companion/KUBERNETES.md b/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://:superSecretPassword@uppy-redis.uppy.svc.cluster.local:6379
   COMPANION_SECRET: "shh!Issa Secret!"
   COMPANION_DROPBOX_KEY: "YOUR DROPBOX KEY"

diff --git a/packages/@uppy/companion/env.test.sh b/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"

diff --git a/packages/@uppy/companion/env_example b/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

diff --git a/packages/@uppy/companion/package.json b/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"
   },

diff --git a/packages/@uppy/companion/src/companion.js b/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