Make file processing queue-based

Author: Ivan Duplenskikh

Tasks/CopyFilesOverSSHV0/copyfilesoverssh.ts

@@ -4,6 +4,7 @@ import * as tl from 'azure-pipelines-task-lib/task';
 import * as minimatch from 'minimatch';
 import * as utils from './utils';
 import { SshHelper } from './sshhelper';
+import Queue from './queue';
 
 // This method will find the list of matching files for the specified contents
 // This logic is the same as the one used by CopyFiles task except for allowing dot folders to be copied
@@ -133,192 +134,190 @@ function getUniqueFolders(filesToCopy: string[]) {
 }
 
 async function run() {
-    let sshHelper: SshHelper;
-
-    try {
-        tl.setResourcePath(path.join(__dirname, 'task.json'));
-
-        // read SSH endpoint input
-        const sshEndpoint = tl.getInput('sshEndpoint', true);
-        const username: string = tl.getEndpointAuthorizationParameter(sshEndpoint, 'username', false);
-        const password: string = tl.getEndpointAuthorizationParameter(sshEndpoint, 'password', true); //passphrase is optional
-        const privateKey: string = process.env['ENDPOINT_DATA_' + sshEndpoint + '_PRIVATEKEY']; //private key is optional, password can be used for connecting
-        const hostname: string = tl.getEndpointDataParameter(sshEndpoint, 'host', false);
-        let port: string = tl.getEndpointDataParameter(sshEndpoint, 'port', true); //port is optional, will use 22 as default port if not specified
-        if (!port) {
-            console.log(tl.loc('UseDefaultPort'));
-            port = '22';
+    tl.setResourcePath(path.join(__dirname, 'task.json'));
+
+    // Read SSH endpoint input
+    const sshEndpoint = tl.getInput('sshEndpoint', true);
+    const username = tl.getEndpointAuthorizationParameter(sshEndpoint, 'username', false);
+    // Passphrase is optional
+    const password = tl.getEndpointAuthorizationParameter(sshEndpoint, 'password', true);
+    // Private key is optional, password can be used for connecting
+    const privateKey = process.env['ENDPOINT_DATA_' + sshEndpoint + '_PRIVATEKEY'];
+    const hostname = tl.getEndpointDataParameter(sshEndpoint, 'host', false);
+    // Port is optional, will use 22 as default port if not specified
+    let port = tl.getEndpointDataParameter(sshEndpoint, 'port', true);
+
+    if (!port) {
+        console.log(tl.loc('UseDefaultPort'));
+        port = '22';
+    }
+
+    const readyTimeout = parseInt(tl.getInput('readyTimeout', true), 10);
+    const useFastPut = !(process.env['USE_FAST_PUT'] === 'false');
+
+    // Set up the SSH connection configuration based on endpoint details
+    let sshConfig: Object = {
+        host: hostname,
+        port: port,
+        username: username,
+        readyTimeout: readyTimeout,
+        useFastPut: useFastPut
+    };
+
+    if (privateKey) {
+        tl.debug('Using private key for ssh connection.');
+
+        sshConfig = {
+            ...sshConfig,
+            privateKey,
+            password
         }
+    } else {
+        // Use password
+        tl.debug('Using username and password for ssh connection.');
 
-        const readyTimeout = getReadyTimeoutVariable();
-        const useFastPut: boolean = !(process.env['USE_FAST_PUT'] === 'false');
-
-        // set up the SSH connection configuration based on endpoint details
-        let sshConfig;
-        if (privateKey) {
-            tl.debug('Using private key for ssh connection.');
-            sshConfig = {
-                host: hostname,
-                port: port,
-                username: username,
-                privateKey: privateKey,
-                passphrase: password,
-                readyTimeout: readyTimeout,
-                useFastPut: useFastPut
-            }
-        } else {
-            // use password
-            tl.debug('Using username and password for ssh connection.');
-            sshConfig = {
-                host: hostname,
-                port: port,
-                username: username,
-                password: password,
-                readyTimeout: readyTimeout,
-                useFastPut: useFastPut
-            }
+        sshConfig = {
+            ...sshConfig,
+            password,
         }
+    }
 
-        // contents is a multiline input containing glob patterns
-        const contents: string[] = tl.getDelimitedInput('contents', '\n', true);
-        const sourceFolder: string = tl.getPathInput('sourceFolder', true, true);
-        let targetFolder: string = tl.getInput('targetFolder');
+    // Contents is a multiline input containing glob patterns
+    const contents = tl.getDelimitedInput('contents', '\n', true);
+    const sourceFolder = tl.getPathInput('sourceFolder', true, true);
+    let targetFolder = tl.getInput('targetFolder');
 
-        if (!targetFolder) {
-            targetFolder = "./";
-        } else {
-            // '~/' is unsupported
-            targetFolder = targetFolder.replace(/^~\//, "./");
-        }
+    if (!targetFolder) {
+        targetFolder = "./";
+    } else {
+        // '~/' is unsupported
+        targetFolder = targetFolder.replace(/^~\//, "./");
+    }
 
-        // read the copy options
-        const cleanTargetFolder: boolean = tl.getBoolInput('cleanTargetFolder', false);
-        const overwrite: boolean = tl.getBoolInput('overwrite', false);
-        const failOnEmptySource: boolean = tl.getBoolInput('failOnEmptySource', false);
-        const flattenFolders: boolean = tl.getBoolInput('flattenFolders', false);
+    // Read the copy options
+    const cleanTargetFolder = tl.getBoolInput('cleanTargetFolder', false);
+    const overwrite = tl.getBoolInput('overwrite', false);
+    const failOnEmptySource = tl.getBoolInput('failOnEmptySource', false);
+    const flattenFolders = tl.getBoolInput('flattenFolders', false);
 
-        if (!tl.stats(sourceFolder).isDirectory()) {
-            throw tl.loc('SourceNotFolder');
-        }
+    if (!tl.stats(sourceFolder).isDirectory()) {
+        tl.setResult(tl.TaskResult.Failed, tl.loc('SourceNotFolder'));
+        return;
+    }
 
-        // initialize the SSH helpers, set up the connection
-        sshHelper = new SshHelper(sshConfig);
-        await sshHelper.setupConnection();
-
-        if (cleanTargetFolder && await sshHelper.checkRemotePathExists(targetFolder)) {
-            console.log(tl.loc('CleanTargetFolder', targetFolder));
-            const isWindowsOnTarget: boolean = tl.getBoolInput('isWindowsOnTarget', false);
-            const cleanHiddenFilesInTarget: boolean = tl.getBoolInput('cleanHiddenFilesInTarget', false);
-            const cleanTargetFolderCmd: string = utils.getCleanTargetFolderCmd(targetFolder, isWindowsOnTarget, cleanHiddenFilesInTarget);
-            try {
-                await sshHelper.runCommandOnRemoteMachine(cleanTargetFolderCmd, null);
-            } catch (err) {
-                throw tl.loc('CleanTargetFolderFailed', err);
-            }
+    // Initialize the SSH helpers, set up the connection
+    const sshHelper = new SshHelper(sshConfig);
+    await sshHelper.setupConnection();
+
+    if (cleanTargetFolder && await sshHelper.checkRemotePathExists(targetFolder)) {
+        console.log(tl.loc('CleanTargetFolder', targetFolder));
+        const isWindowsOnTarget = tl.getBoolInput('isWindowsOnTarget', false);
+        const cleanHiddenFilesInTarget = tl.getBoolInput('cleanHiddenFilesInTarget', false);
+        const cleanTargetFolderCmd = utils.getCleanTargetFolderCmd(targetFolder, isWindowsOnTarget, cleanHiddenFilesInTarget);
+
+        try {
+            await sshHelper.runCommandOnRemoteMachine(cleanTargetFolderCmd, null);
+        } catch (error) {
+            tl.setResult(tl.TaskResult.Failed, tl.loc('CleanTargetFolderFailed', error));
+            tl.debug('Closing the client connection');
+            await sshHelper.closeConnection();
+            return;
         }
+    }
 
-        if (contents.length === 1 && contents[0] === "**") {
-            tl.debug("Upload a directory to a remote machine");
+    if (contents.length === 1 && contents[0] === "**") {
+        tl.debug("Upload a directory to a remote machine");
 
-            try {
-                const completedDirectory = await sshHelper.uploadFolder(sourceFolder, targetFolder);
-                console.log(tl.loc('CopyDirectoryCompleted', completedDirectory));
-            } catch (err) {
-                throw tl.loc("CopyDirectoryFailed", sourceFolder, err);
-            }
-        } else {
-            // identify the files to copy
-            const filesToCopy = getFilesToCopy(sourceFolder, contents);
+        try {
+            const completedDirectory = await sshHelper.uploadFolder(sourceFolder, targetFolder);
+            tl.setResult(tl.TaskResult.Succeeded, tl.loc('CopyDirectoryCompleted', completedDirectory));
+        } catch (error) {
+            tl.setResult(tl.TaskResult.Failed, tl.loc("CopyDirectoryFailed", sourceFolder, error));
+        }
 
-            // copy files to remote machine
-            if (filesToCopy) {
-                const preparedFiles = prepareFiles(filesToCopy, sourceFolder, targetFolder, flattenFolders);
+        tl.debug('Closing the client connection');
+        await sshHelper.closeConnection();
+        return;
+    }
 
-                tl.debug('Number of files to copy = ' + preparedFiles.length);
-                tl.debug('filesToCopy = ' + preparedFiles);
+    // Identify the files to copy
+    const filesToCopy = getFilesToCopy(sourceFolder, contents);
 
-                console.log(tl.loc('CopyingFiles', preparedFiles.length));
+    // Copy files to remote machine
+    if (filesToCopy.length === 0) {
+        if (failOnEmptySource) {
+            tl.setResult(tl.TaskResult.Failed, tl.loc('NothingToCopy'));
+            return;
+        } else {
+            tl.warning(tl.loc('NothingToCopy'));
+            return;
+        }
+    }
 
-                // Create remote folders structure
-                const folderStructure = getUniqueFolders(preparedFiles.map(x => x[1]).sort());
+    const preparedFiles = prepareFiles(filesToCopy, sourceFolder, targetFolder, flattenFolders);
 
-                for (const foldersPath of folderStructure) {
-                    try {
-                        await sshHelper.createRemoteDirectory(foldersPath);
-                        console.log(tl.loc("FolderCreated", foldersPath));
-                    } catch (error) {
-                        throw tl.loc('TargetNotCreated', foldersPath, error);
-                    }
-                }
+    tl.debug(`Number of files to copy = ${preparedFiles.length}`);
+    tl.debug(`filesToCopy = ${preparedFiles}`);
 
-                console.log(tl.loc("FoldersCreated", folderStructure.length));
+    console.log(tl.loc('CopyingFiles', preparedFiles.length));
 
-                // Upload files to remote machine
-                const chunkSize = 10;
-                const chunks = Math.ceil(preparedFiles.length / chunkSize);
-                let successedFiles = 0;
+    // Create remote folders structure
+    const folderStructure = getUniqueFolders(preparedFiles.map(x => x[1]).sort());
 
-                for (let i = 0; i < chunks; i++) {
-                    // Splice files to copy into chunks
-                    const chunk = preparedFiles.slice(i * chunkSize, i * chunkSize + chunkSize);
+    for (const foldersPath of folderStructure) {
+        try {
+            await sshHelper.createRemoteDirectory(foldersPath);
+            console.log(tl.loc("FolderCreated", foldersPath));
+        } catch (error) {
+            await sshHelper.closeConnection();
+            tl.setResult(tl.TaskResult.Failed, tl.loc('TargetNotCreated', foldersPath, error));
+            return;
+        }
+    }
 
-                    const results = await Promise.allSettled(chunk.map(async (pathTuple) => {
-                        const [ fileToCopy, targetPath ] = pathTuple;
+    console.log(tl.loc("FoldersCreated", folderStructure.length));
 
-                        tl.debug('fileToCopy = ' + fileToCopy);
+    // Upload files to remote machine
+    const q = new Queue({
+        concurrent: 5,
+        delay: 0,
+        throwOnError: false
+    });
 
-                        console.log(tl.loc('StartedFileCopy', fileToCopy, targetPath));
+    q.enqueue(preparedFiles.map((pathTuple) => async () => {
+        const [ fileToCopy, targetPath ] = pathTuple;
 
-                        if (!overwrite) {
-                            const fileExists: boolean = await sshHelper.checkRemotePathExists(targetPath);
+        tl.debug('fileToCopy = ' + fileToCopy);
 
-                            if (fileExists) {
-                                throw tl.loc('FileExists', targetPath);
-                            }
-                        }
+        console.log(tl.loc('StartedFileCopy', fileToCopy, targetPath));
 
-                        return sshHelper.uploadFile(fileToCopy, targetPath);
-                    }));
+        if (!overwrite && await sshHelper.checkRemotePathExists(targetPath)) {
+            throw tl.loc('FileExists', targetPath);
+        }
 
-                    var errors = results.filter(p => p.status === 'rejected') as PromiseRejectedResult[];
+        return await sshHelper.uploadFile(fileToCopy, targetPath);
+    }));
 
-                    if (errors.length > 0) {
-                        errors.forEach(x => tl.error(x.reason));
-                        tl.setResult(tl.TaskResult.Failed, tl.loc('NumberFailed', errors.length));
-                        return;
-                    }
+    const errors = [];
+    let successfullyCopiedFilesCount = 0;
 
-                    successedFiles += results.filter(p => p.status === 'fulfilled').length;
-                }
+    q.on('reject', (error) => errors.push(error));
+    q.on('resolve', () => successfullyCopiedFilesCount++);
+    q.on('empty', () => tl.debug('Queue is empty'));
+    q.on('end', async () => {
+        tl.debug('End of the queue processing');
+        tl.debug(`Errors count ${errors.length}`);
 
-                console.log(tl.loc('CopyCompleted', successedFiles));
-            } else if (failOnEmptySource) {
-                throw tl.loc('NothingToCopy');
-            } else {
-                tl.warning(tl.loc('NothingToCopy'));
-            }
-        }
-    } catch (err) {
-        tl.setResult(tl.TaskResult.Failed, err);
-    } finally {
-        // close the client connection to halt build execution
-        if (sshHelper) {
-            tl.debug('Closing the client connection');
+        if (errors.length > 0) {
+            errors.forEach(x => tl.error(x));
             await sshHelper.closeConnection();
+            tl.setResult(tl.TaskResult.Failed, tl.loc('NumberFailed', errors.length));
+            return;
         }
-    }
-}
 
-run().then(() => {
-        tl.debug('Task successfully accomplished');
-    })
-    .catch(err => {
-        tl.debug('Run was unexpectedly failed due to: ' + err);
+        await sshHelper.closeConnection();
+        tl.setResult(tl.TaskResult.Succeeded, tl.loc('CopyCompleted', successfullyCopiedFilesCount));
     });
-
-function getReadyTimeoutVariable(): number {
-    let readyTimeoutString: string = tl.getInput('readyTimeout', true);
-    const readyTimeout: number = parseInt(readyTimeoutString, 10);
-
-    return readyTimeout;
 }
+
+run();