liurunyu
6 天以前 0909caac96ddac21f007d6c7ba6b348de9d2981a
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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
package com.dy.rtuMw.server.rtuData.pSdV1;
 
import com.dy.common.mw.protocol4Mqtt.MqttSubMsg;
import com.dy.common.mw.protocol4Mqtt.pSdV1.upVos.WeatherVo;
import com.dy.pipIrrGlobal.pojoPr.PrStWeather;
import com.dy.pipIrrGlobal.pojoRm.RmWeatherHistory;
import com.dy.pipIrrGlobal.pojoRm.RmWeatherLast;
import com.dy.rtuMw.server.rtuData.TaskSurpport;
import com.dy.rtuMw.server.rtuData.dbSv.DbSv;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
 
/**
 * @Author: liurunyu
 * @Date: 2025/6/25 11:37
 * @Description
 */
public class TkDealWeatherSdV1 extends TaskSurpport {
 
    private static Logger log = LogManager.getLogger(TkDealWeatherSdV1.class.getName());
 
    //类ID,一定与Tree.xml配置文件中配置一致
    public static final String taskId = "TkDealWeatherSdV1";
 
    /**
     * 执行节点任务
     *
     * @param data 需要处理的数据
     */
    @Override
    public void execute(Object data) {
        //前面的任务已经判断了data不为空且为气象数据
        MqttSubMsg msg = (MqttSubMsg) data;
        WeatherVo stVo = (WeatherVo) msg.vo4Up;
        Object[] objs = this.getTaskResults(TkPreGenObjs4WeatherSdV1.taskId);
        DbSv sv = (DbSv) objs[0];
        PrStWeather stPo = (PrStWeather) objs[1];
        try{
            this.doDeal(sv, stPo, msg, stVo);
        }catch (Exception e){
            log.error("保存气象数据时发生异常", e);
        }
    }
 
    /**
     * 处理上行消息数据
     * @param sv 服务
     * @param stPo 实体对象
     * @param msg 上行的订阅消息
     * @param stVo 上行的设备数据
     */
    private void doDeal(DbSv sv,
                        PrStWeather stPo,
                        MqttSubMsg msg,
                        WeatherVo stVo) throws Exception {
        RmWeatherHistory hpo = this.saveHistory(sv, stPo, msg, stVo);
        this.saveOrUpdateLast(sv, stPo, msg, stVo, hpo);
    }
    /**
     * 处理上行消息数据
     * @param sv 服务
     * @param stPo 实体对象
     * @param msg 上行的订阅消息
     * @param stVo 上行的设备数据
     * @param hpo 历史记录最新数据
     */
    private void saveOrUpdateLast(DbSv sv,
                                  PrStWeather stPo,
                                  MqttSubMsg msg,
                                  WeatherVo stVo,
                                  RmWeatherHistory hpo) throws Exception {
        RmWeatherLast po = sv.getRmWeatherLast(stPo.id) ;
        if(po == null){
            po = new RmWeatherLast();
            po.valueFrom(msg, stVo);
            po.weatherId = stPo.id ;
            po.lastHistoryId = hpo==null?null:hpo.id ;
            sv.saveRmWeatherLast(po) ;
        }else{
            po.valueFrom(msg, stVo);
            po.weatherId = stPo.id ;
            po.lastHistoryId = hpo==null?null:hpo.id ;
            sv.updateRmWeatherLast(po);
        }
    }
 
    /**
     * 处理上行消息数据
     * @param sv 服务
     * @param stPo 实体对象
     * @param msg 上行的订阅消息
     * @param stVo 上行的设备数据
     */
    private RmWeatherHistory saveHistory(DbSv sv,
                                      PrStWeather stPo,
                                      MqttSubMsg msg,
                                      WeatherVo stVo) throws Exception {
        RmWeatherHistory po = new RmWeatherHistory();
        po.valueFrom(msg, stVo);
        po.weatherId = stPo.id ;
        sv.saveRmWeatherHistory(po);
        return po ;
    }
 
}