package com.ard.sample.component.mqtt.handler; import com.ard.sample.component.mqtt.model.Chan; 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; import java.io.IOException; /** * @author sean * @version 1.2 * @date 2022/9/9 */ @Component public class ServicesReplyHandler { @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 * @throws IOException */ @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() {}); 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() {}); receiver.setData(list); } else { ServiceReply reply = mapper.convertValue(receiver.getData(), new TypeReference() {}); receiver.setData(reply); } Chan> 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. } }