feat: concurrent file uploads (#205)

The current implementation upload files one by one, which doesn't perform really well for files smaller than 5MB. With that change, files can be uploaded in parallel, allowing faster upload speeds.

__tests__/boards/actions/__snapshots__/add-files.js.snap

@@ -31,5 +31,3 @@ Array [
   },
 ]
 `;
-
-exports[`Add files action should throw an error if arguments are not provided 1`] = `[Error: There was an error when adding files to the board.]`;

__tests__/boards/actions/add-files.js

@@ -1,5 +1,6 @@
 const routes = require('../../../src/config/routes');
 const addFilesAction = require('../../../src/boards/actions/add-files');
+const enqueueFileTaskAction = require('../../../src/actions/queues/files-queue');
 const { RemoteBoard } = require('../../../src/boards/models');
 
 const createRemoteMockFile = require('../../fixtures/create-remote-file');
@@ -10,7 +11,7 @@ describe('Add files action', () => {
   const mocks = {};
   beforeEach(() => {
     mocks.request = { send: jest.fn() };
-    mocks.uploadFileToBoard = jest.fn();
+    mocks.uploadFile = jest.fn();
     mocks.request.send.mockReturnValue([
       createRemoteMockFile({
         size: 195906,
@@ -25,7 +26,9 @@ describe('Add files action', () => {
     addFiles = addFilesAction({
       routes,
       request: mocks.request,
-      uploadFileToBoard: mocks.uploadFileToBoard,
+      enqueueFileTask: enqueueFileTaskAction({
+        uploadFile: mocks.uploadFile,
+      }),
     });
 
     board = new RemoteBoard({
@@ -42,7 +45,7 @@ describe('Add files action', () => {
   it('should create an upload file request', async () => {
     const files = await addFiles(board, [createLocalMockFile({ content: [] })]);
     expect(files).toMatchSnapshot();
-    expect(mocks.uploadFileToBoard).toHaveBeenCalledWith(
+    expect(mocks.uploadFile).toHaveBeenCalledWith(
       board,
       {
         id: 'random-hash',
@@ -56,15 +59,21 @@ describe('Add files action', () => {
     );
   });
 
+  it('should throw an error if request fails', async () => {
+    mocks.uploadFile.mockImplementation(() =>
+      Promise.reject(new Error('Network error.'))
+    );
+    await expect(
+      addFiles(board, [createLocalMockFile({ content: [] })])
+    ).rejects.toThrow('There was an error when adding files to the board.');
+  });
+
   it('should throw an error if arguments are not provided', async () => {
     mocks.request.send.mockReturnValue(() =>
       Promise.reject(new Error('Network error.'))
     );
-
-    try {
-      await addFiles();
-    } catch (error) {
-      expect(error).toMatchSnapshot();
-    }
+    await expect(addFiles()).rejects.toThrow(
+      'There was an error when adding files to the board.'
+    );
   });
 });

__tests__/transfers/actions/create.js

@@ -10,7 +10,7 @@ describe('Create transfer action', () => {
   const mocks = {};
   beforeEach(() => {
     mocks.request = { send: jest.fn() };
-    mocks.uploadFileToTransfer = jest.fn();
+    mocks.enqueueFileTask = jest.fn(() => Promise.resolve());
     mocks.finalizeTransfer = jest.fn((transfer) => transfer);
 
     mocks.request.send.mockReturnValue({
@@ -24,7 +24,7 @@ describe('Create transfer action', () => {
     create = createAction({
       routes,
       request: mocks.request,
-      uploadFileToTransfer: mocks.uploadFileToTransfer,
+      enqueueFileTask: mocks.enqueueFileTask,
       finalizeTransfer: mocks.finalizeTransfer,
     });
   });
@@ -59,18 +59,18 @@ describe('Create transfer action', () => {
       files: [createLocalMockFile({ content: [] })],
     });
     expect(transfer).toMatchSnapshot();
-    expect(mocks.uploadFileToTransfer).toHaveBeenCalledWith(
-      transfer,
-      {
+    expect(mocks.enqueueFileTask).toHaveBeenCalledWith({
+      transferOrBoard: transfer,
+      file: {
         id: 'random-hash',
         name: 'kittie.gif',
         size: 1024,
         type: 'file',
         multipart: { chunk_size: 1024, id: 'multipart-id', part_numbers: 1 },
         chunks: [],
       },
-      []
-    );
+      content: [],
+    });
   });
 
   it('should create a finalize transfer request if content is provided', async () => {

src/actions/queues/files-queue.js

@@ -0,0 +1,35 @@
+const queue = require('async/queue');
+
+const config = require('../../config');
+const logger = require('../../config/logger');
+
+module.exports = function({ uploadFile }) {
+  // Expand the data from the task to perform a file upload
+  async function processFileTask(task) {
+    await uploadFile(task.transferOrBoard, task.file, task.content);
+  }
+
+  // Create a queue object with a concurrency defined in the configuration.
+  // processFileTask will be executed for every task.
+  const uploadQueue = queue(processFileTask, config.concurrency);
+
+  return function enqueueFileTask(task) {
+    return new Promise((resolve, reject) => {
+      const fileName = task.file.name;
+
+      logger.debug(`[${fileName}] Queuing file to be uploaded.`);
+
+      uploadQueue.push(task, (error) => {
+        if (error) {
+          logger.debug(`[${fileName}] Queue: file failed to upload.`);
+
+          return reject(error);
+        }
+
+        logger.debug(`[${fileName}] Queue: file upload complete.`);
+
+        resolve(task);
+      });
+    });
+  };
+};

src/boards/actions/add-files.js

@@ -4,7 +4,7 @@ const futureFile = require('../models/future-file');
 const { RemoteFile } = require('../../models');
 const contentForFiles = require('../../utils/content-for-files');
 
-module.exports = function({ request, routes, uploadFileToBoard }) {
+module.exports = function({ request, routes, enqueueFileTask }) {
   /**
    * Check if the user also passed the content for files.
    * In that case, we can upload the files in one go.
@@ -18,6 +18,24 @@ module.exports = function({ request, routes, uploadFileToBoard }) {
     );
   }
 
+  /**
+   * Given the content of the files and a remote board, upload all the files.
+   * @param   {Array}   filesContent An array containing the content of each file
+   * @param   {Object}  remoteBoard  A board object
+   * @returns {Promise}
+   */
+  function uploadFiles(files, filesContent, remoteBoard) {
+    return Promise.all(
+      files.map((file) =>
+        enqueueFileTask({
+          transferOrBoard: remoteBoard,
+          file: file,
+          content: filesContent[file.name],
+        })
+      )
+    );
+  }
+
   /**
    * Add files to an existing board.
    * @param   {Object}  board Existing board object
@@ -37,11 +55,7 @@ module.exports = function({ request, routes, uploadFileToBoard }) {
 
       if (shouldUploadFiles(files)) {
         const filesContent = contentForFiles(files);
-        await Promise.all(
-          boardFiles.map((file) => {
-            return uploadFileToBoard(board, file, filesContent[file.name]);
-          })
-        );
+        await uploadFiles(boardFiles, filesContent, board);
       }
 
       return boardFiles;

src/boards/actions/index.js

@@ -27,10 +27,13 @@ const uploadFileToBoard = require('../../actions/upload-file')({
   enqueueChunk,
   completeFileUpload,
 });
+const enqueueFileTask = require('../../actions/queues/files-queue')({
+  uploadFile: uploadFileToBoard,
+});
 const addFilesToBoard = require('./add-files')({
   request,
   routes,
-  uploadFileToBoard,
+  enqueueFileTask,
 });
 const addLinksToBoard = require('./add-links')({ request, routes });
 

src/transfers/actions/create.js

@@ -7,7 +7,7 @@ const { futureTransfer, RemoteTransfer } = require('../models');
 module.exports = function({
   request,
   routes,
-  uploadFileToTransfer,
+  enqueueFileTask,
   finalizeTransfer,
 }) {
   /**
@@ -32,13 +32,13 @@ module.exports = function({
    */
   async function uploadFilesAndFinalize(filesContent, remoteTransfer) {
     await Promise.all(
-      remoteTransfer.files.map((file) => {
-        return uploadFileToTransfer(
-          remoteTransfer,
-          file,
-          filesContent[file.name]
-        );
-      })
+      remoteTransfer.files.map((file) =>
+        enqueueFileTask({
+          transferOrBoard: remoteTransfer,
+          file: file,
+          content: filesContent[file.name],
+        })
+      )
     );
 
     return await finalizeTransfer(remoteTransfer);

src/transfers/actions/index.js

@@ -21,11 +21,14 @@ const uploadFileToTransfer = require('../../actions/upload-file')({
   enqueueChunk,
   completeFileUpload,
 });
+const enqueueFileTask = require('../../actions/queues/files-queue')({
+  uploadFile: uploadFileToTransfer,
+});
 const finalizeTransfer = require('./finalize')({ request, routes });
 const createTransfer = require('./create')({
   request,
   routes,
-  uploadFileToTransfer,
+  enqueueFileTask,
   finalizeTransfer,
 });
 const findTransfer = require('../../actions/find')({