feat(DOSUpload): add upload queue concurrency

Related #12

Tasks/DigitalOceanSpacesUpload/utils/Upload.ts

@@ -6,6 +6,7 @@ import tl from './tl'
 import { Spaces } from '../common/Spaces'
 import { Parameters } from './Parameters'
 import { findFiles, getMimeTypes } from './utils'
+const { default: PQueue } = require('p-queue')
 import prettyBytes = require('pretty-bytes')
 
 interface NormalizePathParameters {
@@ -15,6 +16,14 @@ interface NormalizePathParameters {
   digitalTargetFolder?: string
 }
 
+interface UploadFileParameters {
+  filePath: string
+  targetPath: string
+  digitalAcl: string
+  digitalBucket: string
+  contentType: string
+}
+
 export class Upload extends Spaces<Parameters> {
   constructor(params: Parameters) {
     super(params)
@@ -39,43 +48,52 @@ export class Upload extends Spaces<Parameters> {
       return
     }
 
+    const uploadQueue = new PQueue({ concurrency: 4 })
+
     for (const filePath of files) {
       const targetPath = this.normalizeKeyPath({ ...this.params, filePath })
 
-      try {
-        const contentType = getMimeTypes({
-          filePath,
-          digitalContentType: this.params.digitalContentType,
-        })
-
-        console.log(tl.loc('UploadingFile', filePath, targetPath, contentType))
-
-        const params: AWS.S3.PutObjectRequest = {
-          Bucket: this.params.digitalBucket,
-          ACL: this.params.digitalAcl,
-          Key: targetPath,
-          Body: fs.createReadStream(filePath),
-          ContentType: contentType,
+      const contentType = getMimeTypes({
+        filePath,
+        digitalContentType: this.params.digitalContentType,
+      })
+
+      console.log(tl.loc('UploadingFile', filePath, targetPath, contentType))
+
+      uploadQueue.add(async () => {
+        try {
+          await this.uploadFile({
+            filePath,
+            targetPath,
+            digitalAcl: this.params.digitalAcl,
+            digitalBucket: this.params.digitalBucket,
+            contentType,
+          })
+          console.log(tl.loc('FileUploadCompleted', filePath, targetPath))
+        } catch (error) {
+          console.error(tl.loc('FileUploadFailed'), error)
+          throw error
         }
-
-        await this.uploadFiles(params)
-
-        console.log(tl.loc('FileUploadCompleted', filePath, targetPath))
-      } catch (err) {
-        console.error(tl.loc('FileUploadFailed'), err)
-        throw err
-      }
+      })
     }
 
+    await uploadQueue.onIdle()
+
     console.log(tl.loc('TaskCompleted'))
   }
 
-  async uploadFiles(
-    objectRequest: AWS.S3.PutObjectRequest
+  async uploadFile(
+    params: UploadFileParameters
   ): Promise<AWS.S3.ManagedUpload.SendData> {
-    const request: AWS.S3.ManagedUpload = this.s3Connection.upload(
-      objectRequest
-    )
+    const sendParams: AWS.S3.PutObjectRequest = {
+      Bucket: params.digitalBucket,
+      ACL: params.digitalAcl,
+      Key: params.targetPath,
+      Body: fs.createReadStream(params.filePath),
+      ContentType: params.contentType,
+    }
+
+    const request: AWS.S3.ManagedUpload = this.s3Connection.upload(sendParams)
 
     request.on('httpUploadProgress', (progress) => {
       console.log(

Tests/DigitalOceanSpacesUpload/utils/Upload.spec.ts

@@ -101,9 +101,7 @@ describe('DOSUpload', () => {
       { type: 'return', value: 'file2-v1.3.1.json' },
     ])
 
-    expect(spyLog.mock.calls[3][0]).toEqual(
-      'Upload progress is 1.34 kB of 2.34 kB - 57%'
-    )
+    expect(spyLog.mock.calls.sort()).toMatchSnapshot()
     expect(spyLog).toHaveBeenCalledTimes(15)
   })
 

Tests/DigitalOceanSpacesUpload/utils/__snapshots__/Upload.spec.ts.snap

@@ -0,0 +1,51 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`DOSUpload should upload file successfully 1`] = `
+Array [
+  Array [
+    "All uploads to Space completed",
+  ],
+  Array [
+    "Completed upload of Tests/fixtures/file-v1.0.1.txt to file-v1.0.1.txt",
+  ],
+  Array [
+    "Completed upload of Tests/fixtures/file1-v1.2.1.txt to file1-v1.2.1.txt",
+  ],
+  Array [
+    "Completed upload of Tests/fixtures/file2-v1.3.1.json to file2-v1.3.1.json",
+  ],
+  Array [
+    "Searching ./Tests/fixtures/ for files to upload",
+  ],
+  Array [
+    "Upload progress is 1.34 kB of 2.34 kB - 57%",
+  ],
+  Array [
+    "Upload progress is 1.34 kB of 2.34 kB - 57%",
+  ],
+  Array [
+    "Upload progress is 1.34 kB of 2.34 kB - 57%",
+  ],
+  Array [
+    "Upload progress is 2.34 kB of 2.34 kB - 100%",
+  ],
+  Array [
+    "Upload progress is 2.34 kB of 2.34 kB - 100%",
+  ],
+  Array [
+    "Upload progress is 2.34 kB of 2.34 kB - 100%",
+  ],
+  Array [
+    "Uploading files from ./Tests/fixtures/ to root in bucket test",
+  ],
+  Array [
+    "Uploading matched file Tests/fixtures/file-v1.0.1.txt in file-v1.0.1.txt text/plain",
+  ],
+  Array [
+    "Uploading matched file Tests/fixtures/file1-v1.2.1.txt in file1-v1.2.1.txt text/plain",
+  ],
+  Array [
+    "Uploading matched file Tests/fixtures/file2-v1.3.1.json in file2-v1.3.1.json application/octet-stream",
+  ],
+]
+`;