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<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);
|
} else {
|
ServiceReply reply = mapper.convertValue(receiver.getData(), new TypeReference<ServiceReply>() {});
|
receiver.setData(reply);
|
}
|
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.
|
}
|
}
|
