北向接口路由转发代码提交

soc-api/soc-api-system/src/main/java/com/xunmei/system/api/RemoteMediatorService.java

@@ -4,9 +4,7 @@ import com.xunmei.common.core.constant.ServiceNameConstants;
 import com.xunmei.common.core.web.domain.AjaxResult;
 import com.xunmei.system.api.factory.RemoteMediatorFallbackFactory;
 import org.springframework.cloud.openfeign.FeignClient;
-import org.springframework.web.bind.annotation.*;
-
-import java.util.List;
+import org.springframework.web.bind.annotation.GetMapping;
 
 /**
  * 字典服务
@@ -18,4 +16,7 @@ public interface RemoteMediatorService {
 
     @GetMapping("/mediator/channelCacheDeal")
     AjaxResult channelCacheDeal();
+
+    @GetMapping("/mediator/videoIntegrityCheck")
+    AjaxResult videoIntegrityCheck();
 }

soc-api/soc-api-system/src/main/java/com/xunmei/system/api/factory/RemoteMediatorFallbackFactory.java

@@ -25,6 +25,11 @@ public class RemoteMediatorFallbackFactory implements FallbackFactory<RemoteMedi
             public AjaxResult channelCacheDeal() {
                 return null;
             }
+
+            @Override
+            public AjaxResult videoIntegrityCheck() {
+                return null;
+            }
         };
     }
 }

soc-common/soc-common-core/pom.xml

@@ -71,6 +71,10 @@
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+        </dependency>
 
         <!-- Alibaba Fastjson -->
         <dependency>

soc-common/soc-common-core/src/main/java/com/xunmei/common/core/utils/JacksonUtils.java

@@ -8,11 +8,12 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.databind.node.ArrayNode;
-import lombok.extern.slf4j.Slf4j;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import lombok.extern.slf4j.Slf4j;
+
 import java.io.File;
 import java.io.IOException;
-import java.text.SimpleDateFormat;
 
 /**
  * @author jingyuanchao
@@ -29,15 +30,15 @@ public class JacksonUtils {
 
     //以静态代码块初始化
     static {
-        //对象的所有字段全部列入序列化
+        // 对象的所有字段全部列入序列化
         objectMapper.setSerializationInclusion(JsonInclude.Include.ALWAYS);
-        //取消默认转换timestamps形式
+        // 取消默认转换 timestamps 形式
         objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
-        //忽略空Bean转json的错误
+        // 忽略空 Bean 转 json 的错误
         objectMapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
-        //所有的日期格式都统一为以下的格式，即yyyy-MM-dd HH:mm:ss
-        objectMapper.setDateFormat(new SimpleDateFormat(STANDARD_FORMAT));
-        //忽略 在json字符串中存在，但在java对象中不存在对应属性的情况。防止错误
+        // 注册 JavaTimeModule 以支持 Java 8 日期时间 API
+        objectMapper.registerModule(new JavaTimeModule());
+        // 忽略 在 json 字符串中存在，但在 java 对象中不存在对应属性的情况。防止错误
         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
     }
 
@@ -145,7 +146,6 @@ public class JacksonUtils {
     }
 
 
-    /**===========以下是从JsonNode对象中获取key值的方法，个人觉得有点多余，直接用JsonNode自带的取值方法会好点，出于纠结症，还是补充进来了*/
     public static String getString(JsonNode jsonObject, String key) {
         String s = jsonObject.get(key).asText();
         return s;

soc-modules/soc-modules-job/src/main/java/com/xunmei/job/task/MediatorTask.java

@@ -3,7 +3,6 @@ package com.xunmei.job.task;
 import cn.hutool.core.lang.UUID;
 import com.alibaba.fastjson2.JSON;
 import com.xunmei.common.core.web.domain.AjaxResult;
-import com.xunmei.system.api.RemoteDrillService;
 import com.xunmei.system.api.RemoteMediatorService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -26,5 +25,13 @@ public class MediatorTask {
         log.info("处理缓存中通道信息数据结束....,当前任务 id:{},当前时间:{},结果:{}", id, new Date(), JSON.toJSONString(result));
     }
 
+    //定时获取录像完整性检查数据
+    public void videoIntegrityCheck() {
+        String id = UUID.fastUUID().toString();
+        log.info("开始处理录像完整性检查,当前任务 id:{},当前时间:{}", id, new Date());
+        AjaxResult result = remoteMediatorService.videoIntegrityCheck();
+        log.info("录像完整性检查结束....,当前任务 id:{},当前时间:{},结果:{}", id, new Date(), JSON.toJSONString(result));
+    }
+
 
 }

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/api/north/controller/MediatorController.java

@@ -3,8 +3,10 @@ package com.xunmei.mediator.api.north.controller;
 
 import com.xunmei.common.core.web.domain.AjaxResult;
 import com.xunmei.mediator.api.host.service.IIotDvrChannelService;
+import com.xunmei.mediator.api.video.service.IVideoIntegrityCheckService;
 import io.swagger.annotations.ApiOperation;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
 
@@ -21,6 +23,9 @@ public class MediatorController {
     @Resource
     private IIotDvrChannelService iotDvrChannelService;
 
+    @Resource
+    IVideoIntegrityCheckService videoIntegrityCheckService;
+
 
     @ApiOperation("处理缓存中的通道信息数据")
     @GetMapping({"/mediator/channelCacheDeal"})
@@ -28,4 +33,12 @@ public class MediatorController {
         this.iotDvrChannelService.channelCacheDeal();
         return AjaxResult.success();
     }
+
+
+    @ApiOperation("定时任务获取所有的录像完整性数据")
+    @GetMapping({"/mediator/videoIntegrityCheck"})
+    public AjaxResult videoIntegrityCheck() {
+        this.videoIntegrityCheckService.invokeRecordInfosForSchedule();
+        return AjaxResult.success();
+    }
 }

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/api/video/service/IVideoIntegrityCheckService.java

@@ -2,10 +2,8 @@ package com.xunmei.mediator.api.video.service;
 
 import com.baomidou.mybatisplus.extension.service.IService;
 import com.xunmei.common.core.domain.video.MediatorVideoIntegrityCheck;
-import com.xunmei.system.api.dto.protection.ReceiveErrorDto;
 import com.xunmei.mediator.domain.dto.videoIntegrityCheck.VideoIntegrityCheckDto;
-
-import java.util.Date;
+import com.xunmei.system.api.dto.protection.ReceiveErrorDto;
 
 /**
  * <p>
@@ -17,9 +15,15 @@ import java.util.Date;
  */
 public interface IVideoIntegrityCheckService extends IService<MediatorVideoIntegrityCheck> {
 
+
+    VideoIntegrityCheckDto transform(Object data);
+
     ReceiveErrorDto saveData(VideoIntegrityCheckDto videoIntegrityCheckDto, String msgId);
 
     MediatorVideoIntegrityCheck findByOrgIdAndEquipmentCodeAndAndChannelCodeAndRecordDate(Long orgId, String equipmentCode, String channelCode);
 
     MediatorVideoIntegrityCheck findByOrgIdAndEquipmentCodeAndAndChannelCode(Long orgId, String equipmentCode, String channelCode);
+
+    void invokeRecordInfosForSchedule();
+
 }

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/api/video/service/impl/VideoIntegrityCheckServiceImpl.java

@@ -13,6 +13,7 @@ import com.xunmei.common.core.domain.video.MediatorVideoDaysCheck;
 import com.xunmei.common.core.domain.video.MediatorVideoIntegrityCheck;
 import com.xunmei.common.core.domain.video.MediatorVideoIntegrityCheckLog;
 import com.xunmei.common.core.enums.iot.VideoIntegrityStatus;
+import com.xunmei.common.core.utils.JacksonUtils;
 import com.xunmei.mediator.api.host.service.IIotDeviceService;
 import com.xunmei.mediator.api.host.service.IIotDvrChannelService;
 import com.xunmei.mediator.api.north.service.NorthErrorService;
@@ -20,13 +21,18 @@ import com.xunmei.mediator.api.video.mapper.VideoIntegrityCheckMapper;
 import com.xunmei.mediator.api.video.service.IVideoDaysCheckService;
 import com.xunmei.mediator.api.video.service.IVideoIntegrityCheckService;
 import com.xunmei.mediator.api.video.service.MediatorVideoIntegrityCheckLogService;
-import com.xunmei.system.api.domain.north.NorthError;
-import com.xunmei.system.api.dto.protection.ReceiveErrorDto;
 import com.xunmei.mediator.domain.dto.redis.RedisKey;
 import com.xunmei.mediator.domain.dto.videoIntegrityCheck.VideoIntegrityCheckDto;
+import com.xunmei.mediator.domain.dto.videoIntegrityCheck.VideoIntegrityCheckReq;
 import com.xunmei.mediator.util.RedisUtil;
+import com.xunmei.mediator.websocket.constant.WebSocketConstants;
+import com.xunmei.mediator.websocket.dto.WebsocketResult;
+import com.xunmei.mediator.websocket.service.RouterService;
+import com.xunmei.mediator.websocket.utils.WebSocketUtils;
 import com.xunmei.system.api.RemoteOrgService;
 import com.xunmei.system.api.domain.SysOrg;
+import com.xunmei.system.api.domain.north.NorthError;
+import com.xunmei.system.api.dto.protection.ReceiveErrorDto;
 import com.xunmei.system.api.util.LogUtils;
 import io.netty.util.internal.StringUtil;
 import org.apache.commons.lang3.StringUtils;
@@ -39,9 +45,7 @@ import java.time.LocalDate;
 import java.time.LocalTime;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 /**
  * <p>
@@ -52,7 +56,7 @@ import java.util.Map;
  * @since 2024-01-23
  */
 @Service
-public class VideoIntegrityCheckServiceImpl extends ServiceImpl<VideoIntegrityCheckMapper, MediatorVideoIntegrityCheck> implements IVideoIntegrityCheckService {
+public class VideoIntegrityCheckServiceImpl extends ServiceImpl<VideoIntegrityCheckMapper, MediatorVideoIntegrityCheck> implements IVideoIntegrityCheckService, RouterService {
 
     @Resource
     RedisUtil redisUtil;
@@ -71,6 +75,46 @@ public class VideoIntegrityCheckServiceImpl extends ServiceImpl<VideoIntegrityCh
     IIotDvrChannelService dvrChannelService;
 
     @Override
+    public String routerKey() {
+        StringJoiner result = new StringJoiner(",");
+        result.add(WebSocketConstants.RECORD_INFOS);
+        return result.toString();
+    }
+
+    @Override
+    public Object execute(Object data) {
+        VideoIntegrityCheckDto checkDto = transform(data);
+        this.saveData(checkDto,"");
+        return data;
+    }
+
+    @Override
+    public void invokeRecordInfosForSchedule() {
+        WebsocketResult websocketResult = new WebsocketResult();
+        websocketResult.setId(UUID.randomUUID().toString());
+        websocketResult.setTimestamp(new Date());
+        HashMap<String, Object> hashMap = new HashMap<>();
+        hashMap.put(WebSocketConstants.SERVICE, WebSocketConstants.GET_RECORD_INFOS_SERVICES);
+        websocketResult.setPayload(hashMap);
+        WebSocketUtils.publishAll(JacksonUtils.toJSONString(websocketResult));
+
+    }
+
+    @Override
+    public VideoIntegrityCheckDto transform(Object data) {
+        VideoIntegrityCheckReq req = JSON.parseObject(JSON.toJSONString(data), VideoIntegrityCheckReq.class);
+        VideoIntegrityCheckDto dst = new VideoIntegrityCheckDto();
+        dst.setEquipmentCode(req.getDvsCode());
+        dst.setChannelCode(req.getIndex());
+        dst.setEquipmentCode(req.getDvsCode());
+        dst.setRecordDate(req.getRecordDate());
+        dst.setLoseSpan(req.getLoseSpan());
+        dst.setCheckSpan(req.getCheckSpan());
+
+        return dst;
+    }
+
+    @Override
     @Transactional(rollbackFor = Exception.class)
     public ReceiveErrorDto saveData(VideoIntegrityCheckDto videoIntegrityCheckDto, String msgId) {
         String msg = this.checkParam(videoIntegrityCheckDto);

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/domain/dto/videoIntegrityCheck/VideoIntegrityCheckReq.java

@@ -0,0 +1,44 @@
+package com.xunmei.mediator.domain.dto.videoIntegrityCheck;
+
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Builder;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+@Builder
+public class VideoIntegrityCheckReq {
+
+    @ApiModelProperty(value = "产品名称")
+    private String productName;
+
+    @ApiModelProperty(value = "设备名称")
+    private String deviceName;
+
+    @ApiModelProperty(value = "录像机编号")
+    private String dvsCode;
+
+    @ApiModelProperty(value = "通道编号")
+    private String index;
+
+    @ApiModelProperty(value = "0: 定时 |  1：移动侦测")
+    private String recordType;
+
+    @ApiModelProperty(value = "索引分析日期")
+    private String recordDate;
+
+    @ApiModelProperty(value = "录像计划模版")
+    private List<String> checkSpan;
+
+    @ApiModelProperty(value = "原始录像索引")
+    private List<String> recordSpan;
+
+    @ApiModelProperty(value = "录像丢失内容")
+    private List<String> loseSpan;
+
+
+
+
+
+}

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/websocket/constant/WebSocketConstants.java

@@ -30,4 +30,63 @@ public interface WebSocketConstants {
      */
     String PONG = "pong";
 
+
+    // ------------------ IOT提供的服务能力开始 ------------------
+
+    /**
+     * 服务端主动请求 固定字段
+     */
+    String SERVICE="service";
+
+    /**
+     * 客户端主动上报 固定字段
+     */
+    String EVENT="event";
+
+
+    /**
+     * 客户端主动上报 固定字段
+     */
+    String ARGS="args";
+
+    /**
+     * IOT主动上报录像完整性事件
+     */
+    String RECORD_INFOS = "recordInfos";
+
+    /**
+     * 服务端主动请求 录像完整性 服务名称
+     */
+    String GET_RECORD_INFOS_SERVICES = "getRecordInfos";
+
+    /**
+     * 获取Dvs状态
+     */
+    String GET_DVS_STATUS_SERVICES = "getDvsStatus";
+    /**
+     * 获取通道详细信息列表,包含摄像机状态
+     */
+    String GET_DVS_CHANNEL_INFOS_SERVICES = "getDvsChannelInfos";
+    /**
+     * 获取磁盘信息，基础信息+状态
+     */
+    String GET_DISK_INFOS_SERVICES = "getDiskInfos";
+    /**
+     * 获取录像计划
+     */
+    String GET_DVS_RECORD_PLANS_SERVICES = "getDvsRecordPlans";
+    /**
+     * 添加视频诊断任务
+     */
+    String EXECUTE_VQD_TASK_SERVICES = "excuteVQDTask";
+
+    /**
+     * 获取该产品下的已添加的所有Dvs设备信息
+     */
+    String GET_DVS_INFOS_SERVICES = "getDvsInfos";
+
+
+
+
+    // ------------------ IOT提供的服务能力结束 ------------------
 }

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/websocket/dto/WebsocketResult.java

@@ -1,10 +1,12 @@
 package com.xunmei.mediator.websocket.dto;
 
+import cn.hutool.core.date.DateUtil;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
 import java.io.Serializable;
+import java.util.Date;
 
 /**
  * @author gaoxiong
@@ -36,5 +38,14 @@ public class WebsocketResult implements Serializable {
      */
     private Object payload;
 
+
+    public String invokeTopic(String productName,String deviceName) {
+       return String.format("things/%s/%s/service/invoke",productName,deviceName);
+    }
+
+    public void  setTimestamp(Date date) {
+        this.timestamp=DateUtil.format(date,"yyyy-MM-dd'T'HH:mm:ss");
+    }
+
 }
 

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/websocket/handler/SocWebSocketHandler.java

@@ -1,25 +1,26 @@
 package com.xunmei.mediator.websocket.handler;
 
-import cn.hutool.core.date.DateUtil;
-import com.alibaba.fastjson.JSON;
+import cn.hutool.core.util.ObjectUtil;
+import com.alibaba.fastjson2.JSON;
 import com.xunmei.common.core.domain.iot.domain.IotServerInfo;
 import com.xunmei.common.core.enums.iot.IotServerConnectStatus;
+import com.xunmei.common.core.utils.JacksonUtils;
 import com.xunmei.common.core.utils.StringUtils;
-import com.xunmei.common.redis.utils.RedisUtils;
 import com.xunmei.mediator.api.server.service.IotServerInfoService;
+import com.xunmei.mediator.websocket.constant.WebSocketConstants;
 import com.xunmei.mediator.websocket.dto.WebsocketResult;
 import com.xunmei.mediator.websocket.enums.WebsocketStatus;
 import com.xunmei.mediator.websocket.holder.WebSocketSessionHolder;
+import com.xunmei.mediator.websocket.service.RouterService;
+import com.xunmei.mediator.websocket.service.RouterServiceHandler;
 import com.xunmei.mediator.websocket.utils.WebSocketUtils;
 import lombok.extern.slf4j.Slf4j;
-import org.redisson.api.RedissonClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.*;
 import org.springframework.web.socket.handler.AbstractWebSocketHandler;
 
 import java.io.IOException;
-import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.Date;
 import java.util.HashMap;
@@ -57,7 +58,7 @@ public class SocWebSocketHandler extends AbstractWebSocketHandler {
                 map.put("statusCode",WebsocketStatus.SUCCESS.getStatusCode());
             }
             WebsocketResult register = createWebsocketResult(null, "register" , map);
-            WebSocketUtils.sendMessage(session, JSON.toJSONString(register));
+            WebSocketUtils.sendMessage(session, JacksonUtils.toJSONString(register));
             log.info("[建立注册连接] sessionId: {},registerCode:{}", session.getId(), registerCode);
             session.close();
             return;
@@ -74,7 +75,7 @@ public class SocWebSocketHandler extends AbstractWebSocketHandler {
             Map<String,Object> map = new HashMap<>();
             map.put("statusCode",WebsocketStatus.SUCCESS.getStatusCode());
             WebsocketResult result = createWebsocketResult(null, "login" , map);
-            WebSocketUtils.sendMessage(session, JSON.toJSONString(result));
+            WebSocketUtils.sendMessage(session, JacksonUtils.toJSONString(result));
             log.info("[建立连接] sessionId: {},token:{}", session.getId(), token);
             return;
         }
@@ -86,7 +87,7 @@ public class SocWebSocketHandler extends AbstractWebSocketHandler {
         result.setId(id);
         result.setTopic(topic);
         //当前时间转换为 格式：2024-07-02T14:17:32
-        result.setTimestamp(DateUtil.format(new Date(),"yyyy-MM-dd'T'HH:mm:ss"));
+        result.setTimestamp(new Date());
         result.setPayload(object);
         return result;
     }
@@ -102,7 +103,31 @@ public class SocWebSocketHandler extends AbstractWebSocketHandler {
     protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
         // 从WebSocket会话中获取登录用户信息
         WebSocketSessionHolder.updateToken(session);
+        String payload = message.getPayload();
         log.info("接收到消息：{}",message.getPayload());
+        if (payload.isEmpty()){
+            return;
+        }
+        try {
+            WebsocketResult websocketResult = JSON.parseObject(payload, WebsocketResult.class);
+            Object obj = websocketResult.getPayload();
+            if (ObjectUtil.isNotEmpty(obj)){
+                Map map = JSON.parseObject(obj.toString(), Map.class);
+                String key = (String) map.get(WebSocketConstants.EVENT);
+                String args = (String) map.get(WebSocketConstants.ARGS);
+                if (ObjectUtil.hasEmpty(key,args)){
+                    log.error("消息内容格式错误：{}",message.getPayload());
+                    return;
+                }
+                RouterService routeService = RouterServiceHandler.getRouteService(key);
+                routeService.execute(args);
+
+            }
+        } catch (Exception e) {
+           log.error("转换消息内容时出错：{}",e);
+        }
+
+
     }
 
     /**

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/websocket/service/RouterService.java

@@ -0,0 +1,9 @@
+package com.xunmei.mediator.websocket.service;
+
+public interface RouterService {
+
+    String routerKey();
+
+    Object execute(Object obj);
+
+}

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/websocket/service/RouterServiceHandler.java

@@ -0,0 +1,27 @@
+package com.xunmei.mediator.websocket.service;
+
+import cn.hutool.extra.spring.SpringUtil;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+
+public class RouterServiceHandler {
+    private static final Map<String, RouterService> CONCURRENT_HASH_MAP = new ConcurrentHashMap<>();
+
+    static {
+
+        SpringUtil.getBeansOfType(RouterService.class).forEach((k, v) -> {
+            CONCURRENT_HASH_MAP.put(v.routerKey(), v);
+        });
+    }
+
+    public static RouterService getRouteService(String routerKey) {
+        for (String key : CONCURRENT_HASH_MAP.keySet()) {
+            if (key.contains(routerKey)) {
+                return CONCURRENT_HASH_MAP.get(key);
+            }
+        }
+        throw new RuntimeException("未找到对应的处理类");
+    }
+}