推流导入支持添加平台信息与目录信息

648540858
1 parent d07a5680
Showing 10 changed files with 145 additions and 30 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEvent.java
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushUploadFileHandler.java
src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformGbStreamMapper.java
src/main/java/com/genersoft/iot/vmp/vmanager/bean/StreamPushExcelDto.java
web_src/src/components/PushVideoList.vue
@@ -131,7 +131,7 @@ public class EventPublisher {
 	}
 	@Async
-	public void catalogEventPublishForStream(String platformId, List<GbStream> gbStreams, String type) {
+	public void catalogEventPublishForStream(String platformId, GbStream[] gbStreams, String type) {
 		CatalogEvent outEvent = new CatalogEvent(this);
 		outEvent.setGbStreams(gbStreams);
 		outEvent.setType(type);
@@ -141,8 +141,7 @@ public class EventPublisher {
 	@Async
 	public void catalogEventPublishForStream(String platformId, GbStream gbStream, String type) {
-		List<GbStream> gbStreamList = new ArrayList<>();
-		gbStreamList.add(gbStream);
-		catalogEventPublishForStream(platformId, gbStreamList, type);
+		GbStream[] gbStreams = {gbStream};
+		catalogEventPublishForStream(platformId, gbStreams, type);
 	}
 }
@@ -20,7 +20,7 @@ public class CatalogEvent  extends ApplicationEvent {
     public static final String UPDATE = "UPDATE";       // 更新
     private List<DeviceChannel> deviceChannels;
-    private List<GbStream> gbStreams;
+    private GbStream[] gbStreams;
     private String type;
     private String platformId;
@@ -48,11 +48,11 @@ public class CatalogEvent  extends ApplicationEvent {
         this.platformId = platformId;
     }
-    public List<GbStream> getGbStreams() {
+    public GbStream[] getGbStreams() {
         return gbStreams;
     }
-    public void setGbStreams(List<GbStream> gbStreams) {
+    public void setGbStreams(GbStream[] gbStreams) {
         this.gbStreams = gbStreams;
     }
 }
@@ -94,7 +94,7 @@ public class CatalogEventLister implements ApplicationListener&lt;CatalogEvent&gt; {
                     if (event.getDeviceChannels() != null) {
                         deviceChannelList.addAll(event.getDeviceChannels());
                     }
-                    if (event.getGbStreams() != null && event.getGbStreams().size() > 0){
+                    if (event.getGbStreams() != null && event.getGbStreams().length > 0){
                         for (GbStream gbStream : event.getGbStreams()) {
                             DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), parentPlatform.getDeviceGBId());
                             deviceChannelList.add(deviceChannelByStream);
@@ -134,7 +134,7 @@ public class CatalogEventLister implements ApplicationListener&lt;CatalogEvent&gt; {
                      if (event.getDeviceChannels() != null) {
                          deviceChannelList.addAll(event.getDeviceChannels());
                      }
-                    if (event.getGbStreams() != null && event.getGbStreams().size() > 0){
+                    if (event.getGbStreams() != null && event.getGbStreams().length > 0){
                         for (GbStream gbStream : event.getGbStreams()) {
                             DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), parentPlatform.getDeviceGBId());
                             deviceChannelList.add(deviceChannelByStream);
@@ -371,7 +371,7 @@ public class ZLMHttpHookListener {
 								}
 							}
 							if (gbStreams.size() > 0) {
-								eventPublisher.catalogEventPublishForStream(null, gbStreams, CatalogEvent.ON);
+								eventPublisher.catalogEventPublishForStream(null, gbStreams.toArray(new GbStream[0]), CatalogEvent.ON);
 							}
 						}else {
@@ -68,4 +68,6 @@ public interface IStreamPushService {
     void batchAdd(List<StreamPushItem> streamPushExcelDtoList);
     boolean batchStop(List<GbStream> streamPushItems);
+
+    void batchAddForUpload(String platformId, String catalogId, List<StreamPushItem> streamPushItems);
 }
@@ -8,6 +8,7 @@ import com.genersoft.iot.vmp.conf.UserSetup;
 import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
 import com.genersoft.iot.vmp.gb28181.bean.GbStream;
 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
+import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog;
 import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
 import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
 import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
@@ -18,10 +19,7 @@ import com.genersoft.iot.vmp.service.IGbStreamService;
 import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.IStreamPushService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
-import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
-import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
-import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
-import com.genersoft.iot.vmp.storager.dao.StreamPushMapper;
+import com.genersoft.iot.vmp.storager.dao.*;
 import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
@@ -44,6 +42,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
     private ParentPlatformMapper parentPlatformMapper;
     @Autowired
+    private PlatformCatalogMapper platformCatalogMapper;
+
+    @Autowired
     private PlatformGbStreamMapper platformGbStreamMapper;
     @Autowired
@@ -95,13 +96,12 @@ public class StreamPushServiceImpl implements IStreamPushService {
         streamPushItem.setMediaServerId(item.getMediaServerId());
         streamPushItem.setStream(item.getStream());
         streamPushItem.setAliveSecond(item.getAliveSecond());
-        streamPushItem.setCreateStamp(item.getCreateStamp());
         streamPushItem.setOriginSock(item.getOriginSock());
         streamPushItem.setTotalReaderCount(item.getTotalReaderCount());
         streamPushItem.setOriginType(item.getOriginType());
         streamPushItem.setOriginTypeStr(item.getOriginTypeStr());
         streamPushItem.setOriginUrl(item.getOriginUrl());
-        streamPushItem.setCreateStamp(item.getCreateStamp());
+        streamPushItem.setCreateStamp(item.getCreateStamp() * 1000);
         streamPushItem.setAliveSecond(item.getAliveSecond());
         streamPushItem.setStatus(true);
         streamPushItem.setStreamType("push");
@@ -359,6 +359,27 @@ public class StreamPushServiceImpl implements IStreamPushService {
     }
     @Override
+    public void batchAddForUpload(String platformId, String catalogId, List<StreamPushItem> streamPushItems) {
+        streamPushMapper.addAll(streamPushItems);
+        gbStreamMapper.batchAdd(streamPushItems);
+        if (platformId != null) {
+            ParentPlatform platform = parentPlatformMapper.getParentPlatByServerGBId(platformId);
+            if (platform != null) {
+                if (catalogId == null) {
+                    catalogId = platform.getCatalogId();
+                }else {
+                    PlatformCatalog catalog = platformCatalogMapper.select(catalogId);
+                    if (catalog == null) {
+                        return;
+                    }
+                }
+                platformGbStreamMapper.batchAdd(platformId, catalogId, streamPushItems);
+                eventPublisher.catalogEventPublishForStream(platformId, streamPushItems.toArray(new GbStream[0]), CatalogEvent.ADD);
+            }
+        }
+    }
+
+    @Override
     public boolean batchStop(List<GbStream> gbStreams) {
         if (gbStreams == null || gbStreams.size() == 0) {
             return false;
@@ -7,10 +7,7 @@ import com.genersoft.iot.vmp.service.IStreamPushService;
 import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto;
 import org.springframework.util.StringUtils;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPushExcelDto> {
@@ -18,10 +15,13 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener&lt;StreamPus
     private IStreamPushService pushService;
     private String defaultMediaServerId;
     private List<StreamPushItem> streamPushItems = new ArrayList<>();
+    private Map<String, UploadData> streamPushItemsForPlatform = new HashMap<>();
     private Set<String> streamPushStreamSet = new HashSet<>();
     private Set<String> streamPushGBSet = new HashSet<>();
     private List<String> errorStreamList = new ArrayList<>();
     private List<String> errorGBList = new ArrayList<>();
+    // 读取数量计数器
+    private int loadedSize = 0;
     public StreamPushUploadFileHandler(IStreamPushService pushService, String defaultMediaServerId, ErrorDataHandler errorDataHandler) {
         this.pushService = pushService;
@@ -33,6 +33,16 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener&lt;StreamPus
         void handle(List<String> streams, List<String> gbId);
     }
+    private class UploadData{
+        public String platformId;
+        public Map<String, List<StreamPushItem>> catalogData = new HashMap<>();
+        public List<StreamPushItem> streamPushItems = new ArrayList<>();
+
+        public UploadData(String platformId) {
+            this.platformId = platformId;
+        }
+    }
+
     @Override
     public void invoke(StreamPushExcelDto streamPushExcelDto, AnalysisContext analysisContext) {
         if (StringUtils.isEmpty(streamPushExcelDto.getApp())
@@ -43,10 +53,10 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener&lt;StreamPus
         if (streamPushGBSet.contains(streamPushExcelDto.getGbId())) {
             errorGBList.add(streamPushExcelDto.getGbId());
         }
-        if (streamPushStreamSet.contains(streamPushExcelDto.getApp() + streamPushExcelDto.getStream())) {
+        if (streamPushStreamSet.contains(streamPushExcelDto.getApp() + streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId())) {
             errorStreamList.add(streamPushExcelDto.getApp() + "/" + streamPushExcelDto.getStream());
         }
-        if (streamPushGBSet.contains(streamPushExcelDto.getGbId()) || streamPushStreamSet.contains(streamPushExcelDto.getApp() + streamPushExcelDto.getStream())) {
+        if (streamPushGBSet.contains(streamPushExcelDto.getGbId()) || streamPushStreamSet.contains(streamPushExcelDto.getApp() + streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId())) {
             return;
         }
@@ -62,24 +72,69 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener&lt;StreamPus
         streamPushItem.setOriginType(2);
         streamPushItem.setOriginTypeStr("rtsp_push");
         streamPushItem.setTotalReaderCount("0");
-        streamPushItems.add(streamPushItem);
+        streamPushItem.setPlatformId(streamPushExcelDto.getPlatformId());
+        streamPushItem.setCatalogId(streamPushExcelDto.getCatalogId());
+        if (StringUtils.isEmpty(streamPushExcelDto.getPlatformId())) {
+            streamPushItems.add(streamPushItem);
+        }else {
+            UploadData uploadData = streamPushItemsForPlatform.get(streamPushExcelDto.getPlatformId());
+            if (uploadData == null) {
+                uploadData = new UploadData(streamPushExcelDto.getPlatformId());
+                streamPushItemsForPlatform.put(streamPushExcelDto.getPlatformId(), uploadData);
+            }
+            if (!StringUtils.isEmpty(streamPushExcelDto.getCatalogId())) {
+                List<StreamPushItem> streamPushItems = uploadData.catalogData.get(streamPushExcelDto.getCatalogId());
+                if (streamPushItems == null) {
+                    streamPushItems = new ArrayList<>();
+                    uploadData.catalogData.put(streamPushExcelDto.getCatalogId(), streamPushItems);
+                }
+                streamPushItems.add(streamPushItem);
+            }else {
+                uploadData.streamPushItems.add(streamPushItem);
+            }
+
+        }
+
         streamPushGBSet.add(streamPushExcelDto.getGbId());
         streamPushStreamSet.add(streamPushExcelDto.getApp()+streamPushExcelDto.getStream());
-        if (streamPushItems.size() > 300) {
-            pushService.batchAdd(streamPushItems);
-            // 存储完成清理 list
+        loadedSize ++;
+        if (loadedSize > 1000) {
+            saveData();
             streamPushItems.clear();
+            streamPushItemsForPlatform.clear();
+            loadedSize = 0;
         }
+
     }
     @Override
     public void doAfterAllAnalysed(AnalysisContext analysisContext) {
         // 这里也要保存数据，确保最后遗留的数据也存储到数据库
-        if (streamPushItems.size() > 0) {
-            pushService.batchAdd(streamPushItems);
-        }
+        saveData();
         streamPushGBSet.clear();
         streamPushStreamSet.clear();
         errorDataHandler.handle(errorStreamList, errorGBList);
     }
+
+    private void saveData(){
+        if (streamPushItems.size() > 0) {
+            pushService.batchAddForUpload(null, null, streamPushItems);
+        }
+        // 处理已分配到平台的流
+        if (streamPushItemsForPlatform.size() > 0){
+            for (String platformId : streamPushItemsForPlatform.keySet()) {
+                UploadData uploadData = streamPushItemsForPlatform.get(platformId);
+                if (uploadData.streamPushItems.size() > 0) {
+                    pushService.batchAddForUpload(platformId, null, uploadData.streamPushItems);
+                }
+                if (uploadData.catalogData.size() > 0) {
+                    for (String catalogId : uploadData.catalogData.keySet()) {
+                        if (uploadData.catalogData.get(catalogId).size() > 0) {
+                            pushService.batchAddForUpload(platformId, catalogId, uploadData.catalogData.get(catalogId));
+                        }
+                    }
+                }
+            }
+        }
+    }
 }
@@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
 import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog;
 import com.genersoft.iot.vmp.gb28181.bean.PlatformGbStream;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
 import org.apache.ibatis.annotations.*;
 import org.springframework.stereotype.Repository;
@@ -19,6 +20,17 @@ public interface PlatformGbStreamMapper {
             "('${app}', '${stream}', '${platformId}', '${catalogId}')")
     int add(PlatformGbStream platformGbStream);
+
+    @Insert("<script> " +
+            "REPLACE into platform_gb_stream " +
+            "(app, stream, platformId, catalogId) " +
+            "values " +
+            "<foreach collection='streamPushItems' index='index' item='item' separator=','> " +
+            "('${item.app}', '${item.stream}', '${platformId}', '${catalogId}')" +
+            "</foreach> " +
+            "</script>")
+    int batchAdd(String platformId, String catalogId, List<StreamPushItem> streamPushItems);
+
     @Delete("DELETE FROM platform_gb_stream WHERE app=#{app} AND stream=#{stream}")
     int delByAppAndStream(String app, String stream);
@@ -82,4 +94,7 @@ public interface PlatformGbStreamMapper {
             "</foreach>" +
             "</script>")
     void delByGbStreams(List<GbStream> gbStreams);
+
+
+
 }
@@ -16,6 +16,12 @@ public class StreamPushExcelDto {
     @ExcelProperty("国标ID")
     private String gbId;
+    @ExcelProperty("平台ID")
+    private String platformId;
+
+    @ExcelProperty("目录ID")
+    private String catalogId;
+
     public String getName() {
         return name;
     }
@@ -47,4 +53,21 @@ public class StreamPushExcelDto {
     public void setGbId(String gbId) {
         this.gbId = gbId;
     }
+
+
+    public String getPlatformId() {
+        return platformId;
+    }
+
+    public void setPlatformId(String platformId) {
+        this.platformId = platformId;
+    }
+
+    public String getCatalogId() {
+        return catalogId;
+    }
+
+    public void setCatalogId(String catalogId) {
+        this.catalogId = catalogId;
+    }
 }
@@ -233,7 +233,7 @@
 			dateFormat: function(/** timestamp=0 **/) {
         let ts = arguments[0] || 0;
         let t,y,m,d,h,i,s;
-				t = ts ? new Date(ts*1000) : new Date();
+				t = ts ? new Date(ts) : new Date();
 				y = t.getFullYear();
 				m = t.getMonth()+1;
 				d = t.getDate();