feat(browser): websockets improvements and bundle optimizations

[robertsLando]

• Create an util class BufferedDuplex to replace duplexify on ws ali and wx
• Partially fixes WebSocket connection errors not catched #876 (will keep this for another PR)
• Fix bug on ws _writev function
• Fix bug on socket write, seems that when optionbrowserBufferSize is reached (defaults to 512kB) this not only wasn't blocking the writes to socket but was also causing duplicated messages. For sure there were some open issues caused by this
• Improve types in ws. Use window.WebSocket for Browser and ws module types for Node
• Significatly reduces bundle size by ~24% (90kB)
• Improved browser tests to subscribe to different topics to prevent false positives when running concurrently

Fixes #1647

[robertsLando]

@mcollina We actually use duplexify in ws wsx and ali protocols. If I understood it correctly, on browsers we do this:

1. Create the websocket const socket = createBrowserWebSocket(client, opts). This uses browser WebSocket to init a Socket with our broker
2. Create a proxy const proxy = buildProxy(opts, socketWriteBrowser, socketEndBrowser). proxy is a Transform stream used to ensure data written to socket is a Buffer and not a string (when opts.objectMode !== true).
3. If the socket is open we return the proxy as stream, if not we create a Duplex using duplexify and when the socket is open we set proxy as the duplex readable and writeable. The only reason I found to do this is that duplexify, when no readable and writeable are set, buffers all writes and flush them when setWritable is called.

If all my suppositions are correct I would like to know how in your opinion I could use readable-stream Duplex to replace duplexify. In my idea I could simply create a util class BufferedDuplex that extends Duplex that does the same, see here

[codecov]

Codecov Report

Attention: 43 lines in your changes are missing coverage. Please review.

Comparison is base (24b39a5) 81.08% compared to head (cb2bffa) 79.52%.

❗ Current head cb2bffa differs from pull request most recent head 99ea4a7. Consider uploading reports for the commit 99ea4a7 to get more accurate results

Files	Patch %	Lines
src/lib/BufferedDuplex.ts	3.22%	30 Missing ⚠️
src/lib/connect/ws.ts	23.52%	13 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1732      +/-   ##
==========================================
- Coverage   81.08%   79.52%   -1.56%     
==========================================
  Files          22       23       +1     
  Lines        1369     1397      +28     
  Branches      323      326       +3     
==========================================
+ Hits         1110     1111       +1     
- Misses        182      209      +27     
  Partials       77       77              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[robertsLando]

I also noticed this should have been chunks[i].chunk. Wondering how this cold ever be working before. Corrected here

[robertsLando]

cc @vishnureddy17 if you also would like to give a look at this your feedback would be welcome :)

[robertsLando]

Wondering how this could have work before without the bind, maybe Transform automatically handles the bind the _writev is set?

[vishnureddy17]

I like the changes. Unfortunately, my knowledge on Streams in JS is poor, so I don't think my review will be of much use.

I share your concerns regarding:

• lack of bind on writev before
• the chunks[i].chunk issue
• return after the setTimeout in socketWriteBrowser

[robertsLando]

Wondering if there should be a return after setTimeout here ?

UPDATE

Seems it should: https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/bufferedAmount

Without this send would be called anyway and even worse this would cause duplicated messages sent to socket 😨

[robertsLando]

Seems that the reason why socket error in browser are not triggering client error is that the error emitted doesn't contain a code that is recognized as a socket error code, in fact it is a generic window Event, seems that we could get a code in close event but that will always be 1006 (see reference. The fix could be to emit the error with a static code, maybe ECONREFUSED so that will trigger the error on client and close the connection, this will remove the need to call destroy on stream.

UPDATE

I decided to give up on this for now to keep it for another PR as it creates too much confusion here. I only pass the error to stream but that error will not cause client to emit it as it will not be recognized as a socket error. We could also decide this is enough

[getlarge]

As i have no knowledge of the MQTTjs codebase, i only reviewed BufferedDuplex and its usage in ws.ts.
I'll keep on reading ...but here is my firs thought.

[getlarge]

I am a bit concerned by this asynchronous _write method.
Originally Stream classes implement synchronous _write methods, this kind of break the expected signature and requires to wait for BufferedDuplex._write to resolve (or reject).

Maybe buffering the chunks to be written until the socket is ready would be more appropriate ? with something like :

Also it could handles buffering in a similar way as you wrote in ws.ts inside socketWriteBrowser ?

class BufferedDuplex {
   // ....
    private writeQueue: Array<{chunk: any, encoding: string, cb: (err?: Error) => void}>;

    constructor(opts: IClientOptions, proxy: Transform, socket: WebSocket) {
        // ...
        this.writeQueue = [];
    }

   _write(chunk: any, encoding: string, cb: (err?: Error) => void) {
        if (!this.isSocketOpen) {
            // Buffer the data in a queue
            this.writeQueue.push({chunk, encoding, cb});
        } else {
            this.writeToProxy(chunk, encoding, cb);
        }
    }

    socketReady() {
        // ...
        this.processWriteQueue();
    }

    private writeToProxy(chunk: any, encoding: string, cb: (err?: Error) => void) {
        if (this.proxy.write(chunk, encoding) === false) {
            this.proxy.once('drain', cb);
        } else {
            cb();
        }
    }

    private processWriteQueue() {
        while (this.writeQueue.length > 0) {
            const {chunk, encoding, cb} = this.writeQueue.shift()!;
            this.writeToProxy(chunk, encoding, cb);
        }
    }
...
}

[robertsLando]

That's the first way I implemented it (I mean by using a queue) but preferred this solution as it doesn't require another variable. Not sure if making the method async makes any difference, the underlying code handling stream expects it to be sync so will never await it anyway (also unhandled rejections are impossible)

[mcollina]

This method must not be async

[robertsLando]

@mcollina Ok, fixed now 🙏🏼

[robertsLando]

@mcollina kindly ping

[mcollina]

This method must not be async

[robertsLando]

@mcollina ping

[mcollina]

lgtm