1
\$\begingroup\$

I have a nodejs server with websocket which is pulling the new data from InfluxDB and send it to client.

The piece of code is this:

const ws = require("ws");
const { QueryAPI, Bucket } = require('../db/dao');

//insert data from rowObject to intoData
//return the time of current series
function insertData(rowObject, intoData){

    const blockchain = rowObject["_measurement"];
    const pool = rowObject["pool"];
    const time = rowObject["_time"];
    const priceType = rowObject["_field"];
    const priceValue = rowObject["_value"];

    //init the path
    //blockchain
    if(!intoData[blockchain]){ intoData[blockchain] = {}; }
    //blockchain -> pool
    if(!intoData[blockchain][pool]){ intoData[blockchain][pool] = {};}
    //blockchain -> pool -> time
    if(!intoData[blockchain][pool][time]){ intoData[blockchain][pool][time] = {}; }
    //blockchain -> pool -> time -> priceType(sell/buy) = priceValue;
    intoData[blockchain][pool][time][priceType] = priceValue;

    return time;

}

//create the ws router
const wsRouter = (server) => {
    
    const wss = new ws.Server({ noServer: true });

    //on connection
    wss.on("connection", (ws, req) => {

        //init
        const ipClient = req.headers['x-forwarded-for']
        console.log("WebSocket - connection: " + ipClient);
        ws.send(JSON.stringify({type: "connection", message: "online"}));

        //ping-pong
        const ping = setInterval(() => {
            if(ws.readyState === 1) { //1 - OPEN
                ws.ping();
            }
        }, 60000)
        ws.on('pong', () => {
            //console.log("WebSocket - pong: " + ipClient);
        });

        //send new prices to client
        //freq 1000ms
        let data;
        let lastTime = new Date();
        lastTime.setMinutes(lastTime.getMinutes() - 5); //get now() - 5 minutes; for initial pull of data
        lastTime = lastTime.toISOString();
        const newPrices = setInterval(() => {

            data = {};
            QueryAPI.queryRows(
                `from(bucket: "` + Bucket + `")
                |> range(start: ` + lastTime + `)
                |> filter(fn: (r) => r._time > ` + lastTime + `)`,
                {
                    next: (row, tableMeta) => {

                        //serialize data
                        const rowObject = tableMeta.toObject(row)

                        //insert data
                        const time = insertData(rowObject, data);
                        if(time > lastTime){
                            lastTime = time;
                        }

                    },
                    error: (error) => {
                        console.error("Error", error)
                        ws.send(JSON.stringify({type: "error", message: "error2"}));
                    },
                    complete: () => {
                        console.log("Success")
                        ws.send(JSON.stringify({type: "data", data: data, lastTime: lastTime}));
                    },
                }
            );

        }, 1000)

        //close
        ws.on("close", () => {
            clearInterval(newPrices);
            clearInterval(ping);
            console.log("WebSocket - close: " + ipClient);
        });
        
    });

    //Upgrade HTTP server to WebSocket
    server.on("upgrade", (req, socket, head) => {
        wss.handleUpgrade(req, socket, head, (ws) => {
            wss.emit("connection", ws, req);
        });
    });

};

module.exports = wsRouter;
  • At connection, is creating an interval at 1s, which is pulling newer data than lastTime
  • Initially the lastTime is equal with now() - 5 minutes, which makes to initially pulls, from DB, all data in the last 5 minutes.
  • After this, lastTime will be equal with the newer time from all rows.
  • During the pulling is using insertData function to put them in another object and also saving the newer time in lastTime.
  • When everything is pulled from DB, the data is sent to client.

The code works OK, but I think I have problems with performances.

Every connection to websocket, is making the process(server) of influxdb to increase with 15% CPU. I am connecting with ~15 devices.

As server I am using from DigitalOcean:

Shared CPU
Premium AMD (2 CPUs)
RAM: 2 GB
Disk: NVMe SSD

I am wondering if I can optimize the code to use less CPU per connection, and to pull data to client, with same frequency. Or the only option is to pick a better server.

\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.