AsyncWS/index.js

const express = require('express');
const http    = require( 'http');
const WebSocket  = require('ws');

const app = express();

//initialize a simple http server
const server = http.createServer(app);

//initialize the WebSocket server instance
const wss = new WebSocket.Server({ server });

const messages = [];
const sockets  = [];
let   userCount = 0;






function noPromise(){
    let resolve, reject;

    let np = new Promise((ok, fail) => {resolve = ok; reject = fail});

    np.resolve = resolve;
    np.reject  = reject;

    return np
}

function asynchronize({s, chunkEventName, endEventName}){
    return function* wrapper(){
        const chunks        = {};
        const promises      = {};

        const clear = i => (delete chunks[i], delete promises[i])


        let   chunkCount    = 0;
        let   promiseCount  = 0;
        let   end           = false;

        s.on(chunkEventName, data => {
            chunks[chunkCount] = data

            if (chunkCount in promises){
                promises[chunkCount].resolve(chunks[chunkCount])
                clear(chunkCount)
            }
            chunkCount++
        })

        s.on(endEventName, () => {
            end = true;

            for (let i in promises){
                promises[i].reject(new Error('End Of S'))
            }
        })

        while (!end){
            let p = noPromise()

            if (promiseCount in chunks){
                p.resolve(chunks[promiseCount])
                clear(promiseCount)
            }

            promises[promiseCount] = p;
            promiseCount++

            yield p;
        }
    }
}






function broadcast(wtf){
    for (s of sockets){
        console.log(wtf)
        s.send(JSON.stringify(wtf))
    }
}


let RPC = {
    addMessage({nick, message}){
        messages.push({nick, message, timestamp: (new Date).getTime()})
        broadcast({func: 'addMessage', nick, message})
        return messages.length;
    },

    getMessages({offset=0}){
        return messages.slice(offset)
    },

    getUserCount(){
        return userCount;
    }
}





wss.on('connection', async ws => {

    let gena = asynchronize({s:ws, chunkEventName: 'message', endEventName: 'close'})

    sockets.push(ws)

    userCount++;
    for (let p of gena()){
        try {
            let message = await p;
            let data    = JSON.parse(message)

            if (data.func in RPC){
                ws.send(JSON.stringify(RPC[data.func](data)))
            }
        }
        catch(e){
            if (e.message === 'End Of S'){
                console.log('client OTPAL')
                break;
            }
            console.log(e)
        }
    }

    sockets = sockets.filter(s => s !== ws)

    userCount--;
});

app.use(express.static('public'));

//start our server
server.listen(process.env.PORT || 8999, () => {
    console.log(`Server started on port ${server.address().port} :)`);
});