修复WVP作为下级平台接受recordinfo指令上报上级平台的问题

gaofw189
1 parent 55ee6f5f
Showing 4 changed files with 45 additions and 12 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java
src/main/java/com/genersoft/iot/vmp/gb28181/event/record/RecordEndEventListener.java
src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
 package com.genersoft.iot.vmp.gb28181.bean;
+import lombok.Data;
+
 import java.time.Instant;
 import java.util.List;
@@ -8,6 +10,7 @@ import java.util.List;
  * @author: swwheihei
  * @date:   2020年5月8日 下午2:05:56     
  */
+@Data
 public class RecordInfo {
 	private String deviceId;
@@ -20,6 +23,8 @@ public class RecordInfo {
 	private int sumNum;
+	private int count;
+
 	private Instant lastTime;
 	private List<RecordItem> recordList;
 package com.genersoft.iot.vmp.gb28181.event.record;
 import com.genersoft.iot.vmp.gb28181.bean.RecordInfo;
+import com.genersoft.iot.vmp.utils.redis.RedisUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationListener;
 import org.springframework.stereotype.Component;
@@ -20,25 +22,46 @@ public class RecordEndEventListener implements ApplicationListener&lt;RecordEndEven
     private final static Logger logger = LoggerFactory.getLogger(RecordEndEventListener.class);
+    private Map<String, RecordEndEventHandler> handlerMap = new ConcurrentHashMap<>();
     public interface RecordEndEventHandler{
         void  handler(RecordInfo recordInfo);
     }
-    private Map<String, RecordEndEventHandler> handlerMap = new ConcurrentHashMap<>();
-
     @Override
     public void onApplicationEvent(RecordEndEvent event) {
-        logger.info("录像查询完成事件触发，deviceId：{}, channelId: {}, 录像数量{}条", event.getRecordInfo().getDeviceId(),
-                event.getRecordInfo().getChannelId(), event.getRecordInfo().getSumNum() );
+        String deviceId = event.getRecordInfo().getDeviceId();
+        String channelId = event.getRecordInfo().getChannelId();
+        int count = event.getRecordInfo().getCount();
+        int sumNum = event.getRecordInfo().getSumNum();
+        logger.info("录像查询完成事件触发，deviceId：{}, channelId: {}, 录像数量{}/{}条", event.getRecordInfo().getDeviceId(),
+                event.getRecordInfo().getChannelId(), count,sumNum);
         if (handlerMap.size() > 0) {
-            for (RecordEndEventHandler recordEndEventHandler : handlerMap.values()) {
-                recordEndEventHandler.handler(event.getRecordInfo());
+            RecordEndEventHandler handler = handlerMap.get(deviceId + channelId);
+            if (handler !=null){
+                handler.handler(event.getRecordInfo());
+                if (count ==sumNum){
+                    handlerMap.remove(deviceId + channelId);
+                }
             }
         }
-        handlerMap.clear();
     }
+    /**
+     * 添加
+     * @param device
+     * @param channelId
+     * @param recordEndEventHandler
+     */
     public void addEndEventHandler(String device, String channelId, RecordEndEventHandler recordEndEventHandler) {
         handlerMap.put(device + channelId, recordEndEventHandler);
     }
+    /**
+     * 添加
+     * @param device
+     * @param channelId
+     */
+    public void delEndEventHandler(String device, String channelId) {
+        handlerMap.remove(device + channelId);
+    }
+
 }
 package com.genersoft.iot.vmp.gb28181.session;
 import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.event.record.RecordEndEventListener;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@@ -23,14 +24,17 @@ public class RecordDataCatch {
     @Autowired
     private DeferredResultHolder deferredResultHolder;
+    @Autowired
+    private RecordEndEventListener recordEndEventListener;
-    public int put(String deviceId, String sn, int sumNum, List<RecordItem> recordItems) {
+    public int put(String deviceId,String channelId, String sn, int sumNum, List<RecordItem> recordItems) {
         String key = deviceId + sn;
         RecordInfo recordInfo = data.get(key);
         if (recordInfo == null) {
             recordInfo = new RecordInfo();
             recordInfo.setDeviceId(deviceId);
+            recordInfo.setChannelId(channelId);
             recordInfo.setSn(sn.trim());
             recordInfo.setSumNum(sumNum);
             recordInfo.setRecordList(Collections.synchronizedList(new ArrayList<>()));
@@ -67,6 +71,7 @@ public class RecordDataCatch {
                 msg.setKey(msgKey);
                 msg.setData(recordInfo);
                 deferredResultHolder.invokeAllResult(msg);
+                recordEndEventListener.delEndEventHandler(recordInfo.getDeviceId(),recordInfo.getChannelId());
                 data.remove(key);
             }
         }
@@ -102,8 +102,9 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent 
                         Element recordListElement = rootElementForCharset.element("RecordList");
                         if (recordListElement == null || sumNum == 0) {
                             logger.info("无录像数据");
+                            int count = recordDataCatch.put(take.getDevice().getDeviceId(),channelId, sn, sumNum, new ArrayList<>());
+                            recordInfo.setCount(count);
                             eventPublisher.recordEndEventPush(recordInfo);
-                            recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, new ArrayList<>());
                             releaseRequest(take.getDevice().getDeviceId(), sn);
                         } else {
                             Iterator<Element> recordListIterator = recordListElement.elementIterator();
@@ -137,12 +138,11 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent 
                                     recordList.add(record);
                                 }
                                 recordInfo.setRecordList(recordList);
+                                int count = recordDataCatch.put(take.getDevice().getDeviceId(),channelId, sn, sumNum, recordList);recordInfo.setCount(count);
+                                logger.info("[国标录像]， {}->{}: {}/{}", take.getDevice().getDeviceId(), sn, count, sumNum);
                                 // 发送消息，如果是上级查询此录像，则会通过这里通知给上级
                                 eventPublisher.recordEndEventPush(recordInfo);
-                                int count = recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, recordList);
-                                logger.info("[国标录像]， {}->{}: {}/{}", take.getDevice().getDeviceId(), sn, count, sumNum);
                             }
-
                             if (recordDataCatch.isComplete(take.getDevice().getDeviceId(), sn)){
                                 releaseRequest(take.getDevice().getDeviceId(), sn);
                             }