How do I queue incoming websocket events in javascript for slow execution?

user_5131394 0
Viewed 83

I have an open Websocket connection and it's handing out events. All good, but once a new event arrives, I need to do a whole lot of things and sometimes events arrive so quickly one after the other that there is no time to get the stuff done properly. I need some sort of queue inside this function that tells the events to take it easy and only keep going at most one per second, and otherwise wait in some sort of queue until the second elapses to go ahead and continue.

edit: No external libraries allowed, unfortunately.

ws = new WebSocket(`wss://hallo.com/ws/`);
ws.onmessage = readMessage;

async function readMessage(event) {
  print(event) 
  //do important things
  //but not too frequently!
}

How do I do that?

I found this but it goes over my simple head:

"You can have a queue-like promise that keeps on accumulating promises to make sure they run sequentially:

let cur = Promise.resolve();

function enqueue(f) {
    cur = cur.then(f); }

function someAsyncWork() {
    return new Promise(resolve => {
      setTimeout(() => {
        resolve('async work done');
      }, 5);
    }); }    async function msg() {
    const msg = await someAsyncWork();
    console.log(msg); }

const main = async() => {

    web3.eth.subscribe('pendingTransactions').on("data", function(tx) {
        enqueue(async function() {
            console.log('1st print: ',tx);
            await msg();
            console.log('2nd print: ',tx);
        });
    }) }

main();

"

async-await throttling queue websocket javascript
2 Answers

I'd honestly use something like lodash's throttle to do this. The following snippet should solve your problem.

ws = new WebSocket(`wss://hallo.com/ws/`);
ws.onmessage = _.throttle(readMessage, 1000);

async function readMessage(event) {
  print(event) 
  //do important things
  //but not too frequently!
}
user_145143680

For achieving queuing, you can make use of "settimeout" in simple/core javascript.

Whenever you receive a message from websocket, put the message processing function in a settimeout, this will ensure that the message is processed not immediately as its received, but with a delay, hence in a way you can achieve queuing.

The problem with this is that it does not guarantee that the processing of messages is sequential as they are received if that is needed.

By default settimeout in javascript does give the guarantee of when the function inside will be triggered after the time given is elapsed. Also it may not reduce the load on your message processor service for a high volume situation and since individual messages are queued two/more functions can become ready to be processed from setimeout within some time frame.

An ideal way to do so would be to create a queue. On a high level code flow this can be achieved as follows

    var queue = [];
    function getFromQueue() {  
        return queue.shift();
    }
    function insertQueue(msg) {  //called whenever a new message arrives
        queue.push(msg);
        console.log("Queue state", queue);
    }
    
    // can be used if one does not want to wait for previous message processing to finish
    // (function executorService(){
    //     setTimeout(async () => {
    //         const data = getFromQueue();
    //         await processData(data);
    //         executorService();
    //     }, 1000)
    // })()
    
    (function executorService(){
        return new Promise((res, rej) => {
            setTimeout(async () => {
                const data = getFromQueue();
                console.log("Started processing", data)
                const resp = await processData(data); //waiting for async processing of message to finish
                res(resp);
            }, 2000)
        }).then((data) =>{
            console.log("Successfully processed event", data)
        }).catch((err) => {
            console.log(err)
        }).finally(() => {
            executorService();
        })
    })()

   // to simulate async processing of messages
    function processData(data){
        return new Promise((res, rej) => {
            setTimeout(async () => {
                console.log("Finished processing", data)
                res(data);
            }, 4000)
        })
    }

    // to simulate message received by web socket
    var i = 0;
    var insertRand = setInterval(function(){
        insertQueue(i);  // this must be called on when web socket message received
        i+=1;
    }, 1000)
user_50121750
Related