ric-edge-core.js

const mqtt = require('mqtt').connect('mqtt://sandbox.rightech.io', {
    clientId: 'mqtt-prohazko-fFJlHqZwnH7wYTihXOlH0',
    //key: require('fs').readFileSync('./key.pem').toString(),
    //cert: require('fs').readFileSync('./crt.pem').toString(),
});
const si = require('systeminformation');
const nanoid = require('nanoid');
const { Server } = require('ws');

const ports = {};
const reqs = {};

function rpcError(req, message, code = -32050) {
    return {
        jsonrpc: "2.0",
        id: req.id || null,
        error: { code, message }
    }
}

function rpcOk(req, result = { ok: true }) {
    return {
        jsonrpc: "2.0",
        id: req.id || null,
        result
    }
}

async function reportStats() {
    const cpu = await si.cpuTemperature();
    const mem = await si.mem();

    mqtt.publish('ric-edge/sys/info', JSON.stringify({
        temperature: cpu.main,
        memfree: mem.available / 1024 / 1024,
        memused: mem.active / 1024 / 1024
    }));
    setTimeout(reportStats, 60 * 1000);
}

mqtt.on('connect', () => reportStats());

mqtt.on('message', (topic, message) => {
    const [portName] = topic.replace('ric-edge', '').split('/').filter(x => !!x);

    let req;
    try {
        req = JSON.parse(message.toString());
    } catch (err) {
        return mqtt.publish(`ric-edge/${portName}/response`,
            JSON.stringify(rpcError({}, 'Parse error', -32700)));
    }

    const port = ports[portName];
    if (!port) {
        return mqtt.publish(`ric-edge/${portName}/response`,
            JSON.stringify(rpcError(req, 'Connector not connected')));
    }

    if (!req.id) {
        req.id = nanoid();
    }

    reqs[req.id] = req;
    port.send(JSON.stringify(req));
});


const wss = new Server({ port: 17080 });

wss.on('connection', (ws, req) => {
    const [port] = req.url.split('/').filter(x => !!x);
    if (ports[port]) {
        ports[port].close();
    }
    ports[port] = ws;

    ws.on('close', () => delete ports[port]);
    ws.on('error', () => delete ports[port]);

    ws.on('message', (message) => {
        try {
            message = JSON.parse(message.toString());
            const req = reqs[message.id];
            if (req) {
                delete reqs[message.id];
                if (message.error) {
                    return mqtt.publish(`ric-edge/${port}/response`,
                        JSON.stringify(rpcError(req, message.error.message, message.error.code)));
                }

                return mqtt.publish(`ric-edge/${port}/response`,
                    JSON.stringify(rpcOk(req, message.result)));
            }
        } catch (err) {
            console.log(err);
        }
    });
});