Don't close socket while upload is still in progress (#4479)

* Don't close socket while upload is still in progress

* Don't close sockets randomly in AWS S3 either, close on error/cancel

* First send the socket message, then call .abort()

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

@@ -716,8 +716,8 @@ export default class AwsS3Multipart extends BasePlugin {
       this.uploaderEvents[file.id] = new EventManager(this.uppy)
 
       this.onFileRemove(file.id, () => {
-        queuedRequest.abort()
         socket.send('cancel', {})
+        queuedRequest.abort()
         this.resetUploaderReferences(file.id, { abort: true })
         resolve(`upload ${file.id} was removed`)
       })
@@ -725,8 +725,8 @@ export default class AwsS3Multipart extends BasePlugin {
       this.onFilePause(file.id, (isPaused) => {
         if (isPaused) {
           // Remove this file from the queue so another file can start in its place.
-          queuedRequest.abort()
           socket.send('pause', {})
+          queuedRequest.abort()
         } else {
           // Resuming an upload should be queued, else you could pause and then
           // resume a queued upload to make it skip the queue.
@@ -734,20 +734,22 @@ export default class AwsS3Multipart extends BasePlugin {
           queuedRequest = this.requests.run(() => {
             socket.open()
             socket.send('resume', {})
-            return () => socket.close()
+            return () => {}
           })
         }
       })
 
       this.onPauseAll(file.id, () => {
-        queuedRequest.abort()
+        // First send the message, then call .abort,
+        // just to make sure socket is not closed, which .abort used to do
         socket.send('pause', {})
+        queuedRequest.abort()
       })
 
       this.onCancelAll(file.id, ({ reason } = {}) => {
         if (reason === 'user') {
-          queuedRequest.abort()
           socket.send('cancel', {})
+          queuedRequest.abort()
           this.resetUploaderReferences(file.id)
         }
         resolve(`upload ${file.id} was canceled`)
@@ -762,7 +764,7 @@ export default class AwsS3Multipart extends BasePlugin {
           socket.open()
           socket.send('resume', {})
 
-          return () => socket.close()
+          return () => {}
         })
       })
 
@@ -789,6 +791,7 @@ export default class AwsS3Multipart extends BasePlugin {
       socket.on('error', (errData) => {
         this.uppy.emit('upload-error', file, new Error(errData.error))
         this.resetUploaderReferences(file.id)
+        socket.close()
         queuedRequest.done()
         reject(new Error(errData.error))
       })
@@ -800,6 +803,7 @@ export default class AwsS3Multipart extends BasePlugin {
 
         this.uppy.emit('upload-success', file, uploadResp)
         this.resetUploaderReferences(file.id)
+        socket.close()
         queuedRequest.done()
         resolve()
       })
@@ -811,7 +815,7 @@ export default class AwsS3Multipart extends BasePlugin {
           socket.open()
         }
 
-        return () => socket.close()
+        return () => {}
       })
     })
   }

packages/@uppy/tus/src/index.js

@@ -498,8 +498,8 @@ export default class Tus extends BasePlugin {
       let queuedRequest
 
       this.onFileRemove(file.id, () => {
-        queuedRequest.abort()
         socket.send('cancel', {})
+        queuedRequest.abort()
         this.resetUploaderReferences(file.id)
         resolve(`upload ${file.id} was removed`)
       })
@@ -507,8 +507,8 @@ export default class Tus extends BasePlugin {
       this.onPause(file.id, (isPaused) => {
         if (isPaused) {
           // Remove this file from the queue so another file can start in its place.
-          queuedRequest.abort()
           socket.send('pause', {})
+          queuedRequest.abort()
         } else {
           // Resuming an upload should be queued, else you could pause and then
           // resume a queued upload to make it skip the queue.
@@ -517,20 +517,20 @@ export default class Tus extends BasePlugin {
             socket.open()
             socket.send('resume', {})
 
-            return () => socket.close()
+            return () => {}
           })
         }
       })
 
       this.onPauseAll(file.id, () => {
-        queuedRequest.abort()
         socket.send('pause', {})
+        queuedRequest.abort()
       })
 
       this.onCancelAll(file.id, ({ reason } = {}) => {
         if (reason === 'user') {
-          queuedRequest.abort()
           socket.send('cancel', {})
+          queuedRequest.abort()
           this.resetUploaderReferences(file.id)
         }
         resolve(`upload ${file.id} was canceled`)
@@ -545,7 +545,7 @@ export default class Tus extends BasePlugin {
           socket.open()
           socket.send('resume', {})
 
-          return () => socket.close()
+          return () => {}
         })
       })
 
@@ -599,7 +599,7 @@ export default class Tus extends BasePlugin {
         this.uppy.emit('upload-success', file, uploadResp)
         this.resetUploaderReferences(file.id)
         queuedRequest.done()
-
+        socket.close()
         resolve()
       })
 
@@ -616,7 +616,7 @@ export default class Tus extends BasePlugin {
         // that point this cancellation function is not going to be called.
         // Also, we need to remove the request from the queue _without_ destroying everything
         // related to this upload to handle pauses.
-        return () => socket.close()
+        return () => {}
       })
     })
   }

packages/@uppy/xhr-upload/src/index.js

@@ -435,6 +435,7 @@ export default class XHRUpload extends BasePlugin {
             : Object.assign(new Error(errData.error.message), { cause: errData.error })
           this.uppy.emit('upload-error', file, error)
           queuedRequest.done() // eslint-disable-line no-use-before-define
+          socket.close()
           if (this.uploaderEvents[file.id]) {
             this.uploaderEvents[file.id].remove()
             this.uploaderEvents[file.id] = null
@@ -451,11 +452,12 @@ export default class XHRUpload extends BasePlugin {
           createSocket()
         }
 
-        return () => socket.close()
+        return () => {}
       })
 
       this.onFileRemove(file.id, () => {
         socket?.send('cancel', {})
+        socket.close()
         queuedRequest.abort()
         resolve(`upload ${file.id} was removed`)
       })
@@ -464,6 +466,7 @@ export default class XHRUpload extends BasePlugin {
         if (reason === 'user') {
           socket?.send('cancel', {})
           queuedRequest.abort()
+          // socket.close()
         }
         resolve(`upload ${file.id} was canceled`)
       })
@@ -472,19 +475,13 @@ export default class XHRUpload extends BasePlugin {
         if (socket == null) {
           queuedRequest.abort()
         } else {
-          socket.send('pause', {})
           queuedRequest.done()
         }
         queuedRequest = this.requests.run(() => {
-          if (!file.isPaused) {
-            if (socket == null) {
-              createSocket()
-            } else {
-              socket.send('resume', {})
-            }
+          if (socket == null) {
+            createSocket()
           }
-
-          return () => socket.close()
+          return () => {}
         })
       }
       this.onRetry(file.id, onRetryRequest)