Merge branch '648540858:wvp-28181-2.0' into wvp-28181-2.0

hotcoffie

2022-05-17 1d753b48c023090430c2931fd8e3a45a382eddf8

 src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java

New file
@@ -0,0 +1,91 @@
package com.genersoft.iot.vmp.gb28181.session;

import com.genersoft.iot.vmp.gb28181.bean.*;
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;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.Instant;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author lin
 */
@Component
public class RecordDataCatch {

    public static Map<String, RecordInfo> data = new ConcurrentHashMap<>();

    @Autowired
    private DeferredResultHolder deferredResultHolder;


    public int put(String deviceId, 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.setSn(sn.trim());
            recordInfo.setSumNum(sumNum);
            recordInfo.setRecordList(Collections.synchronizedList(new ArrayList<>()));
            recordInfo.setLastTime(Instant.now());
            recordInfo.getRecordList().addAll(recordItems);
            data.put(key, recordInfo);
        }else {
            // 同一个设备的通道同步请求只考虑一个，其他的直接忽略
            if (!Objects.equals(sn.trim(), recordInfo.getSn())) {
                return 0;
            }
            recordInfo.getRecordList().addAll(recordItems);
            recordInfo.setLastTime(Instant.now());
        }
        return recordInfo.getRecordList().size();
    }

    @Scheduled(fixedRate = 5 * 1000)   //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时
    private void timerTask(){
        Set<String> keys = data.keySet();
        // 获取五秒前的时刻
        Instant instantBefore5S = Instant.now().minusMillis(TimeUnit.SECONDS.toMillis(5));
        for (String key : keys) {
            RecordInfo recordInfo = data.get(key);
            // 超过五秒收不到消息任务超时， 只更新这一部分数据
            if ( recordInfo.getLastTime().isBefore(instantBefore5S)) {
                // 处理录像数据， 返回给前端
                String msgKey = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + recordInfo.getDeviceId() + recordInfo.getSn();

                WVPResult<RecordInfo> wvpResult = new WVPResult<>();
                wvpResult.setCode(0);
                wvpResult.setMsg("success");
                // 对数据进行排序
                Collections.sort(recordInfo.getRecordList());
                wvpResult.setData(recordInfo);

                RequestMessage msg = new RequestMessage();
                msg.setKey(msgKey);
                msg.setData(wvpResult);
                deferredResultHolder.invokeAllResult(msg);
                data.remove(key);
            }
        }
    }

    public boolean isComplete(String deviceId, String sn) {
        RecordInfo recordInfo = data.get(deviceId + sn);
        return recordInfo != null && recordInfo.getRecordList().size() == recordInfo.getSumNum();
    }

    public RecordInfo getRecordInfo(String deviceId, String sn) {
        return data.get(deviceId + sn);
    }

    public void remove(String deviceId, String sn) {
        data.remove(deviceId + sn);
    }
}