@smartface/library-manager/libraryManager.js

#!/usr/bin/env node

const execSync = require('child_process').execSync;
const args = require('minimist')(process.argv.slice(2));
const chokidar = require('chokidar');
const WebSocket = require('ws');
const bytes = require('bytes');
const queue = require('async/queue');
const RawIPC = require('node-ipc').IPC;
const ipc = new RawIPC();
const pingClient = new RawIPC();
const WebSocketServer = WebSocket.Server;
const library = require("./src/library");
const CONSTANTS = require("./constants");

const BRACKET_END = CONSTANTS.BRACKET_END;
const SERVER_PORT = CONSTANTS.SERVER_PORT;
const APP_NAME = CONSTANTS.SERVER_APP_NAME;

var watcherEnabled = true;
var latestWriteData = "";

pingClient.config.id = 'sfLibraryManagerPingClient';
pingClient.config.encoding = "utf8";

pingClient.config.logger = function () { };

ipc.config.id = APP_NAME;
ipc.config.retry = 5000;
ipc.config.encoding = "utf8";
ipc.config.maxConnections = 15;
ipc.config.logger = function (a) { args.verbose && console.log(a) };
let sockets = [];
let wses = [];
let subscribersForRead = [];
let watcher = null;
let _libraryFolder = CONSTANTS.LIBRARY_DIRECTORY;

process.on("uncaughtException", logError);
process.on('unhandledRejection', logError);

if (args.restart) {
    try {
        execSync(`fuser -k ${SERVER_PORT}`);
    }
    catch (ex) {
        // do nothing; process is not running; everything is OK
    }
}

function logError(reason) {
    console.error("├─── ! ☠️ ! ──────── LibraryManager ───── Unhandled Error ─────────────────────────┤\n" +
        reason.stack ? reason.stack.toString("utf8") : reason.toString("utf8"));
}

function initLibraryTask(data, cb) {
    library.createLibDirFile(_libraryFolder, data.initialLibraryData, e => {
        e && console.error(e);
        cb && cb(e);
        watch(_libraryFolder);
    });
}

function watch(libraryFolder) {
    watcher && watcher.close();
    watcher = chokidar.watch(`${libraryFolder}/*.!(pgx)cpx`, {
        ignoreInitial: true
    });
    var readCounter = 0;
    watcher.on("all", (e, filename) => {
        if (!watcherEnabled)
            return;
        ++readCounter;
        setTimeout(() => {
            --readCounter;
            if (readCounter !== 0)
                return;
            console.log("Watcher ..> ", e, " . ", filename);
            readTask((e, res) => {
                notifyBrowsers("Watcher -> ReadDone: ", e, res);
                notifySubscribers("Watcher -> ReadDone notify subscriber : ", e, res);
            });
        }, 800);
    });
}

function readTask(cb) {
    cb = typeof cb === "function" ? cb : function () { };
    library.read(_libraryFolder, function (err, comps) {
        if (err) {
            console.error(err);
            cb(err);
        }
        else {
            return cb(null, JSON.stringify(comps));
        }
    });
}

function writeTask(data, _cb) {
    const rawData = data.rawData || data;
    if (latestWriteData === rawData) {
        console.log("Library already saved!");
        return _cb(null);
    }
    var cb = err => {
        if (err)
            return _cb(err);
        latestWriteData = rawData;
        _cb();
    };
    let componentsWithChildren = data.parsedData;
    if (!data.parsedData) {
        try {
            componentsWithChildren = JSON.parse(data);
        }
        catch (e) {
            console.error(e);
            return cb(e);
        }
    }
    library.saveComponents(_libraryFolder, componentsWithChildren, err => {
        if (err) {
            console.error(err);
            return cb(err);
        }
        notifySubscribersAndwriteUninitializedComp(componentsWithChildren, cb);
    });
}

function notifySubscribersAndwriteUninitializedComp(componentsWithChildren, cb) {
    var unInitializedComps = [];
    componentsWithChildren.forEach(comps => {
        if (!comps[0].initialized) {
            console.log("Uninitialized comp: ", comps[0].props.name);
            unInitializedComps.push(comps);
        }
    });
    library.getLibraryPageWithChildren(_libraryFolder, componentsWithChildren, (err, comps) => {
        if (comps) {
            notifySubscribers("ReceivedNewData -> notify subscribers :", err, JSON.stringify(comps));
        }
        if (unInitializedComps.length) {
            unInitializedComps.forEach(comps => comps[0].initialized = true);
            notifyBrowsers("Initialized components -> ", err, JSON.stringify(comps));
            library.saveComponents(_libraryFolder, unInitializedComps, cb, true);
        }
        else {
            cb();
        }
    });
}

function notifySubscribers(title, err, res) {
    subscribersForRead.forEach(s => {
        console.log(title, s.name, " ", bytes(res ? res.length : 0, { unitSeparator: ' ' }), " send.");
        ipc.server.emit(
            s,
            err ? 'error' : 'read',
            err ? err.toString() : res
        );
    });
}

function notifyBrowsers(title, err, res) {
    sockets.forEach(s => {
        console.log(title, s.name, " ", bytes(res ? res.length : 0, { unitSeparator: ' ' }), " send.");
        ipc.server.emit(
            s,
            err ? 'error' : 'read',
            err ? err.toString() : res
        );
    });
    if (!err) {
        wses.forEach((ws) => {
            ws.send(res);
            ws.send(`\r\n${BRACKET_END}`);
        });
    }
}

var writeTaskQueue = queue((data, next) => {
    watcherEnabled = false;
    writeTask(data, next);
}, 1);

writeTaskQueue.drain = () => setTimeout(() => (watcherEnabled = true), 1000);

ipc.serveNet(
    SERVER_PORT,
    () => {
        console.log(`Smartface libraryManager (${SERVER_PORT}) is up and running...`);
        ipc.server.on(
            'writeTask',
            (data, socket) => {
                console.log("WriteTask: ", bytes(data.length, { unitSeparator: ' ' }));
                writeTaskQueue.push(data);
            }
        );

        ipc.server.on(
            'readTask',
            (data, socket) => {
                console.log("ReadTask ", data.name);
                _libraryFolder = _libraryFolder || data.libraryFolder;
                socket.name = data && data.name;
                readTask((e, res) => {
                    ipc.server.emit(
                        socket,
                        'read',
                        res
                    );
                });
            }
        );
        ipc.server.on(
            'subscribeForRead',
            (data, socket) => {
                console.log("SubscribeForRead: ", data.name);
                _libraryFolder = _libraryFolder || data.libraryFolder;
                socket.name = data && data.name;
                subscribersForRead.push(socket);
                readTask((e, res) => {
                    ipc.server.emit(
                        socket,
                        'read',
                        res
                    );
                });
            }
        );

        ipc.server.on(
            'init',
            (data, socket) => {
                console.log("Init ", data.name);
                socket.name = data && data.name;
                _libraryFolder = data.libraryFolder;
                sockets.push(socket);
                initLibraryTask(data, e => {
                    readTask((e, res) => {
                        console.log("ReadDone: ", socket.name, " ", bytes(res ? res.length : 0, { unitSeparator: ' ' }));
                        ipc.server.emit(
                            socket,
                            'read',
                            res
                        );
                    });
                });
            }
        );

        ipc.server.on(
            "ping",
            (data, socket) => {
                ipc.server.emit(socket, 'ping', "OK");
            });

        ipc.server.on(
            'socket.disconnected',
            (socket) => {
                sockets = sockets.filter(s => s !== socket);
                subscribersForRead = subscribersForRead.filter(s => s !== socket);
                socket.name && console.log('DISCONNECTED ', socket.name);
            }
        );
    }
);

ipc.server.on(
    'error',
    (err) => {
        console.log('Got an ERROR!', err);
    }
);
ipc.server.start();


function handleWsTask(parsedData, ws) {
    if (parsedData.task === 'init') {
        initLibraryTask(parsedData, () => {
            readTask((e, res) => {
                ws.send(res);
                ws.send(`\r\n${BRACKET_END}`);
            });
        })
    }
}

const wsServer = new WebSocketServer({ port: CONSTANTS.SERVER_WS_PORT });
console.log(`Smartface libraryManager (${CONSTANTS.SERVER_WS_PORT}) ws server is up and running...`);

wsServer.on('connection', (ws) => {

    wses.push(ws);
    let msgData = '';
    ws.on('message', (msg) => {
        msgData += msg;
        console.log("RECEIVED:> ", msg);
        if (msgData.endsWith(BRACKET_END)) {
            const splitted = msgData.split(BRACKET_END);
            const rawData = splitted[splitted.length - 2];
            const parsedData = JSON.parse(rawData);
            if (parsedData.task) {
                handleWsTask(parsedData, ws);
            } else {
                writeTaskQueue.push({ rawData, parsedData });
            }
            msgData = '';
        }
    });

    ws.on('close', () => {
        wses = wses.filter(w => w !== ws);
        console.info('ws on Close');
    });

});

var counter = 0;
var aliveCounter = 0;
setInterval(_ => {
    pingClient.connectToNet(
        APP_NAME,
        CONSTANTS.SERVER_PORT,
        function () {
            pingClient.of.sfLibraryManager.on(
                'connect',
                _ => {
                    ((++aliveCounter) % 15 == 0)
                    counter = 0;
                    pingClient.disconnect(APP_NAME);
                }
            );
            pingClient.of.sfLibraryManager.on(
                'error',
                (data) => {
                    if (++counter < 3)
                        return;
                    process.stderr.write(data.toString());
                    process.exit(1);
                }
            );
        }
    );
}, 5000);