MqttDemo/store/modules/moduleMqtt.js

import mqtt from '../../common/mqtt.js'
import Vue from 'vue'
//import mqtt from 'mqtt/dist/mqtt.js'

const options = {
	clientId: "wx_" + parseInt(Math.random() * 100 + 800, 10),
	keepalive: 28,//每28秒发送一次心跳
	connectTimeout: 15 * 1000,//连接超时的时间
};


let url = 'wss://mqtt.test.radio1964.com'

// #ifdef MP-WEIXIN
url = 'wxs://mqtt.test.radio1964.com'
// #endif



//订阅设备在线状态
function subscribeDevicesStatus(dispatch,deviceList){
	deviceList.forEach((value) =>{
		let topic = `/${value.uuid}/status/onoffline`
		dispatch('subscribe',topic)
	})
}

//订阅消息
function subscribeCurrDevice(dispatch,device){
	if(typeof device !== 'undefined'){
		let topic = `/${device.uuid}/user/pub_response`
		dispatch('subscribe',topic)
	}
}

export default {
	namespaced: true,
	state: {
		//当前选择的设备
		currentDevice: undefined,
		//用户设备列表
		deviceList: [],
		//mqtt客户端
		mqttClient: undefined,
		//收到的消息的历史记录
		msgList:[],

	},
	getters: {
		//mqtt连接状态
		connected: state => {
			return state.mqttClient !== undefined && state.mqttClient.connected
		},

	},
	mutations: {
		//添加设备
		addDevice(state, device) {
			state.deviceList.push(device)
		},
		//设置当前选择的设备
		setCurrentDevice(state, device) {
			state.currentDevice = device
		},
		//移除设备
		removeDevice(state, device) {
			let index = state.deviceList.indexOf(device)
			state.deviceList.slice(index, 1)
		},
		//连接mqtt服务器 clientId为wx_加上用户的uid
		connect(state, clientId) {
			console.log(`connect clientId = ${clientId}`);
			options.clientId = clientId
			state.mqttClient = mqtt.connect(url, options)
		},
		//token失效，退出登录，需要清除数据
		release(state) {
			deviceList = []
			currentDevice = undefined
			state.mqttClient = undefined
		},
		//设置设备的在线状态
		updateDeviceState(state,data){
			Vue.set(state.deviceList,data.index,data.device)
			//state.deviceList[data.index] = {...state.deviceList[data.index], online:data.online}
			//state.deviceList[data.index].online = data.online
		}

	},
	actions: {
		//添加设备
		addDevice({state, dispatch,commit, getters },data){
			let clientId = (data.clientId) ? data.clientId : options.clientId
			let device = data.device

			console.log(`deviceList = ${state.deviceList} device = ${device}`);
			//判断设备是否存在deviceList中
			let findDevice = state.deviceList.find((item) =>{
				return item.uuid === device.uuid
			})
			if(typeof findDevice === 'undefined'){
				if(state.deviceList.length === 0){
					//设置当前的设备为添加的设备
					commit('setCurrentDevice',device)
				}
				//添加设备
				state.deviceList.push(device)
				//如果已经连接
				if(getters.connected){
					//订阅设备是否在线
					subscribeDevicesStatus(dispatch,[device])
					if(typeof state.currentDevice !== 'undefined' && state.currentDevice.uuid === device.uuid){
						//订阅当前设备的消息
						subscribeCurrDevice(dispatch,state.currentDevice)
					}

				}else{//如果没有连接
					//连接服务器
					dispatch('connect',clientId)
				}
				
			}
		},
		//删除设备
		removeDevice({state, dispatch,commit, getters },device){
			//判断设备是否存在deviceList中
			let findDevice = state.deviceList.find((item) =>{
				return item.uuid === device.uuid
			})
			if(typeof findDevice !== 'undefined'){
				//取消此设备的订阅
				let topic = `/${device.uuid}/status/onoffline`
				dispatch('unsubscribe',topic)
				topic = `/${device.uuid}/status/pub_response`
				dispatch('unsubscribe',topic)
				//在设备列表中删除此设备
				commit('removeDevice',findDevice)
				//如果删除的是选中的设备
				if(typeof state.currentDevice !== 'undefined' && state.currentDevice.uuid == device.uuid){
					//清除当前选中的设备 todo 还是自动选中在线的设备
					commit('setCurrentDevice',undefined)
				}
				
				if(state.deviceList.length === 0 ){//所有的设备都已经删除
					dispatch()
				}
			}
		},

		//连接设备
		connect({ state, dispatch,commit, getters }, clientId) {
			if (typeof state.mqttClient === 'undefined') {
				//连接服务器
				commit('connect', clientId)
				//连接成功回调
				state.mqttClient.on("connect", function () {
					console.log("connect success!");
					//订阅设备是否在线
					subscribeDevicesStatus(dispatch,state.deviceList)
					//订阅当前设备的消息
					subscribeCurrDevice(dispatch,state.currentDevice)
				});

				//异常回调
				state.mqttClient.on("error", function (err) {
					console.log(err);
				});

				state.mqttClient.on('offline', function () {

					console.log(`offline callback`);
				})

				state.mqttClient.on('end', function () {

					console.log(`end callback`);
				})
				//网络断开的回调
				state.mqttClient.on('close', function () {
					console.log(`close callback`);
				})
				//收到消息的回调
				state.mqttClient.on('message', function (topic, message) {
					console.log(`message = ${message.toString()}`);
					let arr = topic.split("/");
					if(arr.length >1){
						let fromDeviceUuid = arr[1]
						
						//设备在线和离线的消息
						if(topic.endsWith('onoffline')){
							let index = -1
							for (var i = 0; i < state.deviceList.length; i++) {
								if(state.deviceList[i].uuid === fromDeviceUuid){
									index = i;
									break
								}
							}
							console.log(`state index = ${index}`);
							if(index >= 0){
								let stateJson = JSON.parse(message)
								//console.log(stateJson);
								let device = state.deviceList[i];
								if(stateJson.state === 'online'){
									device.online = true
									commit('updateDeviceState',{index,device})
									//commit('updateDeviceState',{index,online:true})
									//state.deviceList[index].online = true
									if(typeof state.currentDevice !== 'undefined' && state.currentDevice.uuid == device.uuid){
										dispatch({
											type:'publishWithType',
											mqttType:'get_position',
										})
									}
								}else{
									device.online = false
									commit('updateDeviceState',{index,device})
									//commit('updateDeviceState',{index,online:true})
									//state.deviceList[index].online = false
								}
								uni.$emit('mqtt_onoffline',stateJson)
								// state.deviceList.forEach(element => {
								// 	console.log(`online = ${element.online}`);
								//   });
							}
							
						}
					}
					//这是测试的
					state.msgList.push({
						topic: topic.toString(),
						msg: message.toString(),
					});
					


				})
			}else{
				if(!state.mqttClient.connected){
					console.log("reconnect");
					state.mqttClient.reconnect()
					//重连成功的回调
					state.mqttClient.on("reconnect", function() {
						console.log("reconnect success!");
						
					});
				}
			}
		},
		//断开mqtt的连接
		disconnect({state,commit},isRelease) {
			console.log('disconnect')
			if (state.mqttClient !== undefined) {
				state.mqttClient.end(false,() =>{
					console.log(`state.mqttClient.connected = ${state.mqttClient.connected}`);
				})
			}
			if(isRelease){
				//清除数据
				commit('release')
			}
		},
		//发布消息
		publishWithType({state, dispatch,getters},payload){
			//console.log(payload);
			//console.log(options);
			let type = payload.mqttType;
			let other = payload.other;
			if(!getters.connected){
				return new Promise((resolve, reject) => {
					reject('连接已断开')
				});
			}else if(typeof state.currentDevice === undefined || !state.currentDevice.online){
				return new Promise((resolve, reject) => {
					reject('设备已经离线')
				});
			}else{
				let topic = `/${state.currentDevice.uuid}/user/sub_control`;
				//console.log(payload);
				//console.log(payload.mqttType);
				let json = {
				  'DstDeviceName':state.currentDevice.uuid,
				  'SrcDeviceName':options.clientId,
				  'type':type,
				}
				if(typeof other !== 'undefined'){
					json.other = other
				}
				let payload = JSON.stringify(json)
				console.log(`publishWithType topic = ${topic} payload = ${payload}`);
				return dispatch('publish',{topic,payload})

			}

		},

		//发布消息
		publish({ state, commit, getters }, message) {
			console.log('publish');
			console.log(message);
			return new Promise((resolve, reject) => {
				if (getters.connected) {
					//发布消息
					state.mqttClient.publish(message.topic, message.payload, (err) => {
						if (err) {
							console.log(err);
							reject(err)
						} else {
							resolve()
						}
					})
				} else {
					reject("mqttClient is not connected")
				}
			})
		},
		//订阅消息
		subscribe({ state, commit, getters }, topic) {
			console.log(`subscribe topic = ${topic}`);
			return new Promise((resolve, reject) => {
				if (getters.connected) {
					//发布消息
					state.mqttClient.subscribe(topic, (err) => {
						if (err) {
							console.log(err);
							reject(err)
						} else {
							resolve()
						}
					})
				} else {
					reject("mqttClient is not connected")
				}
			})
		},
		//解除订阅消息
		unsubscribe({ state, commit, getters }, topic){
			console.log(`unsubscribe topic = ${topic}`);
			return new Promise((resolve, reject) => {
				if (getters.connected) {
					//发布消息
					state.mqttClient.unsubscribe(topic, (err) => {
						if (err) {
							console.log(err);
							reject(err)
						} else {
							resolve()
						}
					})
				} else {
					reject("mqttClient is not connected")
				}
			})
		}

	}

}