forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/component/mqtt/handler/ServicesReplyHandler.java
@@ -4,12 +4,15 @@ import com.ard.sample.component.mqtt.model.ChannelName; import com.ard.sample.component.mqtt.model.CommonTopicReceiver; import com.ard.sample.component.mqtt.model.ServiceReply; import com.ard.sample.component.websocket.enums.MessageType; import com.ard.sample.component.websocket.handler.DockWebSocketHandler; import com.ard.sample.manage.model.enums.LogsFileMethodEnum; import com.ard.sample.manage.model.receiver.LogsFileUploadList; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.mqtt.support.MqttHeaders; import org.springframework.messaging.Message; import org.springframework.stereotype.Component; @@ -26,6 +29,9 @@ @Autowired private ObjectMapper mapper; @Autowired private DockWebSocketHandler _dockWebSocketHandler; /** * Handle the reply message from the pilot side to the on-demand video. * @param message reply message @@ -33,9 +39,17 @@ */ @ServiceActivator(inputChannel = ChannelName.INBOUND_SERVICE_REPLY) public void serviceReply(Message<?> message) throws IOException { String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString(); // thing/product/7CTDLCE00A1UPE/services_reply String[] parts = topic.split("/"); // 目标子串是分割后数组的第3个元素（索引2） String deviceSn = parts.length > 2 ? parts[2] : ""; // 加判断避免数组越界 byte[] payload = (byte[])message.getPayload(); CommonTopicReceiver receiver = mapper.readValue(payload, new TypeReference<CommonTopicReceiver>() {}); receiver.setGateway(deviceSn); receiver.setType(MessageType.CONTROL.getType()); if (LogsFileMethodEnum.FILE_UPLOAD_LIST.getMethod().equals(receiver.getMethod())) { LogsFileUploadList list = mapper.convertValue(receiver.getData(), new TypeReference<LogsFileUploadList>() {}); receiver.setData(list); @@ -46,5 +60,12 @@ Chan<CommonTopicReceiver<?>> chan = Chan.getInstance(); // Put the message to the chan object. chan.put(receiver); // WebSocket 接口，发给前端 String dockControl=""; dockControl=mapper.writeValueAsString(receiver); _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl); // DockControlProgressWebSockerHandler. } } ard-sample/src/main/java/com/ard/sample/component/mqtt/model/CommonTopicReceiver.java
@@ -31,4 +31,6 @@ private Integer needReply; private String type; } ard-sample/src/main/java/com/ard/sample/component/mqtt/model/OutputProgressReceiver.java
@@ -1,5 +1,6 @@ package com.ard.sample.component.mqtt.model; import com.fasterxml.jackson.annotation.JsonProperty; import lombok.Data; /** @@ -10,9 +11,18 @@ @Data public class OutputProgressReceiver { @JsonProperty("percent") private Integer percent; @JsonProperty("current_step") private String currentStep; @JsonProperty("step_key") private String stepKey; @JsonProperty("step_result") private Integer stepResult; @JsonProperty("total_steps") private String totalSteps; } ard-sample/src/main/java/com/ard/sample/manage/service/impl/ControlServiceImpl.java
@@ -6,6 +6,8 @@ import com.ard.sample.component.mqtt.service.IMessageSenderService; import com.ard.common.core.redis.RedisConst; import com.ard.common.core.redis.RedisOpsUtils; import com.ard.sample.component.websocket.enums.MessageType; import com.ard.sample.component.websocket.handler.DockWebSocketHandler; import com.ard.sample.component.websocket.service.ISendMessageService; import com.ard.sample.manage.model.dto.BasicDeviceProperty; import com.ard.sample.manage.model.dto.DeviceDTO; @@ -37,7 +39,13 @@ private IMessageSenderService messageSenderService; @Autowired private DockWebSocketHandler _dockWebSocketHandler; @Autowired private ISendMessageService webSocketMessageService; @Autowired private IDeviceService _deviceService; // @Autowired // private IWebSocketManageService webSocketManageService; @@ -105,12 +113,19 @@ if (RedisOpsUtils.getExpire(key) <= 0) { return; } // 更新设备列表状态--关闭舱盖中... String method = receiver.getMethod(); String deviceSn = receiver.getGateway(); if(method.equals("cover_close")){ _deviceService.updateWorkType(deviceSn, 2L); } String sn = RedisOpsUtils.get(key).toString(); receiver.setType(MessageType.PROGRESS.getType()); EventsReceiver<EventsOutputReceiver> eventsReceiver = mapper.convertValue(receiver.getData(), new TypeReference<EventsReceiver<EventsOutputReceiver>>(){}); eventsReceiver.setBid(receiver.getBid()); eventsReceiver.setSn(sn); eventsReceiver.setType(MessageType.PROGRESS.getType()); log.info("SN: {}, {} ===> Control progress: {}", sn, receiver.getMethod(), eventsReceiver.getOutput().getProgress().toString()); @@ -133,6 +148,20 @@ // .timestamp(System.currentTimeMillis()) // .bizCode(receiver.getMethod()) // .build()); // 更新设备列表状态--飞行器开机中... if(method.equals("drone_open")){ _deviceService.updateWorkType(device.getChildDeviceSn(), 7L); } String dockProgress =""; try { dockProgress = mapper.writeValueAsString(eventsReceiver); } catch (Exception e) { e.printStackTrace(); } // 设备控制进度消息websocket发给前端 _dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.PROGRESS.getType(), dockProgress); if (receiver.getNeedReply() != null && receiver.getNeedReply() == 1) { String topic = headers.get(MqttHeaders.RECEIVED_TOPIC) + TopicConst._REPLY_SUF; ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOSDServiceImpl.java
@@ -2,15 +2,17 @@ import com.ard.sample.component.mqtt.model.CommonTopicReceiver; import com.ard.sample.component.websocket.config.ConcurrentWebSocketSession; import com.ard.sample.component.websocket.enums.MessageType; import com.ard.sample.component.websocket.handler.DockWebSocketHandler; import com.ard.sample.component.websocket.model.BizCodeEnum; import com.ard.sample.component.websocket.model.CustomWebSocketMessage; import com.ard.sample.component.websocket.service.impl.WebSocketDockServer; import com.ard.sample.manage.model.dto.DeviceDTO; import com.ard.sample.manage.model.dto.TelemetryDTO; import com.ard.sample.manage.model.enums.DeviceDomainEnum; import com.ard.sample.manage.model.receiver.OsdDockReceiver; import com.ard.sample.manage.service.IDeviceService; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ObjectNode; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -25,7 +27,7 @@ public class DockOSDServiceImpl extends AbstractTSAService { @Autowired WebSocketDockServer _websocketdockserver;// = new WebSocketDockServer(); private DockWebSocketHandler _dockWebSocketHandler; @Autowired private IDeviceService _deviceService; @@ -50,32 +52,45 @@ if (DeviceDomainEnum.DOCK.getVal() == device.getDomain()) { wsMessage.setBizCode(BizCodeEnum.DOCK_OSD.getCode()); OsdDockReceiver data = mapper.convertValue(receiver.getData(), OsdDockReceiver.class); receiver.setType(MessageType.STATUS.getType()); String dockOsd=""; String dockOsdData=""; try{ dockOsd=mapper.writeValueAsString(receiver.getData()); if (dockOsd.contains("flighttask_step_code")) { JsonNode jsonNode = mapper.readTree(dockOsd); dockOsd=mapper.writeValueAsString(receiver); dockOsdData=mapper.writeValueAsString(receiver.getData()); if (dockOsdData.contains("flighttask_step_code")) { JsonNode jsonNode = mapper.readTree(dockOsdData); if (jsonNode.has("flighttask_step_code")) { String deviceSn = device.getDeviceSn(); Long code = jsonNode.get("flighttask_step_code").asLong(); _deviceService.updateWorkType(deviceSn,code); if(code != null) { //只要状态 "0":"作业准备中","1":"飞行作业中","5":"任务空闲" if(code == 0 || code == 1 || code == 5) { _deviceService.updateWorkType(deviceSn,code); } } } } else if (dockOsd.contains("network_state")) { JsonNode jsonNode = mapper.readTree(dockOsd); if (jsonNode.has("drone_in_dock")) { } else if (dockOsdData.contains("network_state")) { JsonNode jsonNode = mapper.readTree(dockOsdData); if (jsonNode.has("drone_in_dock") && jsonNode.get("sub_device").has("device_online_status")) { String childDeviceSn=jsonNode.get("sub_device").get("device_sn").asText(); Long inDock=jsonNode.get("drone_in_dock").asLong(); _deviceService.updateWorkType(childDeviceSn,inDock); Long deviceOnlineStatus=jsonNode.get("sub_device").get("device_online_status").asLong(); if(inDock.equals(1L) && deviceOnlineStatus.equals(0L)) {// 飞行器在舱内且关机状态为6 _deviceService.updateWorkType(childDeviceSn,6L); } } } else if (dockOsd.contains("job_number")) { } else if (dockOsdData.contains("job_number")) { } }catch (Exception e){ e.printStackTrace(); } wsMessage.getData().setHost(data); sendMessageService.sendBatch(webSessions, wsMessage); _websocketdockserver.sendToAll(dockOsd); // 发送给websocket客户端 _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATUS.getType(), dockOsd); } } }
