highland_example_1.js

//Earlier
var H = require('highland');

//Later

/**
 * Wrapper for fs.watch
 * @param path
 * @returns {*}
 */

function fsWatchAsync(path) {
    return H(function (push, next) {
        fs.watch(path, function (event, filename) {
            push(null, arguments);
            push(null, H.nil);
        });
    });
}

/**
 * Wrapper for fs.exists
 * @param path
 * @returns {*}
 */

function fsExists(path) {
    return H(function (push, next) {
        fs.exists(path, function (exists) {
            push(null, exists);
            push(null, H.nil);
        });
    });
}

function fsReadFile(file) {
    var fsRead = H.wrapCallback(fs.readFile);
    return fsRead(file);
}
function fsRename(from, to) {
    var _fsRename = H.wrapCallback(fs.rename);
    return _fsRename(from, to);
}

/**
 * Watch migrations folder for migration.json, process directives, and rename to {date}.json
 * @param children
 */
function migrationWatcher(children) {

    var migration_folder = path.join(__dirname, '/migrations/');
    var migration_filename = 'migration.json';
    var migration_path = path.join(migration_folder, migration_filename)

    fsWatchAsync(migration_folder) //Watch for folder events
        //{ '0': 'rename', '1': 'migration.json' }
        .map(function (data) { //Map file names to array
            return data[1]; //filename
        })
        // [migration.json]
        .filter(function (filename) { //Filter values for the migration file
            return filename === migration_filename;
        })
        // [migration.json]
        .map(function (file) { //Map migration path onto the stream array
            return migration_path;
        })
        // [FULL_PATH/migrations/migration.json]
        .flatFilter(fsExists) //Check if file exists
        // [] or if path exists [FULL_PATH/migrations/migration.json]
        .flatMap(fsReadFile) //Read the file
        // <Buffer ....
        .flatMap(function (file) {
            var migration = JSON.parse(file);
            return processMigration(children, migration); //Processes migration JSON and returns processes changed
        })
        //[ 'server1', 'server2' ]
        .collect()//Condense stream to a single element ['server1', 'server2'] to [ ['server1', 'server2'] ]
        .flatMap(function (item) { //Map single event to rename the migrations file
            var completed_migration = path.join(__dirname, '/migrations/', (new Date()).toJSON() + '.json'); //Rename migration.js to {date}.json
            return fsRename(migration_path, completed_migration); //Rename
        })
        // [ undefined ]
        .errors(function (err) { //Process any and all errors on the stream without tanking the whole process
            console.error(err);
            console.error(err.stack);
        })
        .toArray(function () {
            console.log('Finished Migrations'); //TODO Meaningful logging message later
        });
}