|
|
@@ -1,275 +1,42 @@
|
|
|
package com.care.bms.mqtt;
|
|
|
|
|
|
|
|
|
-import cn.hutool.core.date.DateUtil;
|
|
|
-
|
|
|
-import com.alibaba.fastjson.JSON;
|
|
|
-
|
|
|
-import java.util.Date;
|
|
|
-import java.util.Map;
|
|
|
-
|
|
|
-import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
|
|
|
-import com.care.bms.service.BmsEventOrderService;
|
|
|
-import com.care.common.entity.CareDevice;
|
|
|
-import com.care.common.entity.CareEventOrder;
|
|
|
-import com.care.common.entity.CareHouse;
|
|
|
-import com.care.common.entity.CareMqttMsg;
|
|
|
-import com.care.common.enums.OrderStatusEnum;
|
|
|
-import com.care.common.enums.OrderTypeEnum;
|
|
|
-import com.care.common.service.*;
|
|
|
-import com.care.common.util.CommonConfUtil;
|
|
|
+import com.care.bms.service.MqttMsgService;
|
|
|
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
|
|
|
import org.eclipse.paho.client.mqttv3.MqttMessage;
|
|
|
|
|
|
+import java.util.Map;
|
|
|
+
|
|
|
|
|
|
/**
|
|
|
* 订阅信息监听类
|
|
|
*/
|
|
|
public class MqttMessageListener implements IMqttMessageListener {
|
|
|
- private Map<String,Thread> threadMap;
|
|
|
- private CareDeviceService careDeviceService;
|
|
|
- private BmsEventOrderService bmsEventOrderService;
|
|
|
- private CareEventOrderService careEventOrderService;
|
|
|
- private CareHouseService careHouseService;
|
|
|
- private CareOrganizationService careOrganizationService;
|
|
|
- private CareStationService careStationService;
|
|
|
- private CareMqttMsgService careMqttMsgService;
|
|
|
+ private MqttMsgService mqttMsgService;
|
|
|
|
|
|
- public Map<String, Thread> getThreadMap() {
|
|
|
- return threadMap;
|
|
|
+ private Map<String,Thread> threadMap;
|
|
|
+ public void setMqttMsgService(MqttMsgService mqttMsgService) {
|
|
|
+ this.mqttMsgService = mqttMsgService;
|
|
|
}
|
|
|
|
|
|
public void setThreadMap(Map<String, Thread> threadMap) {
|
|
|
this.threadMap = threadMap;
|
|
|
}
|
|
|
|
|
|
- public CareMqttMsgService getCareMqttMsgService() {
|
|
|
- return careMqttMsgService;
|
|
|
- }
|
|
|
-
|
|
|
- public void setCareMqttMsgService(CareMqttMsgService careMqttMsgService) {
|
|
|
- this.careMqttMsgService = careMqttMsgService;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- public CareOrganizationService getCareOrganizationService() {
|
|
|
- return careOrganizationService;
|
|
|
- }
|
|
|
-
|
|
|
- public void setCareOrganizationService(CareOrganizationService careOrganizationService) {
|
|
|
- this.careOrganizationService = careOrganizationService;
|
|
|
- }
|
|
|
-
|
|
|
- public CareStationService getCareStationService() {
|
|
|
- return careStationService;
|
|
|
- }
|
|
|
-
|
|
|
- public void setCareStationService(CareStationService careStationService) {
|
|
|
- this.careStationService = careStationService;
|
|
|
- }
|
|
|
-
|
|
|
- public CareHouseService getCareHouseService() {
|
|
|
- return careHouseService;
|
|
|
- }
|
|
|
-
|
|
|
- public void setCareHouseService(CareHouseService careHouseService) {
|
|
|
- this.careHouseService = careHouseService;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- public CareEventOrderService getCareEventOrderService() {
|
|
|
- return careEventOrderService;
|
|
|
- }
|
|
|
-
|
|
|
- public void setCareEventOrderService(CareEventOrderService careEventOrderService) {
|
|
|
- this.careEventOrderService = careEventOrderService;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- public BmsEventOrderService getBmsEventOrderService() {
|
|
|
- return bmsEventOrderService;
|
|
|
- }
|
|
|
-
|
|
|
- public void setBmsEventOrderService(BmsEventOrderService bmsEventOrderService) {
|
|
|
- this.bmsEventOrderService = bmsEventOrderService;
|
|
|
- }
|
|
|
-
|
|
|
- public CareDeviceService getCareDeviceService() {
|
|
|
- return careDeviceService;
|
|
|
- }
|
|
|
-
|
|
|
- public void setCareDeviceService(CareDeviceService careDeviceService) {
|
|
|
- this.careDeviceService = careDeviceService;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
@Override
|
|
|
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
|
|
|
- try {
|
|
|
// System.out.println(mqttMessage.getId());
|
|
|
// System.out.println(mqttMessage.getPayload());
|
|
|
// System.out.println(mqttMessage.getQos());
|
|
|
// System.out.println(mqttMessage.isRetained());
|
|
|
// System.out.println(mqttMessage.isDuplicate());
|
|
|
-
|
|
|
- String mqttMessageStr = mqttMessage.toString();
|
|
|
- System.out.println("mqttMessageStr == " + mqttMessageStr);
|
|
|
- if(mqttMessageStr == null || !mqttMessageStr.startsWith("{")){
|
|
|
- return;
|
|
|
- }
|
|
|
-
|
|
|
- QueryWrapper<CareDevice> queryWrapper = new QueryWrapper<>();
|
|
|
- queryWrapper.lambda().eq(CareDevice::getTopic,topic);
|
|
|
- CareDevice careDevice = careDeviceService.getOne(queryWrapper);
|
|
|
- if(careDevice != null) {
|
|
|
- CareMqttMsg careMqttMsg = new CareMqttMsg();
|
|
|
- careMqttMsg.setMqttMessage(mqttMessageStr);
|
|
|
- careMqttMsg.setTopic(topic);
|
|
|
- careMqttMsg.setDevId(careDevice.getId());
|
|
|
- careMqttMsg.setCreateTime(new Date());
|
|
|
- this.careMqttMsgService.save(careMqttMsg);
|
|
|
- } else {
|
|
|
- System.out.println("数据库中没找到对应设备");
|
|
|
- return;
|
|
|
- }
|
|
|
-
|
|
|
- Map map = JSON.parseObject(mqttMessageStr);
|
|
|
- String type = (String) map.get("type");
|
|
|
- Map msgMap = (Map)map.get("msg");
|
|
|
- if ("BreathHeartRate".equals(type)) {
|
|
|
- Map msg = (Map) map.get("msg");
|
|
|
- Integer breath = (Integer) msg.get("breath");
|
|
|
- Integer heart = (Integer) msg.get("heart");
|
|
|
- careDevice.setBreathRate(breath);
|
|
|
- careDevice.setHeartRate(heart);
|
|
|
- careDeviceService.updateById(careDevice);
|
|
|
- } else if ("FallDown".equals(type)) { //跌到
|
|
|
- Integer conf = (Integer)msgMap.get("conf");
|
|
|
- if(conf == 255){ //取消
|
|
|
- threadMap.forEach((key,value) -> {
|
|
|
- if(key.startsWith("FallDown")) {
|
|
|
- if(!value.isInterrupted()){
|
|
|
- value.interrupt();//停止线程
|
|
|
- threadMap.remove(key);
|
|
|
- value = null;
|
|
|
- }
|
|
|
- }
|
|
|
- });
|
|
|
- } else {
|
|
|
- Integer fallResponseTime = careDevice.getFallResponseTime();
|
|
|
- MyThread myThread = new MyThread();
|
|
|
- myThread.setCareDevice(careDevice);
|
|
|
- myThread.setOrderType(OrderTypeEnum.DI_EDAO.getValue());
|
|
|
- myThread.setSleepTime(fallResponseTime);
|
|
|
- myThread.setDate(DateUtil.date());
|
|
|
- threadMap.put("FallDown-" + System.currentTimeMillis(),myThread);
|
|
|
- myThread.start();
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- } else if ("StayTooLong".equals(type)) { //久滞
|
|
|
- Integer conf = (Integer)msgMap.get("conf");
|
|
|
- if(conf == 255){ //取消
|
|
|
- threadMap.forEach((key,value) -> {
|
|
|
- if(key.startsWith("StayTooLong")) {
|
|
|
- if(!value.isInterrupted()){
|
|
|
- value.interrupt();//停止线程
|
|
|
- threadMap.remove(key);
|
|
|
- value = null;
|
|
|
- }
|
|
|
- }
|
|
|
- });
|
|
|
- } else {
|
|
|
- Integer longlagResponseTime = careDevice.getLonglagResponseTime();
|
|
|
- MyThread myThread = new MyThread();
|
|
|
- myThread.setCareDevice(careDevice);
|
|
|
- myThread.setOrderType(OrderTypeEnum.JIU_ZHI.getValue());
|
|
|
- myThread.setSleepTime(longlagResponseTime);
|
|
|
- myThread.setDate(DateUtil.date());
|
|
|
- threadMap.put("StayTooLong-" + System.currentTimeMillis(),myThread);
|
|
|
- myThread.start();
|
|
|
- }
|
|
|
-
|
|
|
- }
|
|
|
- } catch (Exception e){
|
|
|
- e.printStackTrace();
|
|
|
+ String mqttMessageStr = mqttMessage.toString();
|
|
|
+ System.out.println("mqttMessageStr == " + mqttMessageStr);
|
|
|
+ if(mqttMessageStr == null || !mqttMessageStr.startsWith("{")){
|
|
|
+ return;
|
|
|
}
|
|
|
- System.out.println(topic);
|
|
|
- }
|
|
|
-
|
|
|
- private void saveOrder(CareDevice careDevice,String orderType,Date date){
|
|
|
- CareHouse careHouse = careHouseService.getById(careDevice.getHouseId());
|
|
|
-
|
|
|
- CareEventOrder order = new CareEventOrder();
|
|
|
- order.setOrgId(careHouse.getOrgId());
|
|
|
- order.setOrgName(careOrganizationService.getById(careHouse.getOrgId()).getName());
|
|
|
- order.setStationId(careHouse.getStationId());
|
|
|
- order.setStationName(careStationService.getById(careHouse.getStationId()).getName());
|
|
|
- order.setHouseId(careHouse.getId());
|
|
|
- order.setDevId(careDevice.getId());
|
|
|
- order.setDevName(careDevice.getDevName());
|
|
|
- order.setHouseName(careHouse.getName());
|
|
|
- order.setMemberId(careDevice.getMemberId());
|
|
|
- order.setTitle(careHouse.getAddr());
|
|
|
- order.setOrderType(orderType);
|
|
|
- order.setStatus(OrderStatusEnum.TODO.getValue());
|
|
|
- order.setCreateTime(date);
|
|
|
- order.setModifyTime(date);
|
|
|
- careEventOrderService.saveOrder(order);
|
|
|
-
|
|
|
+ mqttMsgService.handleMessage(threadMap,topic,mqttMessageStr);
|
|
|
}
|
|
|
|
|
|
- class MyThread extends Thread {
|
|
|
- private CareDevice careDevice;
|
|
|
- private String orderType;
|
|
|
- private Integer sleepTime; //分钟
|
|
|
- private Date date;
|
|
|
-
|
|
|
- public Date getDate() {
|
|
|
- return date;
|
|
|
- }
|
|
|
-
|
|
|
- public void setDate(Date date) {
|
|
|
- this.date = date;
|
|
|
- }
|
|
|
-
|
|
|
- public CareDevice getCareDevice() {
|
|
|
- return careDevice;
|
|
|
- }
|
|
|
-
|
|
|
- public void setCareDevice(CareDevice careDevice) {
|
|
|
- this.careDevice = careDevice;
|
|
|
- }
|
|
|
-
|
|
|
- public String getOrderType() {
|
|
|
- return orderType;
|
|
|
- }
|
|
|
-
|
|
|
- public void setOrderType(String orderType) {
|
|
|
- this.orderType = orderType;
|
|
|
- }
|
|
|
-
|
|
|
- public Integer getSleepTime() {
|
|
|
- return sleepTime;
|
|
|
- }
|
|
|
-
|
|
|
- public void setSleepTime(Integer sleepTime) {
|
|
|
- this.sleepTime = sleepTime;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- @Override
|
|
|
- public void run(){
|
|
|
- try {
|
|
|
- if(sleepTime != null && sleepTime > 0){
|
|
|
- Thread.sleep(sleepTime * 60 * 1000);
|
|
|
- }
|
|
|
- saveOrder(careDevice,orderType,date);
|
|
|
- } catch (Exception e){
|
|
|
- e.printStackTrace();
|
|
|
- }
|
|
|
-
|
|
|
- }
|
|
|
- }
|
|
|
|
|
|
}
|
suntianwu