feat(browser): websockets improvements and bundle optimizations (#1732)

esbuild.js

@@ -15,6 +15,7 @@ const options = {
     format: 'iife',
     platform: 'browser',
     globalName: 'mqtt',
+    sourcemap: false, // this can be enabled while debugging, if we decide to keep this enabled we should also ship the `src` folder to npm
     define: {
         'global': 'window'
     },

package.json

@@ -113,7 +113,6 @@
     "commist": "^3.2.0",
     "concat-stream": "^2.0.0",
     "debug": "^4.3.4",
-    "duplexify": "^4.1.2",
     "help-me": "^4.2.0",
     "lru-cache": "^10.0.1",
     "minimist": "^1.2.8",
@@ -130,7 +129,6 @@
     "@esm-bundle/chai": "^4.3.4-fix.0",
     "@release-it/conventional-changelog": "^7.0.2",
     "@types/chai": "^4.3.10",
-    "@types/duplexify": "^3.6.4",
     "@types/node": "^20.9.0",
     "@types/sinon": "^17.0.1",
     "@types/tape": "^5.6.4",

src/lib/BufferedDuplex.ts

@@ -0,0 +1,105 @@
+import { Duplex, Transform } from 'readable-stream'
+import { IClientOptions } from './client'
+
+/**
+ * Utils writev function for browser, ensure to write Buffers to socket (convert strings).
+ */
+export function writev(
+	chunks: { chunk: any; encoding: string }[],
+	cb: (err?: Error) => void,
+) {
+	const buffers = new Array(chunks.length)
+	for (let i = 0; i < chunks.length; i++) {
+		if (typeof chunks[i].chunk === 'string') {
+			buffers[i] = Buffer.from(chunks[i].chunk, 'utf8')
+		} else {
+			buffers[i] = chunks[i].chunk
+		}
+	}
+
+	this._write(Buffer.concat(buffers), 'binary', cb)
+}
+
+/**
+ * How this works:
+ * - `socket` is the `WebSocket` instance, the connection to our broker.
+ * - `proxy` is a `Transform`, it ensure data written to the `socket` is a `Buffer`.
+ * This class buffers the data written to the `proxy` (so then to `socket`) until the `socket` is ready.
+ * The stream returned from this class, will be passed to the `MqttClient`.
+ */
+export class BufferedDuplex extends Duplex {
+	public socket: WebSocket
+
+	private proxy: Transform
+
+	private isSocketOpen: boolean
+
+	private writeQueue: Array<{
+		chunk: any
+		encoding: string
+		cb: (err?: Error) => void
+	}>
+
+	constructor(opts: IClientOptions, proxy: Transform, socket: WebSocket) {
+		super({
+			objectMode: true,
+		})
+		this.proxy = proxy
+		this.socket = socket
+		this.writeQueue = []
+
+		if (!opts.objectMode) {
+			this._writev = writev.bind(this)
+		}
+
+		this.isSocketOpen = false
+
+		this.proxy.on('data', (chunk) => {
+			this.push(chunk)
+		})
+	}
+
+	_read(size?: number): void {
+		this.proxy.read(size)
+	}
+
+	_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)
+		}
+	}
+
+	_final(callback: (error?: Error) => void): void {
+		this.writeQueue = []
+		this.proxy.end(callback)
+	}
+
+	/** Method to call when socket is ready to stop buffering writes */
+	socketReady() {
+		this.emit('connect')
+		this.isSocketOpen = true
+		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)
+		}
+	}
+}

src/lib/client.ts

@@ -440,6 +440,10 @@ export default class MqttClient extends TypedEventEmitter<MqttClientEventCallbac
 
 	public pingTimer: any
 
+	/**
+	 * The connection to the Broker. In browsers env this also have `socket` property
+	 * set to the `WebSocket` instance.
+	 */
 	public stream: IStream
 
 	public queue: { packet: Packet; cb: PacketCallback }[]

src/lib/connect/ali.ts

@@ -1,12 +1,12 @@
 import { Buffer } from 'buffer'
 import { Transform } from 'readable-stream'
-import duplexify, { Duplexify } from 'duplexify'
 import { StreamBuilder } from '../shared'
 import MqttClient, { IClientOptions } from '../client'
+import { BufferedDuplex } from '../BufferedDuplex'
 
 let my: any
 let proxy: Transform
-let stream: Duplexify
+let stream: BufferedDuplex
 let isInitialized = false
 
 function buildProxy() {
@@ -64,9 +64,7 @@ function bindEventHandler() {
 	isInitialized = true
 
 	my.onSocketOpen(() => {
-		stream.setReadable(proxy)
-		stream.setWritable(proxy)
-		stream.emit('connect')
+		stream.socketReady()
 	})
 
 	my.onSocketMessage((res) => {
@@ -91,8 +89,8 @@ function bindEventHandler() {
 		stream.destroy()
 	})
 
-	my.onSocketError((res) => {
-		stream.destroy(res)
+	my.onSocketError((err) => {
+		stream.destroy(err)
 	})
 }
 
@@ -112,13 +110,14 @@ const buildStream: StreamBuilder = (client, opts) => {
 
 	const url = buildUrl(opts, client)
 	my = opts.my
+	// https://miniprogram.alipay.com/docs/miniprogram/mpdev/api_network_connectsocket
 	my.connectSocket({
 		url,
 		protocols: websocketSubProtocol,
 	})
 
 	proxy = buildProxy()
-	stream = duplexify.obj()
+	stream = new BufferedDuplex(opts, proxy, my)
 
 	bindEventHandler()