zhaowenyu
2026-01-14 eeed071b9da26a1c389ab5572b722d36812c10a7
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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.websocket.enums.MessageType;
import com.ard.sample.component.websocket.handler.DockWebSocketHandler;
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 PropertySetReplyHandler {
 
    @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_PROPERTY_SET_REPLY)
    public void serviceReply(Message<?> message) throws IOException {
        // 解析消息的主题,从中提取设备序列号等信息
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
        // thing/product/7CTDLCE00A1UPE/property/set_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>() {});
        Chan<CommonTopicReceiver<?>> chan = Chan.getInstance();
        // Put the message to the chan object.
        chan.put(receiver);
 
        receiver.setGateway(deviceSn);
        receiver.setType(MessageType.CONTROL.getType());
//        receiver.setMethod();
 
        // 设备属性设置后的处理逻辑
 
        // WebSocket 接口,发给前端
        String dockControl="";
        dockControl=mapper.writeValueAsString(receiver);
 
        _dockWebSocketHandler.sendMessageToSubscribers(deviceSn, MessageType.CONTROL.getType(),dockControl, "");
    }
}