rqian
40c6edda36
# Conflicts: # maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java # maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java
507 lines
21 KiB
Java
507 lines
21 KiB
Java
package com.maibu.mqtt;
|
||
|
||
import java.util.ArrayList;
|
||
import java.util.List;
|
||
|
||
import org.eclipse.paho.client.mqttv3.MqttException;
|
||
import com.maibu.core.business.dto.DeviceMapPushDTO;
|
||
import org.springframework.util.CollectionUtils;
|
||
|
||
import com.fasterxml.jackson.databind.JsonNode;
|
||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||
import com.maibu.constant.Constant;
|
||
import com.maibu.constant.NettyCacheKey;
|
||
import com.maibu.core.business.DeviceRunStatistics;
|
||
import com.maibu.core.business.device.DeviceStatusDetail;
|
||
import com.maibu.core.business.device.NettyDevice;
|
||
import com.maibu.core.business.dto.WebDeviceTaskStatusMessageDTO;
|
||
import com.maibu.core.business.dto.WebStatusMessageDTO;
|
||
import com.maibu.core.business.inter.WebsocketMesDispather;
|
||
import com.maibu.core.enums.DeviceTaskStaus;
|
||
import com.maibu.core.enums.MesType;
|
||
import com.maibu.core.host.HeartBeatDTO;
|
||
import com.maibu.core.host.HostLocationRelTimeDTO;
|
||
import com.maibu.core.host.HostNavigationReplyDTO;
|
||
import com.maibu.core.host.HostTaskStatusDTO;
|
||
import com.maibu.core.host.HostVehicleRelTimeDTO;
|
||
import com.maibu.core.host.LocationMessage;
|
||
import com.maibu.core.redis.RedisCache;
|
||
import com.maibu.enums.DeviceStatusEnum;
|
||
import com.maibu.memory.DeviceSessionManager;
|
||
import com.maibu.memory.GlobalMemory;
|
||
import com.maibu.memory.SiteMemory;
|
||
import com.maibu.utils.StringUtils;
|
||
import com.maibu.utils.json.JsonUtils;
|
||
import com.maibu.utils.spring.SpringUtils;
|
||
|
||
import lombok.extern.slf4j.Slf4j;
|
||
|
||
/**
|
||
* mqtt相关的 上位机处理器
|
||
*/
|
||
@Slf4j
|
||
public class HostMessageHandler implements CustomMqttMessageHandler {
|
||
|
||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||
|
||
private final DeviceSessionManager deviceSessionManager = SpringUtils.getBean(DeviceSessionManager.class);
|
||
|
||
private final WebsocketMesDispather websocketMesDispather = SpringUtils.getBean(WebsocketMesDispather.class);
|
||
|
||
private final String mqttClientId = "maibu-mqtt-client";
|
||
|
||
private final RedisCache redisCache = SpringUtils.getBean(RedisCache.class);
|
||
|
||
@Override
|
||
public String getHandlerType() {
|
||
return Constant.mqttHandleKey;
|
||
}
|
||
|
||
@Override
|
||
public boolean supports(String handlerType) {
|
||
return Constant.mqttHandleKey.equals(handlerType);
|
||
}
|
||
|
||
/**
|
||
* 处理 MQTT 消息
|
||
*
|
||
* <p>
|
||
* 通配符订阅场景下，deviceKey 为配置中的固定标识（如 "host-monitor"），
|
||
* 真正的设备 ID 需从 topic 中解析，通过 {@link #extractDeviceId(String)} 获取。
|
||
* </p>
|
||
*/
|
||
@Override
|
||
public void handleMessage(String deviceKey, String topic, String payload, CustomMqttDeviceConfig config) {
|
||
try {
|
||
String deviceId = extractDeviceId(topic);
|
||
if (deviceId == null) {
|
||
log.warn("无法从topic解析deviceId: {}", topic);
|
||
return;
|
||
}
|
||
|
||
String action = extractAction(topic);
|
||
if (action == null) {
|
||
log.debug("未识别的topic: {}", topic);
|
||
return;
|
||
}
|
||
|
||
switch (action) {
|
||
case "location_post":
|
||
handleLocationPost(deviceId, topic, payload);
|
||
break;
|
||
case "realtime_post":
|
||
handleRealtimePost(deviceId, topic, payload);
|
||
break;
|
||
case "heartbeat_post":
|
||
handleHeartbeatPost(deviceId, topic, payload);
|
||
break;
|
||
case "error_post":
|
||
handleErrorPost(deviceId, topic, payload);
|
||
break;
|
||
// case "route_post":
|
||
// case "point_post":
|
||
// handleTaskPost(deviceId, topic, payload, action);
|
||
// break;
|
||
case "task_status_post":
|
||
handleTaskPost(deviceId, topic, payload, action);
|
||
break;
|
||
case "config_reply":
|
||
handleConfigReply(deviceId, topic, payload);
|
||
break;
|
||
case "control_reply":
|
||
handleControlReply(deviceId, topic, payload);
|
||
break;
|
||
case "route_reply":
|
||
handleRouteReply(deviceId, topic, payload);
|
||
break;
|
||
default:
|
||
log.debug("割草机消息: deviceId={}, topic={}, action={}", deviceId, topic, action);
|
||
break;
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("消息处理异常: topic={}, error={}", topic, e.getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 定位信息 经纬度
|
||
*/
|
||
private void handleLocationPost(String deviceId, String topic, String payload) {
|
||
try {
|
||
log.debug("【位置数据】deviceId={}, payload={}", deviceId, payload);
|
||
HostLocationRelTimeDTO locationRelTimeDTO = JsonUtils.parseObject(payload, HostLocationRelTimeDTO.class);
|
||
if (locationRelTimeDTO != null) {
|
||
WebStatusMessageDTO dto = createWebDeviceLocationMessage(locationRelTimeDTO, deviceId);
|
||
List<NettyDevice> controlMasters = deviceSessionManager
|
||
.getAllSlaveControl(deviceId);
|
||
if (!CollectionUtils.isEmpty(controlMasters)) {
|
||
for (NettyDevice x : controlMasters) {
|
||
try {
|
||
websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(dto));
|
||
} catch (Exception e) {
|
||
log.error("位置数据websocket推送失败: connectorId={}, error={}", x.getConnectorId(),
|
||
e.getMessage());
|
||
}
|
||
}
|
||
}
|
||
// todo 推送实时位置和状态消息到外部
|
||
// todo 记录到内存 开线程池队列 逐一转发
|
||
// GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId,
|
||
// String.format(MqttTopic.DEVICE_LOCATION_TOPIC, deviceId),
|
||
// JsonUtils.toJsonString(dto));
|
||
// todo 保存到内存
|
||
String key = NettyCacheKey.latestPositonKey + deviceId;
|
||
LocationMessage locationMessage = locationRelTimeDTO.getData();
|
||
locationMessage.setDeviceId(deviceId);
|
||
redisCache.setCacheObject(key, locationMessage);
|
||
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("位置消息解析失败: deviceId={}, payload={}, error={}", deviceId, payload, e);
|
||
}
|
||
}
|
||
|
||
public WebStatusMessageDTO createWebDeviceLocationMessage(HostLocationRelTimeDTO locationRelTimeDTO,
|
||
String deviceId) {
|
||
WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO();
|
||
List<DeviceStatusDetail> transferData = new ArrayList<>();
|
||
DeviceStatusDetail lat = new DeviceStatusDetail();
|
||
lat.setName("latitude");
|
||
lat.setValue(String.valueOf(locationRelTimeDTO.getData().getLatitude()));
|
||
lat.setUnit("");
|
||
|
||
transferData.add(lat);
|
||
|
||
DeviceStatusDetail lng = new DeviceStatusDetail();
|
||
lng.setName("longitude");
|
||
lng.setValue(String.valueOf(locationRelTimeDTO.getData().getLongitude()));
|
||
lng.setUnit("");
|
||
transferData.add(lng);
|
||
|
||
DeviceStatusDetail yaw = new DeviceStatusDetail();
|
||
yaw.setName("yaw");
|
||
yaw.setValue(String.valueOf(locationRelTimeDTO.getData().getYaw()));
|
||
yaw.setUnit("");
|
||
transferData.add(yaw);
|
||
|
||
DeviceStatusDetail fix_status = new DeviceStatusDetail();
|
||
fix_status.setName("fix_status");
|
||
fix_status.setValue(String.valueOf(locationRelTimeDTO.getData().getFix_status()));
|
||
fix_status.setUnit("");
|
||
transferData.add(fix_status);
|
||
|
||
DeviceStatusDetail altitude = new DeviceStatusDetail();
|
||
altitude.setName("altitude");
|
||
altitude.setValue(String.valueOf(locationRelTimeDTO.getData().getAltitude()));
|
||
altitude.setUnit("");
|
||
transferData.add(altitude);
|
||
|
||
DeviceStatusDetail status = new DeviceStatusDetail();
|
||
status.setName("status");
|
||
|
||
NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
// todo error故障
|
||
if (nettyDevice != null) {
|
||
if (nettyDevice.getOnlineStatus() == 0) {
|
||
status.setValue(DeviceStatusEnum.offline.getCode());
|
||
} else {
|
||
if (nettyDevice.getTask() != null && nettyDevice.getTask().isTaskAccepted()) {
|
||
status.setValue(DeviceStatusEnum.working.getCode());
|
||
} else {
|
||
status.setValue(DeviceStatusEnum.idle.getCode());
|
||
}
|
||
}
|
||
} else {
|
||
status.setValue(DeviceStatusEnum.offline.getCode());
|
||
}
|
||
status.setUnit("");
|
||
transferData.add(status);
|
||
|
||
webStatusMessageDTO.setData(transferData);
|
||
webStatusMessageDTO.setType(MesType.locationInfo);
|
||
|
||
if (nettyDevice != null && nettyDevice.getDevice() != null) {
|
||
Long siteId = nettyDevice.getDevice().getSiteId();
|
||
Long orgId = nettyDevice.getDevice().getOrgId();
|
||
SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
|
||
if (siteMemory != null) {
|
||
DeviceMapPushDTO dto = new DeviceMapPushDTO();
|
||
dto.setA(Double.parseDouble(altitude.getValue()));
|
||
dto.setId(deviceId);
|
||
// dto.setB();
|
||
// dto.setHa();
|
||
dto.setF(fix_status.getValue());
|
||
dto.setS(status.getValue());
|
||
dto.setLat(Double.parseDouble(lat.getValue()));
|
||
dto.setLng(Double.parseDouble(lng.getValue()));
|
||
dto.setT(nettyDevice.getCurrentTaskId());
|
||
dto.setY(Double.parseDouble(yaw.getValue()));
|
||
siteMemory.saveMaPushDTO(deviceId, dto);
|
||
}
|
||
}
|
||
|
||
return webStatusMessageDTO;
|
||
}
|
||
|
||
public WebStatusMessageDTO createWebDeviceRealtimeMessage(HostVehicleRelTimeDTO vehicleRelTimeDTO) {
|
||
// todo 状态消息
|
||
|
||
WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO();
|
||
List<DeviceStatusDetail> transferData = new ArrayList<>();
|
||
DeviceStatusDetail voltage = new DeviceStatusDetail();
|
||
voltage.setName("voltage");
|
||
voltage.setValue(String.valueOf(vehicleRelTimeDTO.getData().getBattery_voltage()));
|
||
voltage.setUnit("");
|
||
|
||
transferData.add(voltage);
|
||
|
||
DeviceStatusDetail speed = new DeviceStatusDetail();
|
||
speed.setName("speed");
|
||
speed.setValue(String.valueOf(vehicleRelTimeDTO.getData().getSpeed()));
|
||
speed.setUnit("");
|
||
transferData.add(speed);
|
||
|
||
DeviceStatusDetail motion = new DeviceStatusDetail();
|
||
motion.setName("motion");
|
||
motion.setValue(String.valueOf(vehicleRelTimeDTO.getData().getMotion()));
|
||
motion.setUnit("");
|
||
transferData.add(motion);
|
||
|
||
DeviceStatusDetail percentage = new DeviceStatusDetail();
|
||
percentage.setName("percentage");
|
||
percentage.setValue(String.valueOf(vehicleRelTimeDTO.getData().getBattery_percentage()));
|
||
percentage.setUnit("");
|
||
transferData.add(percentage);
|
||
|
||
webStatusMessageDTO.setData(transferData);
|
||
webStatusMessageDTO.setType(MesType.deviceInfo);
|
||
return webStatusMessageDTO;
|
||
}
|
||
|
||
/**
|
||
* 上位机数据
|
||
*/
|
||
private void handleRealtimePost(String deviceId, String topic, String payload) {
|
||
log.debug("【实时数据】deviceId={}, payload={}", deviceId, payload);
|
||
try {
|
||
HostVehicleRelTimeDTO vehicleRelTimeDTO = JsonUtils.parseObject(payload, HostVehicleRelTimeDTO.class);
|
||
if (vehicleRelTimeDTO != null) {
|
||
|
||
// todo 更新里程数 里程数 * 割幅0.8
|
||
vehicleRelTimeDTO.getData().getDistance();
|
||
NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
if (nettyDevice != null) {
|
||
Long orgId = nettyDevice.getDevice().getOrgId();
|
||
Long siteId = nettyDevice.getDevice().getSiteId();
|
||
SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
|
||
if (siteMemory != null) {
|
||
DeviceRunStatistics deviceRunStatistics = siteMemory.getRunStatistics(deviceId);
|
||
if (deviceRunStatistics == null) {
|
||
deviceRunStatistics = new DeviceRunStatistics();
|
||
}
|
||
deviceRunStatistics.setDeviceId(deviceId);
|
||
deviceRunStatistics.setDistance(vehicleRelTimeDTO.getData().getDistance());
|
||
deviceRunStatistics.setWorkArea(vehicleRelTimeDTO.getData().getDistance() * 0.8);
|
||
siteMemory.saveRunStatistics(deviceRunStatistics);
|
||
}
|
||
}
|
||
|
||
WebStatusMessageDTO dto = createWebDeviceRealtimeMessage(vehicleRelTimeDTO);
|
||
List<NettyDevice> controlMasters = deviceSessionManager
|
||
.getAllSlaveControl(deviceId);
|
||
if (!CollectionUtils.isEmpty(controlMasters)) {
|
||
for (NettyDevice x : controlMasters) {
|
||
try {
|
||
websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(dto));
|
||
} catch (Exception e) {
|
||
log.error("实时数据websocket推送失败: connectorId={}, error={}", x.getConnectorId(),
|
||
e.getMessage());
|
||
}
|
||
}
|
||
}
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("实时消息解析失败: deviceId={}, payload={}, error={}", deviceId, payload, e.getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理心跳
|
||
* topic: mower/{deviceId}/event/heartbeat/post
|
||
*/
|
||
private void handleHeartbeatPost(String deviceId, String topic, String payload) {
|
||
log.debug("【心跳】payload={}", payload);
|
||
// 心跳处理
|
||
HeartBeatDTO beatDTO = JsonUtils.parseObject(payload, HeartBeatDTO.class);
|
||
if (beatDTO != null) {
|
||
beatDTO.setSysTimestamp(System.currentTimeMillis());
|
||
GlobalMemory.customMqttDeviceMonitor.saveHeartBeat(beatDTO.getSn(), beatDTO);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理错误告警
|
||
* topic: mower/{deviceId}/event/error/post
|
||
*/
|
||
private void handleErrorPost(String deviceId, String topic, String payload) {
|
||
log.warn("【错误告警】deviceId={}, payload={}", deviceId, payload);
|
||
}
|
||
|
||
/**
|
||
* 处理任务相关上报
|
||
*/
|
||
private void handleTaskPost(String deviceId, String topic, String payload, String action) {
|
||
log.info("【任务上报】deviceId={}, action={}, payload={}", deviceId, action, payload);
|
||
|
||
HostTaskStatusDTO dto = JsonUtils.parseObject(payload, HostTaskStatusDTO.class);
|
||
if (dto != null && !StringUtils.isEmpty(dto.getStatus())) {
|
||
NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
if ("completed".equals(dto.getStatus())) {
|
||
Long taskId = nettyDevice.finishTask();
|
||
|
||
WebDeviceTaskStatusMessageDTO sendDto = new WebDeviceTaskStatusMessageDTO();
|
||
sendDto.setDeviceId(deviceId);
|
||
sendDto.setTaskId(taskId);
|
||
sendDto.setStatus(DeviceTaskStaus.FINISH.toString());
|
||
sendDto.setType(MesType.device_task_change);
|
||
|
||
List<NettyDevice> controlMasters = deviceSessionManager
|
||
.getAllSlaveControl(deviceId);
|
||
if (!CollectionUtils.isEmpty(controlMasters)) {
|
||
for (NettyDevice x : controlMasters) {
|
||
try {
|
||
websocketMesDispather.dispather(x.getConnectorId(), JsonUtils.toJsonString(sendDto));
|
||
} catch (Exception e) {
|
||
log.error("实时数据websocket推送失败: connectorId={}, error={}", x.getConnectorId(),
|
||
e.getMessage());
|
||
}
|
||
}
|
||
}
|
||
//todo 推送实时位置和状态消息到外部
|
||
try {
|
||
GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_STATUS_TOPIC, taskId),
|
||
JsonUtils.toJsonString(sendDto));
|
||
} catch (MqttException e) {
|
||
// TODO Auto-generated catch block
|
||
e.printStackTrace();
|
||
}
|
||
|
||
}
|
||
if ("cancelled".equals(dto.getStatus())) {
|
||
nettyDevice.canceledTask();
|
||
}
|
||
// if("paused".equals(dto.getStatus())){
|
||
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
// nettyDevice.finishTask();
|
||
// }
|
||
if ("failed".equals(dto.getStatus())) {
|
||
nettyDevice.failedTask();
|
||
}
|
||
// if("accepted".equals(dto.getStatus())){
|
||
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
// nettyDevice.finishTask();
|
||
// }
|
||
// if("running".equals(dto.getStatus())){
|
||
// NettyDevice nettyDevice = deviceSessionManager.getDevice(deviceId);
|
||
// nettyDevice.finishTask();
|
||
// }
|
||
}
|
||
|
||
}
|
||
|
||
/**
|
||
* 处理配置响应
|
||
*/
|
||
private void handleConfigReply(String deviceId, String topic, String payload) {
|
||
log.debug("【配置响应】deviceId={}, payload={}", deviceId, payload);
|
||
}
|
||
|
||
/**
|
||
* 处理控制指令响应
|
||
*/
|
||
private void handleControlReply(String deviceId, String topic, String payload) {
|
||
log.debug("【控制响应】deviceId={}, payload={}", deviceId, payload);
|
||
}
|
||
|
||
/**
|
||
* 处理控制指令响应
|
||
*/
|
||
private void handleRouteReply(String deviceId, String topic, String payload) {
|
||
log.debug("【路径任务下发响应】deviceId={}, payload={}", deviceId, payload);
|
||
|
||
// todo 更新任务接收状态
|
||
HostNavigationReplyDTO replyDTO = JsonUtils.parseObject(payload, HostNavigationReplyDTO.class);
|
||
NettyDevice device = deviceSessionManager.getDevice(deviceId);
|
||
if (replyDTO.isAccepted()) {
|
||
if (device != null && device.getTask() != null
|
||
&& String.valueOf(device.getTask().getId()).equals(replyDTO.getTask_id())) {
|
||
device.getTask().setTaskAccepted(true);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 从 Topic 中解析 deviceId
|
||
* Topic 格式：mower/{deviceId}/...
|
||
*/
|
||
private String extractDeviceId(String topic) {
|
||
if (topic == null || topic.isEmpty()) {
|
||
return null;
|
||
}
|
||
String[] parts = topic.split("/");
|
||
if (parts.length < 2 || !"mower".equals(parts[0])) {
|
||
return null;
|
||
}
|
||
return parts[1];
|
||
}
|
||
|
||
/**
|
||
* 从 Topic 中提取动作标识
|
||
* 将 mower/{deviceId}/xxx/yyy/post 转为 xxx_yyy_post 形式
|
||
*/
|
||
private String extractAction(String topic) {
|
||
String[] parts = topic.split("/");
|
||
if (parts.length < 3 || !"mower".equals(parts[0])) {
|
||
return null;
|
||
}
|
||
StringBuilder path = new StringBuilder();
|
||
for (int i = 2; i < parts.length; i++) {
|
||
if (i > 2)
|
||
path.append("/");
|
||
path.append(parts[i]);
|
||
}
|
||
String remaining = path.toString();
|
||
|
||
if (remaining.equals("property/location/post"))
|
||
return "location_post";
|
||
if (remaining.equals("property/realtime/post"))
|
||
return "realtime_post";
|
||
if (remaining.equals("event/heartbeat/post"))
|
||
return "heartbeat_post";
|
||
if (remaining.equals("event/error/post"))
|
||
return "error_post";
|
||
// if (remaining.equals("task/route/post")) return "route_post";
|
||
// if (remaining.equals("task/point/post")) return "point_post";
|
||
if (remaining.equals("task/status/post"))
|
||
return "task_status_post";
|
||
if (remaining.equals("action/config/reply"))
|
||
return "config_reply";
|
||
if (remaining.equals("action/control/reply"))
|
||
return "control_reply";
|
||
if (remaining.equals("task/target/reply"))
|
||
return "route_reply";
|
||
|
||
return null;
|
||
}
|
||
|
||
@Override
|
||
public void onConnectionLost(String deviceKey, Throwable cause) {
|
||
log.warn("主机监控连接断开: deviceKey={}, reason={}", deviceKey, cause.getMessage());
|
||
}
|
||
|
||
@Override
|
||
public void onSubscribed(String deviceKey, String topic) {
|
||
log.info("主机监控订阅主题: deviceKey={}, topic={}", deviceKey, topic);
|
||
}
|
||
}
|