uniapp自定义websocket类实现socket通信、心跳检测、连接检测、重连机制

uniapp自定义websocket类实现socket通信、心跳检测、检测连接、重连机制，仿vue-socket插件功能实现发送序列号进行连接检测，发送消息时42【key,value】格式，根据后端返回数据和需要接收到的数据做nsend与onSocketMessage的修改

//使用socket类
				this.socket = new websocket({
					url: chatUrl,
					onOpen: (res) => {
						this.socket?.nsend({
							type: "bind",
							fromid: uni.getStorageSync("project_user").uid
						})

						uni.$on("message", function(data) {
							that.recvMsg(data)
						})

						console.log('onOpen', res)
					},
					onClose: (res) => {
						console.log('onClose', res)
					},
					onRdFinsh: (res) => {
						console.log('onRdFinsh', res)
					}
				})

import {
	publish,
	subscribe,
	unsubscribe
} from 'pubsub-js'
import {
	isArray
} from "@/utils/validate"

const noop = function() {};
class Socket {
	//心跳检测定时器
	static stopTime = 0;
	//每25s发送报文给服务端监听连接是否正常
	static sendInterval = 0;
	//服务端监听连接异常时提示并进行socket重连
	static isCommunication = false;
	//连接中出错调用心跳检测方法的定时器
	static errorNr = '';
	//websocket已连接数
	static concatCount = 0;
	//连接后连接设备次数（用于拼接传输序号42后的数字）
	static sendTag = 0
	//pc端emit发送消息有回调函数时加入
	static isCallback = ["leave", "join"]
	//连接或断连publish相应isCallback数组内的值给客户端响应
	static sendKey = "";

	//new Socket时执行构造方法，对参数默认值进行初始化
	constructor({
		url = '',
		onOpen = noop,
		onMsg = noop,
		onClose = noop,
		onError = noop,
		onReload = noop,
		onRdFinsh = noop,
		maxInterValCount = 10,
		interValTime = 2000,
		...args
	} = {}) {
		this.isRconnectIng = false; //是否处于重连状态
		this.waiting = Promise.resolve(false); //心跳检查必须等待重连完成后
		this.waitDep = []; //等待时收集依赖的容器

		this.SocketTask = {
			nsend: noop,
			hbDetection: noop,
			nclose: noop,
			nrconnect: noop,
			isconnect: false,
			uniColse: false,
			maxInterValCount,
			interValTime,
			InterValCount: 0,
			eventPatch: null,
			url,
			onOpen,
			onMsg,
			onClose,
			onError,
			onReload,
			onRdFinsh,
			extra: args
		};

		//为对象提供连接方法
		this._EventDispath(this.SocketTask);
		//初始化websocket
		this.initChat(this.SocketTask, this.SocketTask.extra);
		return this.SocketTask;

	}

	set CONCATCOUNT(value) {
		Socket.concatCount = value;
		if (value > 0) this._notify();
	}
	get CONCATCOUNT() {
		return Socket.concatCount
	}

	/**
	 * 每25s给服务端发送一次数据,确保连接未断开
	 */
	initSendTime() {
		Socket.sendInterval = setInterval(() => {
			if (Socket.isCommunication) {
				uni.showToast({
					icon: 'none',
					title: '网络异常',
					duration: 2000
				})
				this.SocketTask.nrconnect()
			} else {
				Socket.isCommunication = true
				this.SocketTask.nsend(2)
			}
		}, 25000)
		subscribe("number", function(msg, data) {
			Socket.isCommunication = false
		})
	}

	//销毁定时器
	destroySend() {
		clearTimeout(Socket.errorNr)
		clearTimeout(Socket.stopTime);
		clearInterval(Socket.sendInterval)
		unsubscribe("number")
	}
	/**
	 * 仅供内部使用，通知所有收集到的依赖
	 */
	_notify() {
		for (let i = 0; i < this.waitDep.length; i++) {
			this.waitDep[i].call(this.SocketTask);
		}
		this.waitDep = [];
	}
	/**
	 * 仅供内部使用，确认当前是否连接成功,收集依赖
	 */
	_chunkConnect(fn) {
		//没有连接时收集起来，当连接成功时开始发送命令等操作
		if (Socket.concatCount > 0) {
			fn();
		} else {
			this.waitDep.push(fn);
		}
	}
	/**
	 * 仅供内部使用，事件注册
	 */
	_EventDispath({
		onReload
	} = {}) {
		let SocketTask = this.SocketTask;
		let events = {
			onOpen: [],
			onMsg: [],
			onClose: [],
			onError: [],
			onReload: [],
			onRdFinsh: [],
		}
		SocketTask.hbDetection = () => {
			this.hbDetection()
		}
		//适配后端websocket发送请求数据
		//vue-socket-io 传入的是数组，第一位为key值
		SocketTask.nsend = (key, value) => {
			let that = this
			this._chunkConnect(() => {
				let data = "42"
				//web端连接或断开连接的时候传入420 421
				if (Socket.isCallback.includes(key)) {
					//收集当前连接的命令key值，等待返回响应时传递给客户端处理逻辑
					Socket.sendKey = key
					//421 422 ··· 依次递增
					data += Socket.sendTag
					Socket.sendTag++
				}
				if (value) {
					data += JSON.stringify([key, value])
				} else {
					//特殊情况，当想要发送的请求不为42【key,value】格式时，key为客户端代码发送传入的值
					data = key + ""
				}

				uni.sendSocketMessage({
					//pc端emit事件有发送完毕回调函数时后端会需要有多一位数字的判断，并且返回时也会产生eg：420 -》 430
					// 没有回调函数时发送和接收命令前面都是42
					data: data,
					complete(e) {},
				})
			})
		}
		SocketTask.nclose = t => {
			this.destroySend()
			this._chunkConnect(() => {
				SocketTask.uniColse = true;
				uni.closeSocket();
			})
		}
		SocketTask.nrconnect = t => {
			this._chunkConnect(() => {
				this.waiting = new Promise(async (resolve) => {
					uni.closeSocket();
					let reloadStatus = false;
					try {
						const res = await this.initChat(SocketTask, SocketTask.extra);
						reloadStatus = res;
					} catch (e) {}
					onReload.call(SocketTask, reloadStatus, SocketTask);
					SocketTask.eventPatch.dispatchEvent('onReload', reloadStatus);
					resolve(reloadStatus);
				})
			})
		}

		function EventDispatcher() {
			this.events = events;
		}
		for (let key in events) {
			//绑定监听方法到EventDispatcher方法原型中
			EventDispatcher.prototype[key] = function(handler) {
				if (typeof handler != 'function') return;
				this.events[key].push(handler)
			}
		}
		//调用监听方法 监听连接状态
		EventDispatcher.prototype.dispatchEvent = function(type, msg) {
			let evenArr = this.events[type];
			if (evenArr.length > 0) {
				for (let i = 0; i < evenArr.length; i++) {
					evenArr[i].call(SocketTask, msg, SocketTask);
				}
			}
		}
		SocketTask.eventPatch = new EventDispatcher();
	}
	/**
	 * 心跳检测
	 */
	async hbDetection() {
		const SocketTask = this.SocketTask;
		if (SocketTask.uniColse) {
			return false;
		}
		clearTimeout(Socket.stopTime);
		if (!SocketTask.isconnect) { //未连接则启动连接
			if (SocketTask.maxInterValCount > SocketTask.InterValCount) {
				Socket.stopTime = setTimeout(async () => {
					try {
						const R_result = await this.waiting;
						if (R_result) return;
						this.isRconnectIng = true;
						const openResult = await this.initChat(SocketTask, SocketTask.extra);
						if (openResult) {
							SocketTask.InterValCount++;
							return;
						}
						return this.hbDetection();
					} catch (e) {
						return this.hbDetection();
					}
				}, SocketTask.interValTime)
			} else {
				SocketTask.onRdFinsh.call(SocketTask, SocketTask.maxInterValCount, SocketTask);
				SocketTask.eventPatch.dispatchEvent('onRdFinsh', SocketTask.maxInterValCount);
			}
		}
	}
	/**
	 * websocket监听事件
	 */
	SocketEvents({
		onOpen,
		onMsg,
		onClose,
		onError,
		onReload,
	} = {}) {
		return new Promise((resolve, reject) => {
			const SocketTask = this.SocketTask;
			uni.onSocketOpen(res => {
				this.CONCATCOUNT += 1;
				this.isRconnectIng = false;
				SocketTask.isconnect = true;
				Socket.sendTag = 0
				Socket.sendKey = ""
				SocketTask.InterValCount = 0;
				SocketTask.uniColse = false;
				resolve(true);
				this.initSendTime()
				onOpen.call(SocketTask, res, SocketTask);
				SocketTask.eventPatch.dispatchEvent('onOpen', res)
			})

			//适配服务器端返回的消息格式
			uni.onSocketMessage(msg => {
				let serialNumber = 0
				let arrayIndex = msg.data.indexOf("[")
				let objIndex = msg.data.indexOf("{")

				//判断返回值是纯数字或json格式
				let dataStart = arrayIndex != -1 && objIndex != -1 ?
					(arrayIndex < objIndex ? arrayIndex : objIndex) :
					(arrayIndex == -1 ? objIndex : arrayIndex)


				if (dataStart == -1) {
					serialNumber = msg.data.substring(0)
					publish("number", serialNumber)
				} else {
					serialNumber = msg.data.substring(0, dataStart)
					let jsonData = JSON.parse(msg.data.substring(dataStart))

					//判断是否为连接事件
					if (serialNumber == `43${Socket.sendTag - 1}`) {
						publish(Socket.sendKey, jsonData[0])
					} else if (isArray(jsonData)) {
						publish(jsonData[0], jsonData[1])
					}

				}

				onMsg.call(SocketTask, msg, SocketTask);
				SocketTask.eventPatch.dispatchEvent('onMsg', msg)
			})
			uni.onSocketClose(async err => {
				this.destroySend()
				SocketTask.isconnect = false;
				SocketTask.InterValCount = 0;
				//微信小程序 连接失败会自动关闭连接，返回false给心跳监测方法
				resolve(false);
				if (!this.isRconnectIng) {
					this.hbDetection();
				}
				onClose.call(SocketTask, err, SocketTask);
				SocketTask.eventPatch.dispatchEvent('onClose', err);
			})
			uni.onSocketError(err => {
				this.destroySend()
				Socket.errorNr = setTimeout(() => {
					this.hbDetection()
				}, 3000)
				onError.call(SocketTask, err, SocketTask);
				SocketTask.eventPatch.dispatchEvent('onError', err)
			})
		})
	}
	/**
	 * 开始初始化chat
	 */
	initChat({
		url,
		onOpen,
		onMsg,
		onClose,
		onError,
		onReload
	} = {}, args) {
		return new Promise(async (resolve, reject) => {
			try {
				await this.connectSocket(url, args);
				let res = await this.SocketEvents({
					onOpen,
					onMsg,
					onClose,
					onError,
					onReload,
				})
				resolve(res);
			} catch (e) {
				console.log('initChat', e)
				reject();
			}
		})
	}
	/**
	 * 连接webSocket
	 */
	connectSocket(url, args) {
		return new Promise((resolve, reject) => {
			uni.connectSocket({
				url,
				success: () => {
					resolve();
				},
				fail: err => {
					reject();
				},
				...args
			})
		})
	}
}
export default Socket