优化通道更新逻辑

648540858

2021-12-28 726963ba772dde3e9bfaf52c7c398c0050ca1859

优化通道更新逻辑

 src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java

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

import java.util.Date;
import java.util.List;

public class CatalogData {
    private int total;
    private List<DeviceChannel> channelList;
    private Date lastTime;
    private Device device;

    public int getTotal() {
        return total;
    }

    public void setTotal(int total) {
        this.total = total;
    }

    public List<DeviceChannel> getChannelList() {
        return channelList;
    }

    public void setChannelList(List<DeviceChannel> channelList) {
        this.channelList = channelList;
    }

    public Date getLastTime() {
        return lastTime;
    }

    public void setLastTime(Date lastTime) {
        this.lastTime = lastTime;
    }

    public Device getDevice() {
        return device;
    }

    public void setDevice(Device device) {
        this.device = device;
    }
}

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

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

import com.genersoft.iot.vmp.gb28181.bean.CatalogData;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
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.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class CatalogDataCatch {

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

    @Autowired
    private DeferredResultHolder deferredResultHolder;

    @Autowired
    private IVideoManagerStorager storager;

    public void put(String key, int total, Device device, List<DeviceChannel> deviceChannelList) {
        CatalogData catalogData = data.get(key);
        if (catalogData == null) {
            catalogData = new CatalogData();
            catalogData.setTotal(total);
            catalogData.setDevice(device);
            catalogData.setChannelList(new ArrayList<>());
            data.put(key, catalogData);
        }
        catalogData.getChannelList().addAll(deviceChannelList);
        catalogData.setLastTime(new Date(System.currentTimeMillis()));
    }

    public List<DeviceChannel> get(String key) {
        CatalogData catalogData = data.get(key);
        if (catalogData == null) return null;
        return catalogData.getChannelList();
    }

    public void del(String key) {
        data.remove(key);
    }

    @Scheduled(fixedRate = 5 * 1000)   //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时
    private void timerTask(){
        Set<String> keys = data.keySet();
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(new Date());
        calendar.set(Calendar.SECOND, calendar.get(Calendar.SECOND) - 5);
        for (String key : keys) {
            CatalogData catalogData = data.get(key);
            if (catalogData.getLastTime().before(calendar.getTime())) {

                storager.resetChannels(catalogData.getDevice().getDeviceId(), catalogData.getChannelList());
                RequestMessage msg = new RequestMessage();
                msg.setKey(key);
                WVPResult<Object> result = new WVPResult<>();
                result.setCode(0);
                result.setMsg("更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条");
                result.setData(catalogData.getDevice());
                msg.setData(result);
                deferredResultHolder.invokeAllResult(msg);
                data.remove(key);
            }
        }
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java

@@ -7,6 +7,7 @@
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.CatalogDataCatch;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
@@ -14,6 +15,7 @@
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.slf4j.Logger;
@@ -27,7 +29,9 @@
import javax.sip.SipException;
import javax.sip.message.Response;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

@@ -39,6 +43,8 @@
    private Logger logger = LoggerFactory.getLogger(CatalogResponseMessageHandler.class);
    private final String cmdType = "Catalog";

    private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Autowired
    private ResponseMessageHandler responseMessageHandler;

@@ -47,6 +53,9 @@

    @Autowired
    private DeferredResultHolder deferredResultHolder;

    @Autowired
    private CatalogDataCatch catalogDataCatch;

    @Autowired
    private DeviceOffLineDetector offLineDetector;
@@ -69,6 +78,12 @@
        try {
            rootElement = getRootElement(evt, device.getCharset());
            Element deviceListElement = rootElement.element("DeviceList");
            Element sumNumElement = rootElement.element("SumNum");
            if (sumNumElement == null || deviceListElement == null) {
                responseAck(evt, Response.BAD_REQUEST, "xml error");
                return;
            }
            int sumNum = Integer.parseInt(sumNumElement.getText());
            Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
            if (deviceListIterator != null) {
                List<DeviceChannel> channelList = new ArrayList<>();
@@ -86,6 +101,10 @@
                    String status = statusElement != null ? statusElement.getText().toString() : "ON";
                    DeviceChannel deviceChannel = new DeviceChannel();
                    deviceChannel.setName(channelName);
                    deviceChannel.setDeviceId(device.getDeviceId());
                    String now = this.format.format(new Date(System.currentTimeMillis()));
                    deviceChannel.setCreateTime(now);
                    deviceChannel.setUpdateTime(now);
                    deviceChannel.setChannelId(channelDeviceId);
                    // ONLINE OFFLINE  HIKVISION DS-7716N-E4 NVR的兼容性处理
                    if (status.equals("ON") || status.equals("On") || status.equals("ONLINE")) {
@@ -153,14 +172,28 @@
                        deviceChannel.setPTZType(Integer.parseInt(getText(itemDevice, "PTZType")));
                    }
                    deviceChannel.setHasAudio(true); // 默认含有音频，播放时再检查是否有音频及是否AAC
                    // TODO 修改为批量插入
                    channelList.add(deviceChannel);
                }
                storager.updateChannels(device.getDeviceId(), channelList);
                RequestMessage msg = new RequestMessage();
                msg.setKey(key);
                msg.setData(device);
                deferredResultHolder.invokeAllResult(msg);

                catalogDataCatch.put(key, sumNum, device, channelList);
                if (catalogDataCatch.get(key).size() == sumNum) {
                    // 数据已经完整接收
                    boolean resetChannelsResult = storager.resetChannels(device.getDeviceId(), catalogDataCatch.get(key));
                    RequestMessage msg = new RequestMessage();
                    msg.setKey(key);
                    WVPResult<Object> result = new WVPResult<>();
                    result.setCode(0);
                    result.setData(device);
                    if (resetChannelsResult) {
                        result.setMsg("更新成功，共" + sumNum + "条，已更新" + catalogDataCatch.get(key).size() + "条");
                    }else {
                        result.setMsg("接收成功，写入失败，共" + sumNum + "条，已接收" + catalogDataCatch.get(key).size() + "条");
                    }
                    msg.setData(result);
                    deferredResultHolder.invokeAllResult(msg);
                    catalogDataCatch.del(key);
                }

                // 回复200 OK
                responseAck(evt, Response.OK);
                if (offLineDetector.isOnline(device.getDeviceId())) {

 src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java

@@ -73,7 +73,6 @@
                    result.put(key, streamPushItem);
                }
            }

        }

        return new ArrayList<>(result.values());

 src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java

@@ -55,7 +55,7 @@
     * @param deviceId 设备id

     * @param channels 多个通道

     */

    public void updateChannels(String deviceId, List<DeviceChannel> channels);

    public int updateChannels(String deviceId, List<DeviceChannel> channels);



    /**

     * 开始播放

@@ -425,4 +425,10 @@
     */

    StreamProxyItem getStreamProxyByAppAndStream(String app, String streamId);



    /**

     * catlog查询结束后完全重写通道信息

     * @param deviceId

     * @param deviceChannelList

     */

    boolean resetChannels(String deviceId, List<DeviceChannel> deviceChannelList);

}


 src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java

@@ -133,7 +133,7 @@
            "'${item.streamId}', ${item.longitude}, ${item.latitude},'${item.createTime}', '${item.updateTime}')" +
            "</foreach> " +
            "</script>")
    void batchAdd(List<DeviceChannel> addChannels);
    int batchAdd(List<DeviceChannel> addChannels);

    @Update({"<script>" +
            "<foreach collection='updateChannels' item='item' separator=';'>" +
@@ -167,7 +167,7 @@
            "WHERE deviceId=#{item.deviceId} AND channelId=#{item.channelId}"+
            "</foreach>" +
            "</script>"})
    void batchUpdate(List<DeviceChannel> updateChannels);
    int batchUpdate(List<DeviceChannel> updateChannels);

    @Select(value = {" <script>" +
            "SELECT * FROM ( "+

 src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java

@@ -156,7 +156,7 @@
    }

    @Override
    public void updateChannels(String deviceId, List<DeviceChannel> channels) {
    public int updateChannels(String deviceId, List<DeviceChannel> channels) {
        List<DeviceChannel> addChannels = new ArrayList<>();
        List<DeviceChannel> updateChannels = new ArrayList<>();
        HashMap<String, DeviceChannel> channelsInStore = new HashMap<>();
@@ -210,13 +210,47 @@
                        if (i + limitCount > updateChannels.size()) {
                            toIndex = updateChannels.size();
                        }
                        deviceChannelMapper.batchAdd(updateChannels.subList(i, toIndex));
                        deviceChannelMapper.batchUpdate(updateChannels.subList(i, toIndex));
                    }
                }else {
                    deviceChannelMapper.batchUpdate(updateChannels);
                }
            }
        }
        return addChannels.size() + updateChannels.size();
    }

    @Override
    public boolean resetChannels(String deviceId, List<DeviceChannel> deviceChannelList) {
        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
        try {
            int cleanChannelsResult = deviceChannelMapper.cleanChannelsByDeviceId(deviceId);
            int limitCount = 300;
            boolean result = cleanChannelsResult <0;
            if (!result && deviceChannelList.size() > 0) {
                if (deviceChannelList.size() > limitCount) {
                    for (int i = 0; i < deviceChannelList.size(); i += limitCount) {
                        int toIndex = i + limitCount;
                        if (i + limitCount > deviceChannelList.size()) {
                            toIndex = deviceChannelList.size();
                        }
                        result = result || deviceChannelMapper.batchAdd(deviceChannelList.subList(i, toIndex)) < 0;
                    }
                }else {
                    result = result || deviceChannelMapper.batchAdd(deviceChannelList) < 0;
                }
            }
            if (result) {
                //事务回滚
                dataSourceTransactionManager.rollback(transactionStatus);
            }
            dataSourceTransactionManager.commit(transactionStatus);     //手动提交
            return true;
        }catch (Exception e) {
            dataSourceTransactionManager.rollback(transactionStatus);
            return false;
        }

    }

    @Override
@@ -711,7 +745,6 @@
                    if (streamProxyItems == null) {
                        platformGbStreamMapper.add(streamPushItem);
                    }

                }
            }
        }

 web_src/package-lock.json

Diff too large

 web_src/src/components/DeviceList.vue

@@ -196,16 +196,16 @@
                    url: '/api/device/query/devices/' + itemData.deviceId + '/sync'
                }).then(function(res) {
                    console.log("刷新设备结果："+JSON.stringify(res));
                    if (!res.data.deviceId) {
                    if (res.data.code !==0) {
                        that.$message({
                            showClose: true,
                            message: res.data,
                            message: res.data.msg,
                            type: 'error'
                        });
                    }else{
                        that.$message({
                            showClose: true,
                            message: '请求成功',
                            message: res.data.msg,
                            type: 'success'
                        });
                    }

 web_src/src/components/PushVideoList.vue

@@ -35,7 +35,7 @@
                    <el-table-column label="操作" width="360" align="center" fixed="right">
                        <template slot-scope="scope">
                            <el-button-group>
                                <el-button size="mini" icon="el-icon-video-play" v-if="scope.row.status" @click="playPuhsh(scope.row)">播放</el-button>
                                <el-button size="mini" icon="el-icon-video-play" v-if="(scope.row.status == false && scope.row.gbId == null) || scope.row.status" @click="playPuhsh(scope.row)">播放</el-button>
                                <el-button size="mini" icon="el-icon-switch-button" type="danger" @click="stopPuhsh(scope.row)">移除</el-button>
                                <el-button size="mini" icon="el-icon-position" type="primary" v-if="!!!scope.row.gbId" @click="addToGB(scope.row)">加入国标</el-button>
                                <el-button size="mini" icon="el-icon-position" type="primary" v-if="!!scope.row.gbId" @click="removeFromGB(scope.row)">移出国标</el-button>

 web_src/src/components/dialog/chooseChannelForGb.vue

@@ -71,7 +71,7 @@
            channelType: "",
            online: "",
            choosed: "",
            currentPage: 0,
            currentPage: 1,
            count: 10,
            total: 0,
            eventEnanle: false
@@ -82,7 +82,7 @@
        platformId(newData, oldData){
            console.log(newData)
            this.initData()
            

        },
    },
    methods: {
@@ -126,7 +126,7 @@
                        delete that.gbChoosechannel[key]
                    }
                }
                 

                 var oldKeys = Object.keys(that.gbChoosechannel);
                if (oldKeys.length > 0) {
                    for (let i = 0; i < oldKeys.length; i++) {
@@ -134,7 +134,7 @@
                        delData.push(that.gbChoosechannel[key])
                    }
                }
                

            }else{
                var oldKeys = Object.keys(that.gbChoosechannel);
                if (oldKeys.length > 0) {
@@ -184,7 +184,7 @@

            this.$axios({
                    method:"get",
                    url:`/api/platform/channel_list`, 
                    url:`/api/platform/channel_list`,
                    params: {
                        page: that.currentPage,
                        count: that.count,
@@ -211,7 +211,7 @@
                                that.$refs.gbChannelsTable.toggleRowSelection(row, true);
                                chooseGBS.push(row)
                                that.gbChoosechannel[row.deviceId+ "_" + row.channelId] = row;
                               

                            }
                        }
                         that.eventEnanle = true;

 web_src/src/components/dialog/chooseChannelForStream.vue

@@ -49,7 +49,7 @@
            channelType: "",
            online: "",
            choosed: "",
            currentPage: 0,
            currentPage: 1,
            count: 10,
            total: 0,
            eventEnanle: false
@@ -60,7 +60,7 @@
        platformId(newData, oldData){
            console.log(newData)
            this.initData()
            

        },
    },
    methods: {
@@ -104,7 +104,7 @@
                        delete that.gbChoosechannel[key]
                    }
                }
                 

                 var oldKeys = Object.keys(that.gbChoosechannel);
                if (oldKeys.length > 0) {
                    for (let i = 0; i < oldKeys.length; i++) {
@@ -112,7 +112,7 @@
                        delData.push(that.gbChoosechannel[key])
                    }
                }
                

            }else{
                var oldKeys = Object.keys(that.gbChoosechannel);
                if (oldKeys.length > 0) {
@@ -191,7 +191,7 @@
                                that.$refs.gbStreamsTable.toggleRowSelection(row, true);
                                chooseGBS.push(row)
                                that.gbChoosechannel[row.app+ "_" + row.stream] = row;
                               

                            }
                        }
                         that.eventEnanle = true;