feature(distribute) add ability to import config from remote export server

server/cloudcmd.js

@@ -17,6 +17,7 @@ const validate = require(DIR + 'validate');
 const prefixer = require(DIR + 'prefixer');
 const pluginer = require(DIR + 'plugins');
 const terminal = require(DIR + 'terminal');
+const distribute = require(DIR + 'distribute');
 
 const currify = require('currify/legacy');
 const apart = require('apart');
@@ -147,6 +148,8 @@ function listen(prefix, socket) {
         auth,
         prefix: prefix + '/gritty',
     });
+    
+    distribute.export(socket);
 }
 
 function cloudcmd(prefix, plugins, modules) {

server/config.js

@@ -6,6 +6,7 @@ const DIR = DIR_SERVER + '../';
 
 const path = require('path');
 const fs = require('fs');
+const Emitter = require('events');
 
 const exit = require(DIR_SERVER + 'exit');
 const CloudFunc = require(DIR_COMMON + 'cloudfunc');
@@ -35,6 +36,7 @@ const send = swap(ponse.send);
 const formatMsg = currify((a, b) => CloudFunc.formatMsg(a, b));
 
 const apiURL = CloudFunc.apiURL;
+const changeEmitter = new Emitter();
 
 const ConfigPath = path.join(DIR, 'json/config.json');
 const ConfigHome = path.join(HOME, '.cloudcmd.json');
@@ -59,8 +61,16 @@ const config = Object.assign({}, rootConfig, configHome);
 const connectionWraped = wraptile(connection);
 
 module.exports          = manage;
-module.exports.save     = _save;
+module.exports.save     = save;
 module.exports.middle   = middle;
+module.exports.subscribe = (fn) => {
+    changeEmitter.on('change', fn);
+};
+
+module.exports.unsubscribe = (fn) => {
+    changeEmitter.removeListener('change', fn);
+};
+
 module.exports.listen   = (socket, auth) => {
     check(socket, auth);
     
@@ -83,6 +93,10 @@ function manage(key, value) {
         return config[key];
     
     config[key] = value;
+    
+    changeEmitter.emit('change', key, value);
+    
+    return `${key} = ${value}`;
 }
 
 function _save(callback) {

server/distribute/export.js

@@ -0,0 +1,114 @@
+'use strict';
+
+const currify = require('currify/legacy');
+const wraptile = require('wraptile/legacy');
+const squad = require('squad/legacy');
+const omit = require('object.omit');
+
+const config = require('../config');
+const log = require('./log');
+
+const exportStr = log.exportStr;
+const connectedStr = log.connectedStr;
+const disconnectedStr = log.disconnectedStr;
+const authTryStr = log.authTryStr;
+
+const makeColor = log.makeColor;
+const getMessage = log.getMessage;
+const getDescription = log.getDescription;
+const logWraped = log.logWraped;
+
+const omitList = [
+    'auth',
+    'username',
+    'password',
+    'algo',
+    'name',
+    'ip',
+    'port',
+    'root',
+    'import',
+    'importUrl',
+    'importToken',
+    'export',
+    'exportToken',
+    'log',
+    'configDialog',
+];
+
+const omitConfig = wraptile((config) => omit(config, omitList));
+
+module.exports = (socket) => {
+    if (!config('export'))
+        return;
+    
+    const prefix = config('prefix');
+    const distributePrefix = `${prefix}/distribute`;
+    
+    const onError = squad(logWraped(exportStr), getMessage);
+    const onConnectError = squad(logWraped(exportStr), getDescription);
+    
+    socket.of(distributePrefix)
+        .on('connection', onConnection(push))
+        .on('error', onError)
+        .on('connect_error', onConnectError);
+};
+
+const push = currify((socket, key, value) => {
+    if (omitList.includes(key))
+        return;
+    
+    socket.emit('change', key, value);
+});
+
+function getHost(socket) {
+    const remoteAddress = socket.request.connection.remoteAddress;
+    const name = socket.handshake.query.name;
+    const port = socket.handshake.query.port;
+    const color = socket.handshake.query.color;
+    
+    if (!name)
+        return `${remoteAddress}:${port}`;
+    
+    const colorName = makeColor(name, color);
+    
+    return `${colorName} [${remoteAddress}:${port}]`;
+}
+
+const connectPush = wraptile((push, socket) => {
+    socket.emit('accept');
+    
+    const host = getHost(socket);
+    const subscription = push(socket);
+    
+    socket.on('disconnect', onDisconnect(subscription, host));
+    
+    log(exportStr, `${connectedStr} to ${host}`);
+    socket.emit('config', omitConfig(config('*')));
+    
+    config.subscribe(subscription);
+});
+
+const onConnection = currify((push, socket) => {
+    const host = getHost(socket);
+    const reject = () => {
+        socket.emit('reject');
+        socket.disconnect();
+    };
+    
+    log(exportStr, `${authTryStr} from ${host}`);
+    socket.on('auth', auth(connectPush(push, socket), reject));
+});
+
+const auth = currify((fn, reject, token) => {
+    if (token === config('exportToken'))
+        return fn();
+    
+    reject();
+});
+
+const onDisconnect = wraptile((subscription, host) => {
+    config.unsubscribe(subscription);
+    log(exportStr, `${disconnectedStr} from ${host}`);
+});
+

server/distribute/export.spec.js

@@ -0,0 +1,45 @@
+'use strict';
+
+const {promisify} = require('util');
+
+const test = require('tape');
+const io = require('socket.io-client');
+
+const {connect} = require('../../test/before');
+const config = require('../config');
+
+test('distribute: export', async (t) => {
+    const defaultConfig = {
+        export: true,
+        exportToken: 'a',
+        vim: true,
+        log: false,
+    };
+    
+    const {port, done} = await connect({
+        config: defaultConfig
+    });
+    
+    const url = `http://localhost:${port}/distribute?port=${1111}`;
+    const socket = io.connect(url);
+    
+    const name = config('name');
+    
+    socket.on('connect', () => {
+        socket.emit('auth', 'a');
+    });
+    
+    socket.on('accept', () => {
+        config('vim', false);
+        config('auth', true);
+    });
+    
+    socket.on('change', async () => {
+        socket.close();
+        await done();
+        
+        t.pass('should emit change');
+        t.end();
+    });
+});
+

server/distribute/import.js

@@ -0,0 +1,131 @@
+'use strict';
+
+const currify = require('currify/legacy');
+const wraptile = require('wraptile/legacy');
+const squad = require('squad/legacy');
+const fullstore = require('fullstore/legacy');
+
+const io = require('socket.io-client');
+const forEachKey = currify(require('for-each-key/legacy'));
+
+const config = require('../config');
+const log = require('./log');
+
+const importStr = log.importStr;
+const connectedStr = log.connectedStr;
+const disconnectedStr = log.disconnectedStr;
+const tokenRejectedStr = log.tokenRejectedStr;
+const authTryStr = log.authTryStr;
+
+const makeColor = log.makeColor;
+const stringToRGB = log.stringToRGB;
+const getMessage = log.getMessage;
+const getDescription = log.getDescription;
+const logWraped = log.logWraped;
+
+const equal = (a, b) => `${a}=${b}`;
+const append = currify((obj, a, b) => obj.value += b && equal(a, b) + '&');
+const wrapApply = (f, disconnect) => (status) => () => f(null, {
+    status,
+    disconnect,
+});
+
+const closeIfNot = wraptile((socket, is) => !is && socket.close());
+const addUrl = currify((url, a) => `${url}: ${a}`);
+
+const getColorUrl = (url, name) => {
+    if (!name)
+        return url;
+    
+    return makeColor(url, stringToRGB(name));
+};
+
+const rmListeners = wraptile((socket, listeners) => {
+    socket.removeListener('connect', listeners.onConnect);
+    socket.removeListener('config', listeners.onConfig);
+    socket.removeListener('error', listeners.onError);
+    socket.removeListener('connection_error', listeners.onError);
+});
+
+const canceled = (f) => f(null, {
+    status: 'canceled',
+    disconnect: () => {},
+});
+
+const done = wraptile((fn, store) => fn(null, {
+    status: store()
+}));
+
+const emitAuth = wraptile((importUrl, socket) => {
+    log(importStr, `${authTryStr} to ${importUrl}`);
+    socket.emit('auth', config('importToken'));
+});
+
+module.exports = (options, fn) => {
+    fn = fn || options;
+    
+    if (!config('import'))
+        return canceled(fn);
+    
+    const importUrl = config('importUrl');
+    const importListen = config('importListen');
+    const name = config('name');
+    const port = config('port');
+    
+    const query = toLine({
+        name,
+        port,
+    });
+    
+    const url = `${importUrl}/distribute?${query}`;
+    const socket = io.connect(url, Object.assign({}, {
+        rejectUnauthorized: false,
+    }, options));
+    
+    const superFn = wrapApply(fn, socket.close.bind(socket));
+    const colorUrl = getColorUrl(importUrl, name);
+    const close = closeIfNot(socket, importListen);
+    
+    const statusStore = fullstore();
+    const statusStoreWraped = wraptile(statusStore);
+    
+    const onConfig = squad(close, logWraped(importStr, `config received from ${colorUrl}`), statusStoreWraped('received'), forEachKey(config));
+    const onError = squad(superFn('error'), logWraped(importStr), addUrl(colorUrl), getMessage);
+    const onConnectError = squad(superFn('connect_error'), logWraped(importStr), addUrl(colorUrl), getDescription);
+    const onConnect = emitAuth(importUrl, socket);
+    const onAccept = logWraped(importStr,`${connectedStr} to ${colorUrl}`);
+    const onDisconnect = squad(done(fn, statusStore), logWraped(importStr, `${disconnectedStr} from ${colorUrl}`), rmListeners(socket, {
+        onError,
+        onConnect,
+        onConfig,
+    }));
+    
+    const onChange = squad(logWraped(importStr), config);
+    const onReject = squad(superFn('reject'), logWraped(importStr, tokenRejectedStr));
+    
+    socket.on('connect', onConnect);
+    socket.on('accept', onAccept);
+    socket.on('disconnect', onDisconnect);
+    socket.on('config', onConfig);
+    socket.on('error', onError);
+    socket.on('connect_error', onConnectError);
+    socket.on('reject', onReject);
+    
+    if (config('importListen'))
+        socket.on('change', onChange);
+};
+
+function toLine(obj) {
+    const result = {
+        value: '',
+    };
+    
+    forEachKey(append(result), obj);
+    
+    const start = 0;
+    const end = 1;
+    const backward = -1;
+    
+    return result.value.slice(start, backward * end);
+}
+

server/distribute/import.spec.js

@@ -0,0 +1,183 @@
+'use strict';
+
+const test = require('tape');
+const {promisify} = require('util');
+const tryToCatch = require('try-to-catch');
+const io = require('socket.io-client');
+const mockRequire = require('mock-require');
+
+const {connect} = require('../../test/before');
+
+const config = require('../config');
+const distribute = {
+    import: promisify(require('./import')),
+};
+
+test('distribute: import: canceled', async (t) => {
+    const {done, port} = await connect({
+        config: {
+            export: false,
+            import: false,
+            importListen: false,
+            log: false,
+        }
+    });
+    
+    const {status} = await distribute.import();
+    
+    await done();
+    
+    t.equal(status, 'canceled', 'should equal');
+    t.end();
+});
+
+test('distribute: import: received: no error', async (t) => {
+    const {done, port} = await connect({
+        config: {
+            import: true,
+            importListen: false,
+            export: true,
+            log: false,
+        }
+    });
+    
+    config('importUrl', `http://localhost:${port}`);
+    
+    const [e] = await tryToCatch(distribute.import);
+    
+    await done();
+    
+    t.notOk(e, 'should not be error');
+    t.end();
+});
+
+test('distribute: import: received', async (t) => {
+    const {done, port} = await connect({
+        config: {
+            name: 'bill',
+            import: true,
+            importToken: 'a',
+            exportToken: 'a',
+            export: true,
+            importListen: false,
+            log: false,
+        }
+    });
+    
+    config('importUrl', `http://localhost:${port}`);
+    
+    const {status} = await distribute.import();
+    await done();
+    
+    t.equal(status, 'received','should equal');
+    t.end();
+});
+
+test('distribute: import: received: auth: reject', async (t) => {
+    const {done, port} = await connect({
+        config: {
+            name: 'bill',
+            import: true,
+            importToken: 'xxxxx',
+            exportToken: 'bbbbb',
+            export: true,
+            importListen: false,
+            log: false,
+        }
+    });
+    
+    config('importUrl', `http://localhost:${port}`);
+    
+    const {status} = await distribute.import();
+    await done();
+    
+    t.equal(status, 'reject','should equal');
+    t.end();
+});
+
+test('distribute: import: received: auth: accept', async (t) => {
+    const {done, port} = await connect({
+        config: {
+            name: 'bill',
+            import: true,
+            importToken: 'xxxxx',
+            exportToken: 'xxxxx',
+            export: true,
+            importListen: false,
+            log: false,
+        }
+    });
+    
+    config('importUrl', `http://localhost:${port}`);
+    
+    const {status} = await distribute.import();
+    await done();
+    
+    t.equal(status, 'received','should equal');
+    t.end();
+});
+
+test('distribute: import: received: no name', async (t) => {
+    const {done, port} = await connect({
+        config: {
+            name: '',
+            import: true,
+            export: true,
+            importListen: false,
+            log: false,
+        }
+    });
+    
+    config('importUrl', `http://localhost:${port}`);
+    
+    const {status} = await distribute.import();
+    await done();
+    
+    t.equal(status, 'received','should equal');
+    t.end();
+});
+
+test('distribute: import: error', async (t) => {
+    const {done, port} = await connect({
+        config: {
+            import: true,
+            export: false,
+            importListen: false,
+            log: false,
+        }
+    });
+    
+    config('importUrl', `http://localhost:0`);
+    
+    const {status} = await distribute.import({
+        reconnection: false,
+    });
+    
+    await done();
+    
+    t.equal(status, 'connect_error','should equal');
+    t.end();
+});
+
+test('distribute: import: config:change: no export', async (t) => {
+    const {done, port} = await connect({
+        config: {
+            import: true,
+            export: false,
+            importListen: true,
+            log: false,
+        }
+    });
+    
+    const {status} = await distribute.import({
+        reconnection: false,
+    });
+    
+    await done();
+    
+    t.equal(status, 'connect_error','should equal');
+    t.end();
+});
+
+process.on('unhandledRejection', console.log);
+

server/distribute/index.js

@@ -0,0 +1,4 @@
+'use strict';
+
+module.exports.import = require('./import');
+module.exports.export = require('./export');

server/distribute/log.js

@@ -0,0 +1,45 @@
+'use strict';
+
+const wraptile = require('wraptile/legacy');
+const chalk = require('chalk');
+
+const config = require('../config');
+const datetime = require('../../common/datetime');
+
+const log = (name, msg) => config('log') && console.log(`${datetime()} -> ${name}: ${msg}`);
+const makeColor = (a, color) => chalk.rgb(color || stringToRGB(a))(a);
+const getMessage = (e) => e.message || e;
+const getDescription = (e) => `${e.type}: ${e.description}`;
+
+module.exports = log;
+module.exports.logWraped = wraptile(log);
+module.exports.stringToRGB = stringToRGB;
+module.exports.makeColor = makeColor;
+module.exports.getMessage = getMessage;
+module.exports.getDescription = getDescription;
+
+module.exports.importStr = 'import';
+module.exports.exportStr = 'export';
+module.exports.connectedStr = chalk.green('connected');
+module.exports.disconnectedStr = chalk.red('disconnected');
+module.exports.tokenRejectedStr = chalk.red('token rejected');
+module.exports.authTryStr = chalk.yellow('try to auth');
+
+function stringToRGB(a) {
+    return [
+        a.charCodeAt(0),
+        a.length,
+        crc(a),
+    ];
+}
+
+const add = (a, b) => {
+    return a + b.charCodeAt(0);
+};
+
+function crc(a) {
+    return a
+        .split('')
+        .reduce(add, 0);
+}
+

server/distribute/log.spec.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const test = require('tape');
+const log = require('./log');
+const config = require('../config');
+
+test('distribute: log: getMessage', (t) => {
+    const e = 'hello';
+    const result = log.getMessage(e)
+    
+    t.equal(e, result, 'should equal');
+    t.end();
+});
+
+test('distribute: log: getMessage: message', (t) => {
+    const message = 'hello';
+    const result = log.getMessage({
+        message
+    })
+    
+    t.equal(result, message, 'should equal');
+    t.end();
+});
+
+test('distribute: log: config', (t) => {
+    const logOriginal = config('log');
+    config('log', true);
+    log('log', 'test message');
+    config('log', logOriginal);
+    
+    t.end();
+});