jsy-app/utils/mqttUtil.js

import * as commonUtils from "@/utils/common.js"
// #ifdef H5
import mqtt from "mqtt/dist/mqtt"
// #endif
// #ifdef APP-PLUS
import {
	connect,
	isConnected,
	subscribe,
	unSubscribe,
	getConfig,
	onConnectLost,
	onReconnect,
	disConnect,
	publishMessage
} from "@/uni_modules/zy-mqtt";
// #endif

//封装一个类（可直接cv）
class MqttUtil {
	//创建公共变量 
	static url;
	static options;
	static client;
	static connected = false;
	static reconnectCount = 0;

	//接收创建来的数据
	constructor(messageEvent) {
		this.messageEvent = messageEvent;

		const url = import.meta.env.VITE_BASE_MQTT;
		// #ifdef H5
		this.url = `mqtt://${url}`;
		// #endif
		// #ifdef APP-PLUS
		this.url = `ws://${url}`;
		// #endif
	}
	//初始化参数
	init(userId) {
		this.userId = userId;
		// let clientId = this.getClientId();
		let username = "app_client";
		let password = "app_client@2023";
		let keepalive = 10; //保持连接的心跳间隔，单位为秒。
		let connectTimeout = 30; //接超时的时间，单位为秒。 
		// #ifdef H5
		this.options = {
			clientId: null, //	string	客户端的唯一标识符。 
			username: username, //	string	连接时的用户名。 
			password: password, //	string	连接时的密码。
			keepalive: keepalive, //	number	60	保持连接的心跳间隔，单位为秒。
			clean: false, //	boolean	true	是否清除会话信息。false 时，断线重连后可继续接收未到达的消息。 
			reconnectPeriod: 1000, //	number	1000	断开后重连的时间间隔，单位为毫秒。当设置为 0 以后将取消自动重连 
			connectTimeout: connectTimeout * 1000, //	number	30 * 1000	连接超时的时间，单位为毫秒。
			// will: { //	object	无	客户端的遗嘱消息。当客户端异常断开时，broker 将会发送该消息。
			// 	topic: '', // 发布消息的主题
			// 	payload: '', // 要发布的消息
			// 	qos: 0, // 消息等级
			// 	retain: false // 保留消息标识
			// },
			// protocolId: "MQTT", //	string	MQTT	协议名称。
			// protocolVersion: 4, //	number	MQTT 协议版本，4 表示 MQTT v3.1.1。
			resubscribe: true, //	boolean	true	断线重连后是否自动重新订阅之前的主题。
		}
		// #endif

		// #ifdef APP-PLUS
		this.options = {
			host: this.url, // 主机ip
			clientId: null, // 客户端id
			userName: username, // 用户名
			password: password, // 密码
			heartBeat: keepalive, // 可选,默认60秒, 心跳
			cleanSession: false, // 可选,默认false,是否开启持续会话
			automaticReconnect: true, // 可选,默认true,是否自动重连
			timeOut: connectTimeout, // 可选,默认30秒,连接超时时间
			keepAlive: false, // 可选,是否保活(息屏保活), 对于Android O及以上版本，如果你的应用处于后台，你需要将服务转变为前台服务来避免系统在一定时间后杀死服务。这需要通过创建一个持续在通知栏显示的通知来实现。
			// notificationContentText, // 可选,通知内容,默认(应用正在后台运行,这有助于提供定制的服务和改善用户体验。)
			// notificationContentTitle, // 可选,通知标题,默认(后台应用持续运行)
		};
		// #endif
	}
	//获取clientId
	getClientId() {
		function S4() {
			return (((1 + Math.random()) * 0x10000) | 0).toString(16).substring(1);
		}
		let UUID = (S4() + S4() + "-" + S4() + "-" + S4() + "-" + S4() + "-" + S4() + S4() + S4());

		return `mqtt_app_${this.userId}_${UUID}`;
	}
	//清理连接信息
	cleanClient() {
		// this.options.clientId = null;
		this.client = null;
		this.connected = false;
		this.reconnectCount = 0;
	}
	//监听消息回调
	onMessageEvent(topic, data) {
		this.messageEvent(topic, data);
	}


	// #ifdef H5
	//mqtt链接
	link(userId) {
		if (!this.options) {
			this.init(userId);
		}
		if (this.connected) {
			return;
		}
		if (!this.options.clientId) {
			this.options.clientId = this.getClientId();
		}

		this.client = mqtt.connect(this.url, this.options);
		this.client.on("connect", error => {
			this.connected = this.client.connected;
			this.reconnectCount = 0;
			console.log("connect:", this.client)
			console.log("connect:", this.connected)
		})
		this.client.on("reconnect", error => {
			this.reconnectCount++;
		})
		this.client.on("error", error => {
			console.log("error", error)
		})
		this.client.on("message", (topic, message) => {
			console.log("接受消息")
			console.log("message：", message.toString())
			this.onMessageEvent(topic, message);
		})
		this.client.on("close", () => {
			console.log("已断开连接1")
			this.connected = false;
		});
	}
	//结束链接
	over() {
		this.client.end(false, null, () => {
			console.log("over")
			this.cleanClient();
		});
	}
	//订阅主题
	subscribes(topics) {
		if (!this.connected) {
			return;
		}
		console.log("subscribes", topics)
		this.client.subscribe(topics, error => {
			if (error) {
				console.log(`订阅主题失败:`, error)
			} else {
				console.log(`订阅主题成功`)
			}
		})
	}
	//取消订阅
	unsubscribes(topics) {
		if (!this.connected) {
			return;
		}
		console.log("unsubscribes", topics)
		this.client.unsubscribe(topics, error => {
			if (!error) {
				console.log("取消订阅成功")
			} else {
				console.log("取消订阅错误", topic)
			}
		})
	}
	//发送消息
	send(message) {
		// this.client.publish("test", JSON.stringify(message), res => {
		// 	console.log(res)
		// })
	}
	// #endif

	// #ifdef APP-PLUS
	//mqtt链接
	link(userId) {
		if (!this.options) {
			this.init(userId);
		}
		if (this.connected) {
			return;
		}
		if (!this.options.clientId) {
			this.options.clientId = this.getClientId();
		}

		// 先连接mqtt
		connect(this.options, (res) => {
			console.log("connect：", res);
			// if (res.code == 200) {}
			this.connected = isConnected();
			this.reconnectCount = 0;
			console.log("connect:", this.connected)
		})
		// 监听连接丢失函数
		onConnectLost((res) => {
			console.log("onConnectLost：", res);
			this.connected = isConnected();
			console.log("onConnectLost:", this.connected)
		})
		// 监听自动重新连接函数
		onReconnect((res) => {
			this.reconnectCount++;
		})
	}
	//结束链接
	over() {
		if (!isConnected()) {
			return;
		}
		disConnect((res) => {
			console.log("disConnect:", res)
			if (res.code == 200) {
				this.cleanClient();
			}
		})
	}
	//订阅主题
	subscribes(topics) {
		if (!this.connected) {
			return;
		}
		console.log("subscribes", topics)
		if (topics && topics.length) {
			topics.forEach(x => {
				subscribe(x, 0, (res) => {
					console.log("接受消息")
					var message = JSON.stringify(res);
					console.log("message：", message.toString());
				})
			});
		}
	}
	//取消订阅
	unsubscribes(topics) {
		if (!this.connected) {
			return;
		}
		console.log("unsubscribes", topics)
		if (topics && topics.length) {
			topics.forEach(x => {
				unSubscribe(x, (res) => {})
			});
		}
	}
	//发送消息
	send(message) {
		// publishMessage('your/topic', 2, message, (res) => {
		// })
	}
	// #endif
}

export default MqttUtil