优化通道更新逻辑

648540858
1 parent 7241e0d2
Showing 7 changed files with 201 additions and 13 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java
src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java
+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;
+    }
+}
+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);
+            }
+        }
+    }
+}
@@ -7,6 +7,7 @@ import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
 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.IMessag
 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.RequestEvent;
 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 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
     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;
@@ -49,6 +55,9 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
     private DeferredResultHolder deferredResultHolder;
     @Autowired
+    private CatalogDataCatch catalogDataCatch;
+
+    @Autowired
     private DeviceOffLineDetector offLineDetector;
     @Autowired
@@ -69,6 +78,12 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
         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 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
                     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 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
                         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())) {
@@ -73,7 +73,6 @@ public class StreamPushServiceImpl implements IStreamPushService {
                     result.put(key, streamPushItem);
                 }
             }
-
         }
         return new ArrayList<>(result.values());
@@ -55,7 +55,7 @@ public interface IVideoManagerStorager {
 	 * @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 @@ public interface IVideoManagerStorager {
 	 */
     StreamProxyItem getStreamProxyByAppAndStream(String app, String streamId);
+	/**
+	 * catlog查询结束后完全重写通道信息
+	 * @param deviceId
+	 * @param deviceChannelList
+	 */
+	boolean resetChannels(String deviceId, List<DeviceChannel> deviceChannelList);
 }
@@ -133,7 +133,7 @@ public interface DeviceChannelMapper {
             "'${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 @@ public interface DeviceChannelMapper {
             "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 ( "+
@@ -156,7 +156,7 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
 	}
 	@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 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
 						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 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
 					if (streamProxyItems == null) {
 						platformGbStreamMapper.add(streamPushItem);
 					}
-
 				}
 			}
 		}