任务提交 完成部分mqtt入库

ruoyi-framework/src/main/java/com/ruoyi/framework/config/mqtt/MqttInboundConfiguration.java

@@ -3,6 +3,7 @@ package com.ruoyi.framework.config.mqtt;
 
 
 import com.ruoyi.framework.utils.SSLUtils;
+import com.ruoyi.system.service.SaveDataService;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -19,6 +20,7 @@ import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
 import org.springframework.integration.mqtt.support.MqttHeaders;
 import org.springframework.messaging.*;
 
+import javax.annotation.Resource;
 import java.util.UUID;
 
 /**
@@ -34,6 +36,9 @@ public class MqttInboundConfiguration {
     @Autowired
     private MqttConfiguration mqttProperties;
 
+    @Resource
+    private SaveDataService saveDataService;
+
     @Bean
     public MessageChannel mqttInputChannel() {
         return new DirectChannel();
@@ -98,9 +103,9 @@ public class MqttInboundConfiguration {
                         + "\nReceive payLoad ===> " + payload
                         + " \tQOS ===> " + qos
                         + "\n Topics: " + recvTopic;
-
-               log.info(handMessage);
-               //TODO 做持久化
+                log.info(handMessage);
+                //做持久化
+                saveDataService.saveDate(payload.toString(),recvTopic);
 
             }
         };

ruoyi-system/src/main/java/com/ruoyi/system/domain/CarPosition.java

@@ -34,5 +34,10 @@ public class CarPosition implements Serializable {
      */
     private Integer status;
 
+    /**
+     * 站码
+     */
+    private String id;
+
 }
 

ruoyi-system/src/main/java/com/ruoyi/system/domain/MeasureUpload.java

@@ -45,5 +45,10 @@ public class MeasureUpload implements Serializable {
      */
     private Integer status;
 
+    /**
+     * 站码
+     */
+    private String id;
+
 }
 

ruoyi-system/src/main/java/com/ruoyi/system/domain/TaskNotice.java

@@ -50,5 +50,10 @@ public class TaskNotice implements Serializable {
      */
     private String code;
 
+    /**
+     * 停泊点编号
+     */
+    private Long pn;
+
 }
 

ruoyi-system/src/main/java/com/ruoyi/system/domain/TaskResult.java

@@ -18,7 +18,7 @@ public class TaskResult implements Serializable {
 /**
      * 成果ID
      */
-    private Integer resultId;
+    private Long resultId;
 /**
      * 站码
      */

ruoyi-system/src/main/java/com/ruoyi/system/mapper/SiteInfoMapper.java

@@ -21,6 +21,14 @@ public interface SiteInfoMapper {
      */
     SiteInfo queryById(Long siteId);
 
+    /**
+     * 通过ID查询单条数据
+     *
+     * @param siteCode 主键
+     * @return 实例对象
+     */
+    SiteInfo queryBySiteCode(String siteCode);
+
     /**
      * 查询指定行数据
      *

ruoyi-system/src/main/java/com/ruoyi/system/service/SaveDataService.java

@@ -0,0 +1,7 @@
+package com.ruoyi.system.service;
+
+public interface SaveDataService {
+
+    void saveDate(String data, String topics);
+
+}

ruoyi-system/src/main/java/com/ruoyi/system/service/impl/SaveDataServiceImpl.java

@@ -0,0 +1,228 @@
+package com.ruoyi.system.service.impl;
+
+import cn.hutool.json.JSONUtil;
+import com.ruoyi.system.domain.*;
+import com.ruoyi.system.mapper.*;
+import com.ruoyi.system.service.SaveDataService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.Date;
+
+/**
+ * classname: 站点信息
+ * description  服务实现类
+ */
+@Service
+@Slf4j
+public class SaveDataServiceImpl implements SaveDataService {
+
+    @Resource
+    private HeartBeatMapper heartBeatMapper;
+
+    @Resource
+    private SiteInfoMapper siteInfoMapper;
+
+    @Resource
+    private TaskNoticeMapper taskNoticeMapper;
+
+    @Resource
+    private TaskResultMapper taskResultMapper;
+
+    @Resource
+    private CarPositionMapper carPositionMapper;
+
+    @Resource
+    private MeasureUploadMapper measureUploadMapper;
+
+    @Override
+    public void saveDate(String data, String topics) {
+        String[] topicsList = topics.split("/");
+        if (topicsList.length == 3) {
+            String action = topicsList[2];
+            switch (action) {
+                case "heartbeat" ->
+                    //存入心跳
+                        heartbeat(data);
+                case "startMeasure" ->
+                    //开始测流
+                        startMeasure(data);
+                case "taskFault" ->
+                    //任务故障
+                        taskFault(data);
+                case "positionUpdate" ->
+                    //小车位置
+                        positionUpdate(data);
+                case "measureBegin" ->
+                    //停泊点开始测流
+                        measureBegin(data);
+                case "measureUpload" ->
+                    //停泊点流速上报
+                        measureUpload(data);
+                case "return" ->
+                    //小车开始返航
+                        xiaoReturn(data);
+                case "taskend" ->
+                    //小车任务完成上报
+                        taskend(data);
+                case "waterLevelUpload" ->
+                    //水位数据上报
+                        waterLevelUpload(data);
+                case "measureStop" ->
+                    //测流停止
+                        measureStop(data);
+                case "resultUpload" ->
+                    //测流上报
+                        resultUpload(data);
+                default -> log.error("topic 命令没找到:{}", topics);
+            }
+        } else {
+            log.error("topic is error:{}",topics);
+        }
+    }
+
+    public void heartbeat(String data){
+        try {
+            HeartBeat heartBeat = JSONUtil.toBean(data, HeartBeat.class);
+            heartBeat.setHeartId(0L);
+            heartBeat.setCreateTime(new Date());
+            SiteInfo siteInfo = siteInfoMapper.queryBySiteCode(heartBeat.getId());
+            if (siteInfo == null){
+                log.error("查询站点失败入库失败 站码:{}",heartBeat.getId());
+                return;
+            }
+            heartBeat.setSiteId(siteInfo.getSiteId());
+            heartBeatMapper.insert(heartBeat);
+            log.info("insert heartbeat:{}",heartBeat);
+        } catch (Exception e) {
+            log.error("存入心跳 入库错误:",e);
+        }
+
+    }
+
+    public void startMeasure(String data){
+        try {
+            TaskNotice taskNotice = JSONUtil.toBean(data, TaskNotice.class);
+            SiteInfo siteInfo = siteInfoMapper.queryBySiteCode(taskNotice.getId());
+            if (siteInfo == null) {
+                log.error("查询站点失败入库失败 站码:{}", taskNotice.getId());
+                return;
+            }
+            taskNotice.setStartId(0L);
+            if (taskNotice.getWorkmode() == 1) {
+                taskNotice.setRemark("时间任务触发 开始测流任务");
+            } else if (taskNotice.getWorkmode() == 2) {
+                taskNotice.setRemark("变幅任务触发 开始测流任务");
+            } else if (taskNotice.getWorkmode() == 3) {
+                taskNotice.setRemark("手动任务触发 开始测流任务");
+            }
+            taskNotice.setCreateTime(new Date());
+            taskNotice.setSiteId(siteInfo.getSiteId());
+            taskNoticeMapper.insert(taskNotice);
+            //新建结果报告 状态设置为测流中
+            TaskResult taskResult = new TaskResult();
+            taskResult.setResultId(0L);
+            taskResult.setTaskid(taskNotice.getTaskid());
+            taskResult.setStatus(0);
+            taskResult.setSiteId(siteInfo.getSiteId());
+            taskResult.setCreateTime(new Date());
+            taskResultMapper.insert(taskResult);
+        } catch (Exception e) {
+            log.error("开始测流 入库错误:",e);
+        }
+
+    }
+
+    public void taskFault(String data) {
+        try {
+            TaskNotice taskNotice = JSONUtil.toBean(data, TaskNotice.class);
+            SiteInfo siteInfo = siteInfoMapper.queryBySiteCode(taskNotice.getId());
+            if (siteInfo == null) {
+                log.error("查询站点失败入库失败 站码:{}", taskNotice.getId());
+                return;
+            }
+            taskNotice.setStartId(0L);
+            taskNotice.setRemark("任务故障");
+            taskNotice.setCreateTime(new Date());
+            taskNotice.setSiteId(siteInfo.getSiteId());
+            taskNoticeMapper.insert(taskNotice);
+        } catch (Exception e) {
+            log.error("任务故障 入库错误:", e);
+        }
+    }
+    public void positionUpdate(String data){
+        try {
+            CarPosition carPosition= JSONUtil.toBean(data, CarPosition.class);
+            SiteInfo siteInfo = siteInfoMapper.queryBySiteCode(carPosition.getId());
+            if (siteInfo == null) {
+                log.error("查询站点失败入库失败 站码:{}", carPosition.getId());
+                return;
+            }
+            carPosition.setSiteId(siteInfo.getSiteId());
+            carPosition.setCarId(0L);
+            carPosition.setCreateTime(new Date());
+            carPositionMapper.insert(carPosition);
+        } catch (Exception e) {
+            log.error("小车位置 入库错误:", e);
+        }
+    }
+
+    public void measureBegin(String data){
+        try {
+            TaskNotice taskNotice = JSONUtil.toBean(data, TaskNotice.class);
+            SiteInfo siteInfo = siteInfoMapper.queryBySiteCode(taskNotice.getId());
+            if (siteInfo == null) {
+                log.error("查询站点失败入库失败 站码:{}", taskNotice.getId());
+                return;
+            }
+            taskNotice.setStartId(0L);
+            taskNotice.setRemark(taskNotice.getPn() + "号停泊点开始测流");
+            taskNotice.setCreateTime(new Date());
+            taskNotice.setSiteId(siteInfo.getSiteId());
+            taskNoticeMapper.insert(taskNotice);
+        } catch (Exception e) {
+            log.error("停泊点开始测流 入库错误:", e);
+        }
+    }
+
+    public void measureUpload(String data){
+        try {
+            MeasureUpload measureUpload= JSONUtil.toBean(data, MeasureUpload.class);
+            SiteInfo siteInfo = siteInfoMapper.queryBySiteCode(measureUpload.getId());
+            if (siteInfo == null) {
+                log.error("查询站点失败入库失败 站码:{}", measureUpload.getId());
+                return;
+            }
+            measureUpload.setSiteId(siteInfo.getSiteId());
+            measureUpload.setMeasureId(0L);
+            measureUpload.setCreateTime(new Date());
+            measureUploadMapper.insert(measureUpload);
+        } catch (Exception e) {
+            log.error("停泊点流速上报 入库错误:", e);
+        }
+    }
+
+    public void taskend(String data){
+        log.info("taskend:{}",data);
+    }
+
+    public void waterLevelUpload(String data){
+        log.info("waterLevelUpload:{}",data);
+    }
+
+    public void xiaoReturn(String data){
+        log.info("xiaoReturn:{}",data);
+    }
+
+    public void measureStop(String data){
+        log.info("measureStop:{}",data);
+    }
+
+    public void resultUpload(String data){
+        log.info("resultUpload:{}",data);
+    }
+
+
+
+}

ruoyi-system/src/main/resources/mapper/CarPositionMapper.xml

@@ -8,12 +8,13 @@
         <result property="createTime" column="create_time" jdbcType="TIMESTAMP"/>
         <result property="siteId" column="site_id" jdbcType="INTEGER"/>
         <result property="status" column="status" jdbcType="INTEGER"/>
+        <result property="id" column="id" jdbcType="VARCHAR"/>
     </resultMap>
 
     <!--查询单个-->
     <select id="queryById" resultMap="CarPositionMap">
         select
-            car_id, position, create_time, site_id, status
+            car_id, position, create_time, site_id, status, id
         from car_position
         where car_id = #{carId}
     </select>
@@ -21,7 +22,7 @@
     <!--查询单个-->
     <select id="queryBysiteId" resultMap="CarPositionMap">
         select
-            car_id, position, create_time, site_id, status
+            car_id, position, create_time, site_id, status, id
         from car_position
         where site_id = #{siteId}
         order by create_time desc
@@ -31,7 +32,7 @@
     <!--查询指定行数据-->
     <select id="queryAllByLimit" resultMap="CarPositionMap">
         select
-car_id, position, create_time, site_id, status
+            car_id, position, create_time, site_id, status, id
         from car_position
         <where>
             <if test="carId != null">
@@ -49,6 +50,9 @@ car_id, position, create_time, site_id, status
             <if test="status != null">
                 and status = #{status}
             </if>
+            <if test="id != null and id != ''">
+                and id = #{id}
+            </if>
         </where>
         limit #{pageable.offset}, #{pageable.pageSize}
     </select>
@@ -73,28 +77,31 @@ car_id, position, create_time, site_id, status
             <if test="status != null">
                 and status = #{status}
             </if>
+            <if test="id != null and id != ''">
+                and id = #{id}
+            </if>
         </where>
     </select>
 
     <!--新增所有列-->
     <insert id="insert" keyProperty="carId" useGeneratedKeys="true">
-        insert into car_position(position, create_time, site_id, status)
-        values (#{position}, #{createTime}, #{siteId}, #{status})
+        insert into car_position(position, create_time, site_id, status, id)
+        values (#{position}, #{createTime}, #{siteId}, #{status}, #{id})
     </insert>
 
     <insert id="insertBatch" keyProperty="carId" useGeneratedKeys="true">
-        insert into car_position(position, create_time, site_id, status)
+        insert into car_position(position, create_time, site_id, status, id)
         values
         <foreach collection="entities" item="entity" separator=",">
-        (#{entity.position}, #{entity.createTime}, #{entity.siteId}, #{entity.status})
+        (#{entity.position}, #{entity.createTime}, #{entity.siteId}, #{entity.status}, #{entity.id})
         </foreach>
     </insert>
 
     <insert id="insertOrUpdateBatch" keyProperty="carId" useGeneratedKeys="true">
-        insert into car_position(position, create_time, site_id, status)
+        insert into car_position(position, create_time, site_id, status,id)
         values
         <foreach collection="entities" item="entity" separator=",">
-            (#{entity.position}, #{entity.createTime}, #{entity.siteId}, #{entity.status})
+            (#{entity.position}, #{entity.createTime}, #{entity.siteId}, #{entity.status},#{entity.id})
         </foreach>
         on duplicate key update
 position = values(position),
@@ -119,6 +126,9 @@ status = values(status)
             <if test="status != null">
                 status = #{status},
             </if>
+            <if test="id != null">
+                id = #{id},
+            </if>
         </set>
         where car_id = #{carId}
     </update>

ruoyi-system/src/main/resources/mapper/MeasureUploadMapper.xml

@@ -10,12 +10,13 @@
         <result property="createTime" column="create_time" jdbcType="TIMESTAMP"/>
         <result property="siteId" column="site_id" jdbcType="INTEGER"/>
         <result property="status" column="status" jdbcType="INTEGER"/>
+        <result property="id" column="id" jdbcType="VARCHAR"/>
     </resultMap>
 
     <!--查询单个-->
     <select id="queryById" resultMap="MeasureUploadMap">
         select
-measure_id, planid, pn, wspeed, create_time, site_id, status
+            measure_id, planid, pn, wspeed, create_time, site_id, status, id
         from measure_upload
         where measure_id = #{measureId}
     </select>
@@ -23,7 +24,7 @@ measure_id, planid, pn, wspeed, create_time, site_id, status
     <!--查询多个-->
     <select id="queryBySiteIdTime" resultMap="MeasureUploadMap">
         select
-            measure_id, planid, pn, wspeed, create_time, site_id, status
+            measure_id, planid, pn, wspeed, create_time, site_id, status, id
         from measure_upload
         where site_id = #{siteId} and create_time >= #{createTime}
     </select>
@@ -31,7 +32,7 @@ measure_id, planid, pn, wspeed, create_time, site_id, status
     <!--查询指定行数据-->
     <select id="queryAllByLimit" resultMap="MeasureUploadMap">
         select
-measure_id, planid, pn, wspeed, create_time, site_id, status
+            measure_id, planid, pn, wspeed, create_time, site_id, status, id
         from measure_upload
         <where>
             <if test="measureId != null">
@@ -55,6 +56,9 @@ measure_id, planid, pn, wspeed, create_time, site_id, status
             <if test="status != null">
                 and status = #{status}
             </if>
+            <if test="id != null and id != ''">
+                and id = #{id}
+            </if>
         </where>
         limit #{pageable.offset}, #{pageable.pageSize}
     </select>
@@ -85,36 +89,40 @@ measure_id, planid, pn, wspeed, create_time, site_id, status
             <if test="status != null">
                 and status = #{status}
             </if>
+            <if test="id != null and id != ''">
+                and id = #{id}
+            </if>
         </where>
     </select>
 
     <!--新增所有列-->
     <insert id="insert" keyProperty="measureId" useGeneratedKeys="true">
-        insert into measure_upload(planid, pn, wspeed, create_time, site_id, status)
-        values (#{planid}, #{pn}, #{wspeed}, #{createTime}, #{siteId}, #{status})
+        insert into measure_upload(planid, pn, wspeed, create_time, site_id, status, id)
+        values (#{planid}, #{pn}, #{wspeed}, #{createTime}, #{siteId}, #{status},#{id})
     </insert>
 
     <insert id="insertBatch" keyProperty="measureId" useGeneratedKeys="true">
-        insert into measure_upload(planid, pn, wspeed, create_time, site_id, status)
+        insert into measure_upload(planid, pn, wspeed, create_time, site_id, status,id)
         values
         <foreach collection="entities" item="entity" separator=",">
-        (#{entity.planid}, #{entity.pn}, #{entity.wspeed}, #{entity.createTime}, #{entity.siteId}, #{entity.status})
+        (#{entity.planid}, #{entity.pn}, #{entity.wspeed}, #{entity.createTime}, #{entity.siteId}, #{entity.status},#{entity.id})
         </foreach>
     </insert>
 
     <insert id="insertOrUpdateBatch" keyProperty="measureId" useGeneratedKeys="true">
-        insert into measure_upload(planid, pn, wspeed, create_time, site_id, status)
+        insert into measure_upload(planid, pn, wspeed, create_time, site_id, status, id)
         values
         <foreach collection="entities" item="entity" separator=",">
-            (#{entity.planid}, #{entity.pn}, #{entity.wspeed}, #{entity.createTime}, #{entity.siteId}, #{entity.status})
+            (#{entity.planid}, #{entity.pn}, #{entity.wspeed}, #{entity.createTime}, #{entity.siteId}, #{entity.status}, #{entity.id})
         </foreach>
         on duplicate key update
-planid = values(planid),
-pn = values(pn),
-wspeed = values(wspeed),
-create_time = values(create_time),
-site_id = values(site_id),
-status = values(status)
+        planid = values(planid),
+        pn = values(pn),
+        wspeed = values(wspeed),
+        create_time = values(create_time),
+        site_id = values(site_id),
+        status = values(status),
+        id = values(id)
     </insert>
 
     <!--通过主键修改数据-->
@@ -139,6 +147,9 @@ status = values(status)
             <if test="status != null">
                 status = #{status},
             </if>
+            <if test="id != null">
+                id = #{id},
+            </if>
         </set>
         where measure_id = #{measureId}
     </update>

ruoyi-system/src/main/resources/mapper/SiteInfoMapper.xml

@@ -32,6 +32,14 @@
         where site_id = #{siteId}
     </select>
 
+    <!--查询单个-->
+    <select id="queryBySiteCode" resultMap="SiteInfoMap">
+        select
+            site_id, id, site_name, site_type, lon, lat, site_time, admin_region, manage_unit, affiliated_unit, contact_name, contact_phone, creat_time, is_del, device_id, video_url, video_user, video_pwd, video_port
+        from site_info
+        where id = #{siteCode}
+    </select>
+
     <!--查询指定行数据-->
     <select id="queryAllByLimit" resultMap="SiteInfoMap">
         select

ruoyi-system/src/main/resources/mapper/TaskNoticeMapper.xml

@@ -12,12 +12,13 @@
         <result property="id" column="id" jdbcType="VARCHAR"/>
         <result property="remark" column="remark" jdbcType="VARCHAR"/>
         <result property="code" column="code" jdbcType="VARCHAR"/>
+        <result property="pn" column="pn" jdbcType="INTEGER"/>
     </resultMap>
 
     <!--查询单个-->
     <select id="queryById" resultMap="TaskNoticeMap">
         select
-start_id, taskid, workmode, planid, create_time, site_id, id, remark, code
+start_id, taskid, workmode, planid, create_time, site_id, id, remark, code, pn
         from task_notice
         where start_id = #{startId}
     </select>
@@ -25,7 +26,7 @@ start_id, taskid, workmode, planid, create_time, site_id, id, remark, code
     <!--查询指定行数据-->
     <select id="queryAllByLimit" resultMap="TaskNoticeMap">
         select
-            start_id, taskid, workmode, planid, create_time, site_id, id, remark, code
+            start_id, taskid, workmode, planid, create_time, site_id, id, remark, code, pn
         from task_notice
         <where>
             <if test="siteId != null">
@@ -67,38 +68,42 @@ start_id, taskid, workmode, planid, create_time, site_id, id, remark, code
             <if test="code != null and code != ''">
                 and code = #{code}
             </if>
+            <if test="pn != null">
+                and pn = #{pn}
+            </if>
         </where>
     </select>
 
     <!--新增所有列-->
     <insert id="insert" keyProperty="startId" useGeneratedKeys="true">
-        insert into task_notice(taskid, workmode, planid, create_time, site_id, id, remark, code)
-        values (#{taskid}, #{workmode}, #{planid}, #{createTime}, #{siteId}, #{id}, #{remark}, #{code})
+        insert into task_notice(taskid, workmode, planid, create_time, site_id, id, remark, code, pn)
+        values (#{taskid}, #{workmode}, #{planid}, #{createTime}, #{siteId}, #{id}, #{remark}, #{code}, #{pn})
     </insert>
 
     <insert id="insertBatch" keyProperty="startId" useGeneratedKeys="true">
-        insert into task_notice(taskid, workmode, planid, create_time, site_id, id, remark, code)
+        insert into task_notice(taskid, workmode, planid, create_time, site_id, id, remark, code, pn)
         values
         <foreach collection="entities" item="entity" separator=",">
-        (#{entity.taskid}, #{entity.workmode}, #{entity.planid}, #{entity.createTime}, #{entity.siteId}, #{entity.id}, #{entity.remark}, #{entity.code})
+        (#{entity.taskid}, #{entity.workmode}, #{entity.planid}, #{entity.createTime}, #{entity.siteId}, #{entity.id}, #{entity.remark}, #{entity.code}, #{entity.pn})
         </foreach>
     </insert>
 
     <insert id="insertOrUpdateBatch" keyProperty="startId" useGeneratedKeys="true">
-        insert into task_notice(taskid, workmode, planid, create_time, site_id, id, remark, code)
+        insert into task_notice(taskid, workmode, planid, create_time, site_id, id, remark, code, pn)
         values
         <foreach collection="entities" item="entity" separator=",">
-            (#{entity.taskid}, #{entity.workmode}, #{entity.planid}, #{entity.createTime}, #{entity.siteId}, #{entity.id}, #{entity.remark}, #{entity.code})
+            (#{entity.taskid}, #{entity.workmode}, #{entity.planid}, #{entity.createTime}, #{entity.siteId}, #{entity.id}, #{entity.remark}, #{entity.code}, #{entity.pn})
         </foreach>
         on duplicate key update
-taskid = values(taskid),
-workmode = values(workmode),
-planid = values(planid),
-create_time = values(create_time),
-site_id = values(site_id),
-id = values(id),
-remark = values(remark),
-code = values(code)
+        taskid = values(taskid),
+        workmode = values(workmode),
+        planid = values(planid),
+        create_time = values(create_time),
+        site_id = values(site_id),
+        id = values(id),
+        remark = values(remark),
+        code = values(code),
+        pn = values(pn)
     </insert>
 
     <!--通过主键修改数据-->
@@ -129,6 +134,9 @@ code = values(code)
             <if test="code != null and code != ''">
                 code = #{code},
             </if>
+            <if test="pn != null">
+                pn = #{pn},
+            </if>
         </set>
         where start_id = #{startId}
     </update>