Stream Implementation In JavaScript

Question

I usually do not create Streams, but it seems that JavaScript doesn't have a built-in one and I figured I would do it for a personal project I am working on. This code seems to work as intended and does what I want just fine, but I have concerns mainly about how often I am checking my buffer. I am mainly looking for a recommendation about how often I should check to see if any data has been written to my buffer. If there are any other improvements/suggestions, I am open to hearing them!

Edit: After thinking about this more, I realized I could do this as event-driven instead and let the system/browser handle the blocking. I left both functions in so people can weigh in on either one. However, I would still like input on the first approach for general knowledge.

const StreamPrototype = {
    _data: [],
    _callbacks: [],

    readAsync: async function* () {
        while(await sleep(25)) {
            let data;
            while((data = this._data.pop()))
                yield data;
        }
    },

    onData: function(callback) {
        if (callback instanceof Function)
        this._callbacks.push(callback);
    },

    push: function(data) {
        if (data) {
            this._pushOntoStack(data);
        }
    },

    _pushOntoStack: async function(data) {
        return new Promise(resolve => {
            this._data.push(data); // Uses Sleeps
            this._callbacks.forEach(item => { // Pushes data out using event-driven
                item(data);
            });
        });
    }
};

// This was just copied from the internet
const sleep = time => new Promise(res => setTimeout(res, time, "done sleeping"));

Answer 1

This doesn't look like a nice approach. For one, you don't really need that 25ms waiting like polling.

So assuming that you're working with WebSockets like

// Create WebSocket connection.
const socket = new WebSocket('ws://localhost:8080');

// Listen for messages
socket.addEventListener('message', (event) => {
    console.log('Message from server ', event.data);
});

You may merge the async "message" event listener and others like "error", "open", "close" etc, under one async iterable and consume with for await of loop as follows. Note that we don't need to use an async generator here since we are yielding promises.

function wsMessageStream(url){
  let _v, // previous resolve
      _x, // previous reject
      p  = new Promise((v,x) => (_v = v, _x = x)),
      ws = new WebSocket(url);

  function* emitterGen(ws){
    ws.addEventListener( "open"
                       , _ => ( _v({data: `Esatablished WebSocket connection for ${url}.`})
                              , p = new Promise((v,x) => (_v = v, _x = x))
                              )
                       );
    ws.addEventListener( "close"
                       , _ => ( _v({data: `Closed WebSocket connection for ${url}.`})
                              , _v = null
                              , _x = null
                              )
                       );
    ws.addEventListener( "message"
                       , m => ( _v(m)
                              , p = new Promise((v,x) => (_v = v, _x = x))
                              )
                       );
    ws.addEventListener( "error"
                       , e => ( _x(e)
                              , p = new Promise((v,x) => (_v = v, _x = x))
                              )
                       );
    while (_v || _x){
      try{
        yield p;
      }
      catch(e){
        console.log(e);
      }
    }
    return Promise.resolve(null); // finalize the iterable
  }
  return { stream: emitterGen(ws)
         , socket: ws
         };
}

let {stream,socket} = wsMessageStream('wss://ws.postman-echo.com/raw'),
    counter = 0;
setTimeout(_ => socket.close(), 5000);

(async function(){

  for await (let msg of stream){
    console.log(`Message received: ${msg.data}`);
    // run msg handlers here
    socket.readyState === 1 && socket.send(`Thank you. ${counter++}`);
  }
})();