在基于 Taro 的微信、钉钉小程序中使用 MQTT

使用的开发环境是：

Taro 版本：4.0.5
MQTT 版本：5.10.3
微信开发者工具：Stable 1.06.2409140
（钉钉）小程序开发者工具：3.1.3

修改好的 MQTT 库文件及依赖文件可以参考这个 Gist。

仅需要在微信小程序中使用 MQTT

如果仅需要编译为微信小程序，可以参考这个 issue 中的回答，基本步骤是：

首先安装依赖：

yarn install mqtt abortcontroller-polyfill

由于 mqtt.min.js 文件是以 iife 格式打包的，因此如果希望在小程序中以 import 的语法导入，需要进行一点小修改，具体来说是在 node_modules/mqtt/dist/mqtt.min.js 中直接导出：

// 将 `"use strict";var mqtt=` 修改为：
"use strict";module.exports=
node_modules/mqtt/dist/mqtt.min.js

修改完成后更新 config/index.ts 文件，让 Taro 不要编译 node_modules/mqtt/dist/mqtt.min.js 文件：

export default defineConfig<"webpack5">(async (merge, _) => {
  // ...
  mini: {
		compile: {
		  // ...
			exclude: [
			  // ...
				path.resolve(__dirname, "..", "node_modules/mqtt/dist/mqtt.min.js"),
				// ...
			],
			// ...
		},
  }
  // ...
}
config/index.ts

配置更新后，按照下面的方式引入和连接 MQTT 服务，然后重新启动 dev server：

import 'abortcontroller-polyfill/dist/abortcontroller-polyfill-only' // import before mqtt.
import 'esbuild-plugin-polyfill-node/polyfills/navigator'
import mqtt from'mqtt/dist/mqtt.min'

const client = mqtt.connect("wxs://test.mosquitto.org", {
  timerVariant: 'native' // more info ref issue: #1797
});

要注意的是，在使用中，不论是微信小程序还是钉钉小程序，都不支持 TextDecoder，因此需要在项目中引入第三方实现，可以参考这个文件。

构建支持在钉钉小程序中使用的 MQTT 库文件

如果不仅要在微信小程序中使用，同时也要在钉钉小程序中使用，则需要调整 mqtt.js 的源码，增加对钉钉小程序的适配，并且调整编译脚本，以便能够通过钉钉小程序开发工具的编译流程。

在 MQTT 中适配钉钉小程序 socket 接口

首先将 MQTT.js clone 到本地，然后新建一个文件——src/lib/connect/dd.ts，并在其中添加如下内容：

import { Buffer } from "buffer";
import { Transform } from "readable-stream";
import type { StreamBuilder } from "../shared";
import MqttClient, { IClientOptions } from "../client";
import { BufferedDuplex } from "../BufferedDuplex";

let my: any;
let proxy: Transform;
let stream: BufferedDuplex;
let isInitialized = false;

function buildProxy() {
	const _proxy = new Transform();
	_proxy._write = (chunk: ArrayBuffer, _encoding: string, next: (_: any) => void) => {
		my.sendSocketMessage({
			data: Buffer.from(chunk.buffer).toString("base64"),
			isBuffer: true,
			success() {
				next();
			},
			fail(e) {
				next(new Error());
			},
		});
	};

	_proxy._flush = (_done) => {
		my.closeSocket({});
	};

	return _proxy;
}

function setDefaultOpts(opts: IClientOptions) {
	if (!opts.hostname) {
		opts.hostname = "localhost";
	}
	if (!opts.path) {
		opts.path = "/";
	}
	if (!opts.wsOptions) {
		opts.wsOptions = {};
	}
	opts.protocol = "wss";
}

function buildUrl(opts: IClientOptions, client: MqttClient) {
	const protocol = opts.protocol === "dds" ? "wss" : "ws";
	let url = `${protocol}://${opts.hostname}${opts.path}`;
	if (opts.port && opts.port !== 80 && opts.port !== 443) {
		url = `${protocol}://${opts.hostname}:${opts.port}${opts.path}`;
	}
	if (typeof opts.transformWsUrl === "function") {
		url = opts.transformWsUrl(url, opts, client);
	}
	return url;
}

function bindEventHandler() {
	if (isInitialized) return;
	isInitialized = true;

	my.onSocketOpen(() => {
		stream.socketReady();
	});

	my.onSocketMessage((res) => {
		if (res.isBuffer) {
			proxy.push(Buffer.from(res.data, "base64"));
		} else {
			proxy.push(Buffer.from(res.data, "utf8"));
		}
	});

	my.onSocketClose(() => {
		stream.emit("close");
		stream.end();
		stream.destroy();
	});

	my.onSocketError((e) => {
		stream.destroy(e);
	});
}

const buildStream: StreamBuilder = (client: MqttClient, opts: IClientOptions) => {
	opts.hostname = opts.hostname || opts.host;

	if (!opts.hostname) {
		throw new Error("Could not determine host. Specify host manually.");
	}

	setDefaultOpts(opts);

	const url = buildUrl(opts, client);

	my = opts.my;

	my.connectSocket({
	  url,
		success: () => console.log("[INFO] connect to socket"),
		fail: (err) => console.log("[ERROR] can not connect to socket", err)
	});

	proxy = buildProxy();
	stream = new BufferedDuplex(opts, proxy, my);
	bindEventHandler();
	return stream;
};

export default buildStream;
src/lib/connect/dd.ts

上面这段代码是从 src/lib/connect/ali.ts 复制过来的，主要是针对钉钉小程序的 socket 相关接口进行了适配，这些改动自上而下分别是：

修改 my.sendSocketMessage 接口，支付宝小程序接口的 data 字段支持传递 ArrayBuffer，而钉钉小程序只支持传递字符串（utf8 字符串或 base64 编码后的 ArrayBuffer），要注意，由于 MQTT 无法解析钉钉接口发送的 utf8 string，因此一定要使用 isBuffer 的格式（即编码为 base64 格式的字符串）发送；
在 buildUrl 方法中使用 dd(s) 替换了 ali(s)；
修改 my.onSocketMessage 接口，针对钉钉小程序的返回值进行了调整；
调整 my.connectSocket 方法，移除了 protocols 字段。dd.connectSocket 方法不支持 protocols 参数，如果添加该参数在模拟器中不会有问题，但是真机上会无法连接（不会报错 ...）。

添加完 src/lib/connect/dd.ts 文件后，需要将 dd(s) 添加到有效的协议中，更新 src/lib/connect/index.ts 文件：

// ...
function connect(
	brokerUrl: string | IClientOptions,
	opts?: IClientOptions,
): MqttClient {
	// ...
	if (opts.cert && opts.key) {
		if (opts.protocol) {
			// add `dds` for dingtalk mini-program
			if (
				["mqtts", "wss", "wxs", "alis", "dds"].indexOf(opts.protocol) === -1
			) {
				switch (opts.protocol) {
					case "mqtt":
						opts.protocol = "mqtts";
						break;
					case "ws":
						opts.protocol = "wss";
						break;
					case "wx":
						opts.protocol = "wxs";
						break;
					case "ali":
						opts.protocol = "alis";
						break;
					// for dingtalk mini-program
					case "dd":
						opts.protocol = "dds";
						break;
					default:
						throw new Error(
							`Unknown protocol for secure connection: "${opts.protocol}"!`,
						);
				}
			}
		} else {
			// A cert and key was provided, however no protocol was specified, so we will throw an error.
			throw new Error("Missing secure protocol key");
		}
	}
	// ...
	// only loads the protocols once
	if (!protocols) {
		protocols = {};
		if (!isBrowser && !opts.forceNativeWebSocket) {
			protocols.ws = require("./ws").streamBuilder;
			protocols.wss = require("./ws").streamBuilder;
			protocols.mqtt = require("./tcp").default;
			protocols.tcp = require("./tcp").default;
			protocols.ssl = require("./tls").default;
			protocols.tls = protocols.ssl;
			protocols.mqtts = require("./tls").default;
		} else {
			protocols.ws = require("./ws").browserStreamBuilder;
			protocols.wss = require("./ws").browserStreamBuilder;
			protocols.wx = require("./wx").default;
			protocols.wxs = require("./wx").default;
			protocols.ali = require("./ali").default;
			protocols.alis = require("./ali").default;
			// 增加 dd(s)
			protocols.dd = require("./dd").default;
			protocols.dds = require("./dd").default;
		}
	}
	// ...
	if (!protocols[opts.protocol]) {
		const isSecure = ["mqtts", "wss"].indexOf(opts.protocol) !== -1;
		// returns the first available protocol based on available protocols (that depends on environment)
		// if no protocol is specified this will return mqtt on node and ws on browser
		// if secure it will return mqtts on node and wss on browser
		opts.protocol = [
			"mqtt",
			"mqtts",
			"ws",
			"wss",
			"wx",
			"wxs",
			"ali",
			"alis",
			"dd", // 增加 dd(s)
			"dds",
		].filter((key, index) => {
			if (isSecure && index % 2 === 0) {
				// Skip insecure protocols when requesting a secure one.
				return false;
			}
			return typeof protocols[key] === "function";
		})[0] as MqttProtocol;
	}
	// ...
}
// ...
src/lib/connect/index.ts

然后更新 src/lib/client.ts 文件，将 dd(s) 加入到类型定义中避免编译错误：

// ...
// 增加 dd(s)
export type BaseMqttProtocol =
	| 'wss'
	| 'ws'
	| 'mqtt'
	| 'mqtts'
	| 'tcp'
	| 'ssl'
	| 'wx'
	| 'wxs'
	| 'ali'
	| 'alis'
	| 'dd'
	| 'dds'
// ...
src/lib/client.ts

更新 MQTT 编译脚本

完成上面的调整后可以使用下面的脚本测试是否能够正常编译：

yarn build

如果没有报错，应当能够在 dist 文件夹下看到 mqtt.js、mqtt.min.js 以及 mqtt.esm.js 文件。但要注意的是直接使用 MQTT 的编译脚本编译出来的文件在钉钉小程序的开发工具（至少我使用的“小程序开发者工具 3.1.3”）中无法正常运行，看报错信息应该是由于开发者工具中的编译工具仅支持 es5，因此需要更新编译脚本，具体来说需要：

更新 tsconfig.json 文件，将编译目标 target 设为 ES5 并且开启 downlevelIteration（为了处理依赖文件中的 Generator 语法）。如果更新完之后直接运行 yarn build:ts 会报一个 big integer literal 相关的错误，这是由于 MQTT 的依赖 readable-stream 中使用了 BigInt 相关的语法，然而 typescript 并不支持直接将这种写法编译成 es5，所以需要手工修改一下依赖文件——node_modules/readable-stream/lib/ours/errors.js，移除 314 行的 big integer literal 表示，具体如下：

// ...
E(
  'ERR_OUT_OF_RANGE',
  (str, range, input) => {
    assert(range, 'Missing "range" argument')
    let received
    if (Number.isInteger(input) && Math.abs(input) > 2 ** 32) {
      received = addNumericalSeparator(String(input))
    } else if (typeof input === 'bigint') {
      received = String(input)
      // 将 2n ** 32n 修改成下面的样子
      if (input > 2 ** 32 || input < -(2 ** 32)) {
        received = addNumericalSeparator(received)
      }
      received += 'n'
    } else {
      received = inspect(input)
    }
    return `The value of "${str}" is out of range. It must be ${range}. Received ${received}`
  },
  RangeError
)
// ...
node_modules/readable-stream/lib/ours/errors.js

修改完成后，使用 yarn build:ts 应当能给顺利编译了，但是为了能够顺利在钉钉小程序中使用，还需要移除编译最终编译结果中的 spread 语法，首先添加 babel 及相关的依赖：

yarn add -D @babel/cli @babel/core @babel/plugin-transform-classes @babel/plugin-transform-spread @babel/preset-env

然后在项目根目录下添加 babel.config.js 文件并添加如下内容：

const presets = [["@babel/preset-env"]];
module.exports = { presets, plugins: ["@babel/plugin-transform-spread", "@babel/plugin-transform-classes"] };
babel.config.js

最后将 esbuild.js 中的 format 从 iife 修改为 cjs，然后执行下面的脚本编译得到可以被钉钉小程序使用的文件：

yarn build && npx babel dist/mqtt.min.js --out-dir dist_for_dd/

将 dist_for_dd/mqtt.min.js 移动到 Taro 项目中以便后续使用。

针对多端的文件名调整

虽然上面构建的代码在编译目标是微信小程序时也可以使用，但是为了适配钉钉小程序，重新编译后的 MQTT 库文件大小达到了 654KB，对于 2M 的限制影响还是很大的；并且如果项目中引入了第三方插件，也可能会发生不可预见的错误，建议为不同的端的 MQTT 库文件添加 env 以实现隔离，例如：

.
├── mqtt.min.weapp.js
├── mqtt.min.dd.js
└── ...

注意，使用这种写法要在 config/index.ts 中将各个端的库文件都排除编译（参见上面更新 config/index.ts 的代码）。后文内容仅针对 mqtt.min.dd.js 文件，不在赘述。

在代码中引入 MQTT 库文件

经过前面的步骤已经完成了对 MQTT 的调整，但是由于钉钉小程序的 JavaScript 引擎不支持访问 globalThis 和 global，因此还需要对上述步骤生成的库文件进行小修改，具体来说：

首先在 /path/to/your/mqtt.min.dd.js 文件的最上方声明一个 globalThis 变量，即：

"use strict"
var globalThis = {}
// 其他代码 ...
path/to/your/mqtt.min.js

由于 MQTT 依赖的 abortcontroller-polyfill 中用到了 global 变量，因此也需要进行相应的调整，我的做法是将 node_modules/abortcontroller-polyfill/dist/abortcontroller-polyfill-only.js 复制到项目代码中，然后修改最后执行 polyfill 部分的代码，即：

// ...
(function (self) {

  if (!polyfillNeeded(self)) {
    return;
  }

  self.AbortController = AbortController;
  self.AbortSignal = AbortSignal;
})(typeof self !== 'undefined' ? self : window); // 将这里的 global 替换为 window
path/to/your/abortcontroller-polyfill-only.js

更新 Taro 的 config/index.ts 文件，不要编译手工引入的两个文件：

export default defineConfig<"webpack5">(async (merge, _) => {
  // ...
  mini: {
		compile: {
		  // ...
			exclude: [
			  // ...
				path.resolve(__dirname, "..", "path/to/your/mqtt.min.dd.js"),
				path.resolve(__dirname, "..", "path/to/your/abortcontroller-polyfill-only.js"),
				// ...
			],
			// ...
		},
  }
  // ...
}
config/index.ts

最后再额外安装两个依赖：

yarn add fast-unique-numbers worker-timers-broker

然后参考下面的代码进行连接：

import 'path/to/your/abortcontroller-polyfill-only.js' // import before mqtt.
import mqtt from'path/to/your/mqtt.min.js'

const client = mqtt.connect("wxs://test.mosquitto.org", {
  timerVariant: 'native' // more info ref issue: #1797
  // @ts-ignore
  my: dd, // 将 dd 实例传给 mqtt 用于处理 WebSocket 相关操作
  log: console.log, // 传递一个 log 方法可以便于调试
});

至此就完成了在基于 Taro 的钉钉小程序中使用 MQTT 的调整，此时启动 dev server 应该就能正常运行了。

如果在小程序开发者工具中导入 Production Build 的时候报了 TypeError: Constructor Map requires 'new' 的错误，请参考后文的解决方法。

钉钉小程序无法编译 Taro Production Build——`TypeError: Constructor Map requires 'new'`

钉钉的小程序开发者工具在启动 Taro Production Build 时报错 TypeError: Constructor Map requires 'new'，根据这个 issue 的说法此问题应该已经修复，但是在我本地依旧会稳定出现，由于钉钉的小程序开发者工具没有办法调整编译选项，因此只能手动修改 Taro 库文件来规避这个问题，具体的修改位置是在 node_modules/@tarojs/runtime/dist/runtime.esm.js 的 1736 行，重新定义一个 EventSource 类，具体如下：

// ...
class EventSource {
	data = {};

	get(id) {
		return this.data[id];
	}

	set(id, value) {
		this.data[id] = value;
	}

	delete(id) {
		delete this.data[id];
	}

	has(id) {
		return id in this.data;
	}

	removeNode(child) {
		const { sid, uid } = child;
		this.delete(sid);
		if (uid !== sid && uid) this.delete(uid);
	}

	removeNodeTree(child) {
		this.removeNode(child);
		const { childNodes } = child;
		childNodes.forEach((node) => this.removeNodeTree(node));
	}
}

// class EventSource extends Map {
//     removeNode(child) {
//         const { sid, uid } = child;
//         this.delete(sid);
//         if (uid !== sid && uid)
//             this.delete(uid);
//     }
//     removeNodeTree(child) {
//         this.removeNode(child);
//         const { childNodes } = child;
//         childNodes.forEach(node => this.removeNodeTree(node));
//     }
// }
const eventSource = new EventSource();
// ...
node_modules/@tarojs/runtime/dist/runtime.esm.js

完成修改后重新 build 一个 production 版本即可。