put-assets

var winston = require('winston');
var commander = require('commander');
var readline = require('readline');
var fs = require('fs');
var RateLimiter = require('limiter').RateLimiter;
var Amplience = require('node-amplience-api');
var EOL = require('os').EOL;

commander
    .version('0.0.0')
    .usage('[options] - input & output can be piped or directed to a file using --input and --output')
    .option('-l, --log [value]', 'output file for errors and warnings')
    .option('-i, --input [value]', 'input file if not std-in')
    .option('-o, --output [value]', 'output file if not std-out')
    .option('-c, --concurrent-api-calls [value]', 'how many concurrent api requests to process (defaults to 5)')
    .option('-r, --rate <n>', 'api requests per minute rate')
    .option('-a, --api [value]', 'api url e.g. http://qa-3-dam-api-ssl.adis.ws')
    .option('-u --username [value]', 'api username')
    .option('-p --password [value]', 'api password')
    .option('-m, --mode [value]', 'rename mode, valid options are renameUnique or overwrite. Default is renameUnique.', /^(renameUnique|overwrite)$/i)
    .parse(process.argv);

if (commander.log) {
    //Log status messages to file
    winston.add(winston.transports.File, { filename: commander.log });
    winston.remove(winston.transports.Console);
}

var concurrentApiCalls = commander['concurrent-api-calls'] || 5;
// 60 requests per minute
var rate = commander.rate || 60;
var limiter = new RateLimiter(rate, 'minute');

winston.info('starting PUT /assets requests at a rate of ' + rate + ' requests per minute');

var API_USERNAME = process.env.AMP_USERNAME || commander.username;
var API_PASSWORD = process.env.AMP_PASSWORD || commander.password;
var apiUrl = commander.api || process.env.AMP_API || 'https://qa-3-dam-api-ssl.adis.ws';

var api = new Amplience({ retries: 0, base: apiUrl, logLevel: 'none' });
var out = commander.output ? fs.createWriteStream(commander.output) : process.stdout;
var queue = [];
var complete = false;
var isDigestingLocked = false;

function lockDigest() {
    isDigestingLocked = true;
}

function unlockDigest() {
    isDigestingLocked = false;
    digestQueue();
}

function digestQueue() {
    if (isDigestingLocked) { return; }

    //Every line scanned and digest complete
    if (complete && queue.length == 0) {
        winston.info('complete');
        return;
    }

    if (queue.length >= concurrentApiCalls || complete) {
        //Submit
        var batch = queue.splice(0, concurrentApiCalls);
        submit(batch);
    }
}

function output(line) {
    out.write(line);
}

function submit(batch) {
    //Stop any more activity until I am done
    lockDigest();

    var remaining = [].concat(batch);
    remaining.forEach(function (x) {
        //Null means failed. Otherwise the id of the inserted item will be set
        x.insertId = null;
    });

    //all puts completed or failed
    function checkPutComplete() {
        if (remaining.length == 0) {
            //Output ids
            for (var i = 0; i < batch.length; i++) {
                var batchItem = batch[i];
                if (batchItem.insertId == null) {
                    //This asset put failed
                    winston.error('put failed', batchItem);
                    output(EOL);
                } else {
                    output(batchItem.insertId + EOL);
                }
            }
            unlockDigest();
        }
    }

    function removeItemFromRemaining(items) {
        remaining.splice(remaining.indexOf(items), 1);
        checkPutComplete();
    }

    function putItem(items) {
        //Items is an array, if the first one fails move onto the next one
        var index = 0;
        function tryNext() {
            var item = items[index];
            index++;
            if (item == null) {
                //Failed, remove me
                removeItemFromRemaining(items);
                return;
            }

            var request = {
                mode: commander.mode || 'renameUnique',
                assets: [item]
            };

            limiter.removeTokens(1, function (err, remainingRequests) {
                api.put('/assets', request).then(function (response) {
                    if (response.error) {
                        tryNext();
                    } else {
                        var result = response.obj;
                        //Overall request status
                        if (result.status != 'success' ||
                            result.content == null ||
                            result.content[0].status != 'succeeded') {
                            tryNext();
                        } else {
                            //Done!
                            items.insertId = result.content[0].id;
                            removeItemFromRemaining(items);
                        }
                    }
                }).catch(function (err) {
                    //problem! move onto the next asset
                    tryNext();
                });
            });
        }

        tryNext();
    }

    remaining.forEach(function (x) { putItem(x); });
    //invoke asset put
    //unlockDigest();

}


var rl = null;

function readInput() {
    //Read std-in line by line
    rl = readline.createInterface({
        input: commander.input ? fs.createReadStream(commander.input) : process.stdin,
        output: commander.output ? fs.createWriteStream(commander.output) : process.stdout,
        terminal: false
    });


    rl.on('line', function (line) {
        if (line == '') {
            return;
        }

        //Line received, is it valid JSON
        try {
            var parsed = JSON.parse(line);
        } catch (e) {
            winston.error('invalid json', line);
            return;
        }

        queue.push(parsed);
        digestQueue();
    });

    rl.on('close', function () {
        //No more input
        complete = true;
        digestQueue();
    });

    rl.resume();
}

function bootstrap() {
    api.login(API_USERNAME, API_PASSWORD).then(function () {
        winston.info('acquired authentication token');
        //Start working
        readInput();
    }, function (err) {
        winston.error('unable to authenticate with api');
        process.exit(1);
    });
}

bootstrap();