mqtt

cl

3 天以前 18c51d40be82435289ba3be6bd5f8e15fdf786f7

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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
package com.zy.integration.iot.handler.impl;
 
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.core.common.Cools;
import com.zy.integration.iot.biz.IotInstructionService;
import com.zy.integration.iot.handler.IotInboundMessageHandler;
import com.zy.integration.iot.publish.IotPublishService;
import com.zy.iot.service.IotDbConfigService;
import com.zy.iot.entity.IotFeedbackMessage;
import com.zy.iot.entity.IotInstructionMessage;
import com.zy.iot.entity.IotTopicConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
/**
 * MQTT 入站分发器。
 * 按 topic 识别 stow/pick/feedback，并在接单后立即尝试回发 feedback。
 */
@Slf4j
@Service
public class IotInboundMessageHandlerImpl implements IotInboundMessageHandler {
 
    @Autowired
    private IotDbConfigService iotDbConfigService;
    @Autowired
    private IotInstructionService iotInstructionService;
    @Autowired
    private IotPublishService iotPublishService;
 
    /**
     * MQTT 回调层只做 topic 分发和报文反序列化，业务动作仍下沉到 service。
     */
    @Override
    public void handleMessage(String topic, String payload) {
        if (!iotDbConfigService.isMqttEnabled() || Cools.isEmpty(topic) || payload == null) {
            return;
        }
        try {
            IotTopicConfig topics = iotDbConfigService.getEffectiveTopics();
            Long recordId = null;
            if (topic.equals(topics.getEgressStow())) {
                IotInstructionMessage stow = parseJsonPayload(payload, topic, IotInstructionMessage.class);
                if (stow == null) {
                    return;
                }
                recordId = iotInstructionService.handleStowInstruction(stow, topic, payload);
            } else if (topic.equals(topics.getEgressPick())) {
                IotInstructionMessage pick = parseJsonPayload(payload, topic, IotInstructionMessage.class);
                if (pick == null) {
                    return;
                }
                recordId = iotInstructionService.handlePickInstruction(pick, topic, payload);
            } else if (topic.equals(topics.getEgressFeedback())) {
                IotFeedbackMessage ack = parseJsonPayload(payload, topic, IotFeedbackMessage.class);
                if (ack == null) {
                    return;
                }
                iotInstructionService.handleFeedbackAck(ack, topic, payload);
            } else {
                log.warn("ignore unknown iot topic={}, payload={}", topic, payload);
            }
            if (recordId != null) {
                // feedback 先写本地记录，再立即尝试发送；传输失败由定时任务补偿。
                iotPublishService.publishRecordNow(recordId);
            }
        } catch (Exception e) {
            log.error("handle iot inbound message failed, topic={}, payload={}", topic, payload, e);
        }
    }
 
    private <T> T parseJsonPayload(String payload, String topic, Class<T> type) {
        try {
            return JSON.parseObject(payload, type);
        } catch (JSONException e) {
            log.warn("IoT MQTT 非合法 JSON topic={} payload={}", topic, payload, e);
            return null;
        }
    }
}