From a42dda2bd3cc1cf8c20cc61e7ad9211eadecbaf3 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Thu, 24 Feb 2022 16:55:06 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=84=E8=8C=83=E6=95=B0=E6=8D=AE=E5=BA=93?= =?UTF-8?q?=EF=BC=8C=E6=B7=BB=E5=8A=A0=E5=BF=85=E8=A6=81=E7=BA=A6=E6=9D=9F?= =?UTF-8?q?=EF=BC=8C=E4=BC=98=E5=8C=96=E9=80=9A=E9=81=93=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E5=AF=BC=E5=85=A5=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pom.xml | 8 ++ sql/mysql.sql | 38 +++--- .../iot/vmp/gb28181/bean/GbStream.java | 11 +- .../vmp/gb28181/bean/PlatformGbStream.java | 6 +- .../iot/vmp/media/zlm/dto/StreamPushItem.java | 12 ++ .../iot/vmp/service/IStreamPushService.java | 3 +- .../vmp/service/impl/GbStreamServiceImpl.java | 4 +- .../service/impl/StreamPushServiceImpl.java | 69 +++++++--- .../impl/StreamPushUploadFileHandler.java | 119 ++++++++++-------- .../iot/vmp/storager/dao/GbStreamMapper.java | 15 +-- .../storager/dao/PlatformGbStreamMapper.java | 24 ++-- .../vmp/storager/dao/StreamPushMapper.java | 25 ++-- .../impl/VideoManagerStoragerImpl.java | 2 +- 13 files changed, 204 insertions(+), 132 deletions(-) diff --git a/pom.xml b/pom.xml index 9331de3b..f1ac4581 100644 --- a/pom.xml +++ b/pom.xml @@ -231,6 +231,14 @@ + + + com.google.guava + guava + 31.0.1-jre + + + org.springframework.boot spring-boot-starter-test diff --git a/sql/mysql.sql b/sql/mysql.sql index 2bb430ab..f2e358fc 100644 --- a/sql/mysql.sql +++ b/sql/mysql.sql @@ -53,7 +53,6 @@ CREATE TABLE `device` ( LOCK TABLES `device` WRITE; /*!40000 ALTER TABLE `device` DISABLE KEYS */; -INSERT INTO `device` VALUES (1,'34020000002000000005','DH-NVR5864-I','Dahua','DH-NVR5864-I','4.001.0000000.3,2020-10-22','UDP','UDP','1','2022-02-22 11:05:10','2022-02-22 11:42:10','192.168.1.19','2022-02-22 10:17:24','2022-02-22 11:42:10',5060,3600,0,'192.168.1.19:5060','gb2312'); /*!40000 ALTER TABLE `device` ENABLE KEYS */; UNLOCK TABLES; @@ -126,7 +125,8 @@ CREATE TABLE `device_channel` ( `createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `updateTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `subCount` int DEFAULT '0', - PRIMARY KEY (`channelId`,`deviceId`) USING BTREE + PRIMARY KEY (`channelId`,`deviceId`) USING BTREE, + UNIQUE KEY `device_channel_pk` (`channelId`,`deviceId`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; /*!40101 SET character_set_client = @saved_cs_client */; @@ -136,7 +136,6 @@ CREATE TABLE `device_channel` ( LOCK TABLES `device_channel` WRITE; /*!40000 ALTER TABLE `device_channel` DISABLE KEYS */; -INSERT INTO `device_channel` VALUES ('34020000001310000001','通道1','Dahua','DH-NVR5864-I','0','340200','','axy','34020000002000000005',0,1,'',0,0,NULL,'0','192.168.1.17',37777,'',0,1,0,0,NULL,'34020000002000000005','0',NULL,'','',0),('34020000001310000065','GB_Chn_065','Dahua','DH-NVR5864-I','0','340200','','axy','34020000002000000005',0,1,'',0,0,NULL,'0','',0,'',0,1,0,0,NULL,'34020000002000000005','0',NULL,'','',0); /*!40000 ALTER TABLE `device_channel` ENABLE KEYS */; UNLOCK TABLES; @@ -183,7 +182,7 @@ DROP TABLE IF EXISTS `gb_stream`; /*!40101 SET @saved_cs_client = @@character_set_client */; /*!50503 SET character_set_client = utf8mb4 */; CREATE TABLE `gb_stream` ( - `id` int NOT NULL AUTO_INCREMENT, + `gbStreamId` int NOT NULL AUTO_INCREMENT, `app` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `stream` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `gbId` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, @@ -194,10 +193,10 @@ CREATE TABLE `gb_stream` ( `mediaServerId` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL, `status` int DEFAULT NULL, `createStamp` bigint DEFAULT NULL, - PRIMARY KEY (`id`) USING BTREE, + PRIMARY KEY (`gbStreamId`) USING BTREE, UNIQUE KEY `app` (`app`,`stream`) USING BTREE, UNIQUE KEY `gbId` (`gbId`) USING BTREE -) ENGINE=InnoDB AUTO_INCREMENT=77 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; +) ENGINE=InnoDB AUTO_INCREMENT=291 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; /*!40101 SET character_set_client = @saved_cs_client */; -- @@ -206,7 +205,6 @@ CREATE TABLE `gb_stream` ( LOCK TABLES `gb_stream` WRITE; /*!40000 ALTER TABLE `gb_stream` DISABLE KEYS */; -INSERT INTO `gb_stream` VALUES (74,'1000','10000000_000','34020000002000000001','测试1',0,0,'push','FQ3TF8yT83wh5Wvz',0,1645500864142),(75,'1000','10000001_000','34020000012000000002','测试2',0,0,'push','FQ3TF8yT83wh5Wvz',0,1645500864142),(76,'1000','10000002_000','34020000022000000003','测试3',0,0,'push','FQ3TF8yT83wh5Wvz',0,1645500864142); /*!40000 ALTER TABLE `gb_stream` ENABLE KEYS */; UNLOCK TABLES; @@ -228,7 +226,7 @@ CREATE TABLE `log` ( `username` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, PRIMARY KEY (`id`) USING BTREE -) ENGINE=InnoDB AUTO_INCREMENT=206 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; +) ENGINE=InnoDB AUTO_INCREMENT=245 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; /*!40101 SET character_set_client = @saved_cs_client */; -- @@ -237,7 +235,6 @@ CREATE TABLE `log` ( LOCK TABLES `log` WRITE; /*!40000 ALTER TABLE `log` DISABLE KEYS */; -INSERT INTO `log` VALUES (163,'登录','GET','/api/user/login','127.0.0.1','200 OK',243,'admin','2022-02-22 10:18:15'),(164,'开始点播','GET','/api/play/start/34020000002000000005/34020000001310000001','127.0.0.1','200 OK',77,'admin','2022-02-22 10:18:35'),(165,'[设备查询] 同步设备通道','POST','/api/device/query/devices/34020000002000000005/sync','127.0.0.1','200 OK',10,'admin','2022-02-22 10:18:47'),(166,'开始点播','GET','/api/play/start/34020000002000000005/34020000001310000065','127.0.0.1','200 OK',19,'admin','2022-02-22 10:18:50'),(167,'停止点播','GET','/api/play/stop/34020000002000000005/34020000001310000065','127.0.0.1','200 OK',159,'admin','2022-02-22 10:19:08'),(168,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',305,'admin','2022-02-22 10:22:02'),(169,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',142,'admin','2022-02-22 10:22:28'),(170,'添加上级平台','POST','/api/platform/save','127.0.0.1','200 OK',11,'admin','2022-02-22 10:22:57'),(171,'登录','GET','/api/user/login','127.0.0.1','200 OK',255,'admin','2022-02-22 10:26:31'),(172,'移除代理','DELETE','/api/proxy/del','127.0.0.1','200 OK',40,'admin','2022-02-22 10:26:44'),(173,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',139,'admin','2022-02-22 10:26:51'),(174,'登录','GET','/api/user/login','127.0.0.1','200 OK',657,'admin','2022-02-22 10:28:21'),(175,'登录','GET','/api/user/login','127.0.0.1','200 OK',302,'admin','2022-02-22 11:04:24'),(176,'移除代理','DELETE','/api/proxy/del','127.0.0.1','200 OK',33,'admin','2022-02-22 11:04:34'),(177,'移除代理','DELETE','/api/proxy/del','127.0.0.1','200 OK',17,'admin','2022-02-22 11:04:35'),(178,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',186,'admin','2022-02-22 11:04:44'),(179,'登录','GET','/api/user/login','127.0.0.1','200 OK',258,'admin','2022-02-22 11:05:05'),(180,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',97,'admin','2022-02-22 11:05:51'),(181,'登录','GET','/api/user/login','127.0.0.1','200 OK',300,'admin','2022-02-22 11:11:30'),(182,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',145,'admin','2022-02-22 11:12:00'),(183,'登录','GET','/api/user/login','127.0.0.1','200 OK',264,'admin','2022-02-22 11:12:42'),(184,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',41793,'admin','2022-02-22 11:13:42'),(185,'登录','GET','/api/user/login','127.0.0.1','200 OK',245,'admin','2022-02-22 11:13:58'),(186,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',86,'admin','2022-02-22 11:14:11'),(187,'登录','GET','/api/user/login','127.0.0.1','200 OK',293,'admin','2022-02-22 11:31:52'),(188,'向上级平台添加国标通道','POST','/api/platform/update_channel_for_gb','127.0.0.1','200 OK',14,'admin','2022-02-22 11:34:55'),(189,'向上级平台添加国标通道','POST','/api/platform/update_channel_for_gb','127.0.0.1','200 OK',11,'admin','2022-02-22 11:35:02'),(190,'保存代理','POST','/api/proxy/save','127.0.0.1','200 OK',16,'admin','2022-02-22 11:37:00'),(191,'登录','GET','/api/user/login','127.0.0.1','200 OK',340,'admin','2022-02-22 11:38:15'),(192,'停用代理','GET','/api/proxy/stop','127.0.0.1','200 OK',16,'admin','2022-02-22 11:38:20'),(193,'停用代理','GET','/api/proxy/stop','127.0.0.1','200 OK',10,'admin','2022-02-22 11:38:28'),(194,'登录','GET','/api/user/login','127.0.0.1','200 OK',288,'admin','2022-02-22 11:39:34'),(195,'停用代理','GET','/api/proxy/stop','127.0.0.1','200 OK',29,'admin','2022-02-22 11:39:36'),(196,'启用代理','GET','/api/proxy/start','127.0.0.1','200 OK',594,'admin','2022-02-22 11:39:38'),(197,'停用代理','GET','/api/proxy/stop','127.0.0.1','200 OK',16,'admin','2022-02-22 11:39:43'),(198,'停用代理','GET','/api/proxy/stop','127.0.0.1','200 OK',27,'admin','2022-02-22 11:41:30'),(199,'停用代理','GET','/api/proxy/stop','127.0.0.1','200 OK',17,'admin','2022-02-22 11:41:42'),(200,'启用代理','GET','/api/proxy/start','127.0.0.1','200 OK',9,'admin','2022-02-22 11:41:44'),(201,'启用代理','GET','/api/proxy/start','127.0.0.1','200 OK',6,'admin','2022-02-22 11:41:47'),(202,'启用代理','GET','/api/proxy/start','127.0.0.1','200 OK',67,'admin','2022-02-22 11:41:48'),(203,'移除代理','DELETE','/api/proxy/del','127.0.0.1','200 OK',14,'admin','2022-02-22 11:41:51'),(204,'移除代理','DELETE','/api/proxy/del','127.0.0.1','200 OK',16,'admin','2022-02-22 11:41:52'),(205,'移除代理','DELETE','/api/proxy/del','127.0.0.1','200 OK',14,'admin','2022-02-22 11:41:53'); /*!40000 ALTER TABLE `log` ENABLE KEYS */; UNLOCK TABLES; @@ -283,7 +280,6 @@ CREATE TABLE `media_server` ( LOCK TABLES `media_server` WRITE; /*!40000 ALTER TABLE `media_server` DISABLE KEYS */; -INSERT INTO `media_server` VALUES ('FQ3TF8yT83wh5Wvz','192.168.1.3','192.168.1.3','192.168.1.3','192.168.1.3',6080,0,10935,0,10000,10554,0,1,'035c73f7-bb6b-4889-a715-d9eb2d1925cc',18000,1,'30000,30500','30000,30500',18081,1,'2022-02-22 11:39:33','2022-02-22 11:39:33',30); /*!40000 ALTER TABLE `media_server` ENABLE KEYS */; UNLOCK TABLES; @@ -316,8 +312,10 @@ CREATE TABLE `parent_platform` ( `rtcp` int DEFAULT NULL, `status` bit(1) DEFAULT NULL, `shareAllLiveStream` int DEFAULT NULL, - PRIMARY KEY (`id`,`serverGBId`) USING BTREE -) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; + PRIMARY KEY (`id`), + UNIQUE KEY `parent_platform_id_uindex` (`id`), + UNIQUE KEY `parent_platform_pk` (`serverGBId`) +) ENGINE=InnoDB AUTO_INCREMENT=9 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; /*!40101 SET character_set_client = @saved_cs_client */; -- @@ -326,7 +324,6 @@ CREATE TABLE `parent_platform` ( LOCK TABLES `parent_platform` WRITE; /*!40000 ALTER TABLE `parent_platform` DISABLE KEYS */; -INSERT INTO `parent_platform` VALUES (4,0,'11111','111111111111111','1111111111','11.111.11.11',2342,'34020000013000000001','192.168.1.3','15060','34020000013000000001','12345678','300','60','UDP','GB2312','111111111111111',1,0,_binary '\0',0),(5,0,'2222','222222222222222','2222222222','222.22.2.2',2222,'34020000013000000001','192.168.1.3','15060','34020000013000000001','12345678','300','60','UDP','GB2312','222222222222222',1,0,_binary '\0',0); /*!40000 ALTER TABLE `parent_platform` ENABLE KEYS */; UNLOCK TABLES; @@ -379,7 +376,6 @@ CREATE TABLE `platform_gb_channel` ( LOCK TABLES `platform_gb_channel` WRITE; /*!40000 ALTER TABLE `platform_gb_channel` DISABLE KEYS */; -INSERT INTO `platform_gb_channel` VALUES (1,'34020000001310000001','34020000002000000005','111111111111111','34020000002000000005_34020000001310000001','111111111111111'),(2,'34020000001310000065','34020000002000000005','222222222222222','34020000002000000005_34020000001310000065','222222222222222'); /*!40000 ALTER TABLE `platform_gb_channel` ENABLE KEYS */; UNLOCK TABLES; @@ -395,8 +391,9 @@ CREATE TABLE `platform_gb_stream` ( `catalogId` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `gbStreamId` int NOT NULL, `id` int NOT NULL AUTO_INCREMENT, - PRIMARY KEY (`id`) -) ENGINE=InnoDB AUTO_INCREMENT=42 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; + PRIMARY KEY (`id`), + UNIQUE KEY `platform_gb_stream_pk` (`platformId`,`catalogId`,`gbStreamId`) +) ENGINE=InnoDB AUTO_INCREMENT=256 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; /*!40101 SET character_set_client = @saved_cs_client */; -- @@ -405,7 +402,6 @@ CREATE TABLE `platform_gb_stream` ( LOCK TABLES `platform_gb_stream` WRITE; /*!40000 ALTER TABLE `platform_gb_stream` DISABLE KEYS */; -INSERT INTO `platform_gb_stream` VALUES ('111111111111111','111111111111111',70,35),('111111111111111','111111111111111',71,36),('111111111111111','111111111111111',72,37),('222222222222222','222222222222222',73,38),('111111111111111','111111111111111',74,39),('111111111111111','111111111111111',75,40),('111111111111111','111111111111111',76,41); /*!40000 ALTER TABLE `platform_gb_stream` ENABLE KEYS */; UNLOCK TABLES; @@ -434,7 +430,8 @@ CREATE TABLE `stream_proxy` ( `status` bit(1) NOT NULL, `enable_remove_none_reader` bit(1) NOT NULL, `createTime` varchar(50) COLLATE utf8mb4_general_ci NOT NULL, - PRIMARY KEY (`id`) + PRIMARY KEY (`id`), + UNIQUE KEY `stream_proxy_pk` (`app`,`stream`) ) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; /*!40101 SET character_set_client = @saved_cs_client */; @@ -466,7 +463,7 @@ CREATE TABLE `stream_push` ( `mediaServerId` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL, PRIMARY KEY (`id`), UNIQUE KEY `stream_push_pk` (`app`,`stream`) -) ENGINE=InnoDB AUTO_INCREMENT=16 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; +) ENGINE=InnoDB AUTO_INCREMENT=310 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; /*!40101 SET character_set_client = @saved_cs_client */; -- @@ -475,7 +472,6 @@ CREATE TABLE `stream_push` ( LOCK TABLES `stream_push` WRITE; /*!40000 ALTER TABLE `stream_push` DISABLE KEYS */; -INSERT INTO `stream_push` VALUES (13,'1000','10000000_000','0',2,'rtsp_push',1645500864142,NULL,'FQ3TF8yT83wh5Wvz'),(14,'1000','10000001_000','0',2,'rtsp_push',1645500864142,NULL,'FQ3TF8yT83wh5Wvz'),(15,'1000','10000002_000','0',2,'rtsp_push',1645500864142,NULL,'FQ3TF8yT83wh5Wvz'); /*!40000 ALTER TABLE `stream_push` ENABLE KEYS */; UNLOCK TABLES; @@ -544,4 +540,4 @@ UNLOCK TABLES; /*!40101 SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION */; /*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */; --- Dump completed on 2022-02-22 11:43:00 +-- Dump completed on 2022-02-24 16:52:12 diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/GbStream.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/GbStream.java index 21b46d82..1b77ddec 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/GbStream.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/GbStream.java @@ -5,7 +5,7 @@ package com.genersoft.iot.vmp.gb28181.bean; */ public class GbStream extends PlatformGbStream{ - private int id; + private Integer gbStreamId; private String app; private String stream; private String gbId; @@ -20,12 +20,13 @@ public class GbStream extends PlatformGbStream{ */ public Long createStamp; - public int getId() { - return id; + @Override + public Integer getGbStreamId() { + return gbStreamId; } - public void setId(int id) { - this.id = id; + public void setGbStreamId(Integer gbStreamId) { + this.gbStreamId = gbStreamId; } public String getApp() { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/PlatformGbStream.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/PlatformGbStream.java index 9883a8f3..f425dfab 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/PlatformGbStream.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/PlatformGbStream.java @@ -1,15 +1,15 @@ package com.genersoft.iot.vmp.gb28181.bean; public class PlatformGbStream { - private String gbStreamId; + private Integer gbStreamId; private String platformId; private String catalogId; - public String getGbStreamId() { + public Integer getGbStreamId() { return gbStreamId; } - public void setGbStreamId(String gbStreamId) { + public void setGbStreamId(Integer gbStreamId) { this.gbStreamId = gbStreamId; } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamPushItem.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamPushItem.java index d3c9b1bd..13368070 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamPushItem.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamPushItem.java @@ -8,6 +8,11 @@ import java.util.List; public class StreamPushItem extends GbStream implements Comparable{ + /** + * id + */ + private Integer id; + /** * 应用名 */ @@ -111,6 +116,13 @@ public class StreamPushItem extends GbStream implements Comparable streamPushItems); - void batchAddForUpload(String platformId, String catalogId, List streamPushItems); + void batchAddForUpload(List streamPushItems, Map> streamPushItemsForAll); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java index b1158bc9..2b4515f7 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java @@ -159,7 +159,9 @@ public class GbStreamServiceImpl implements IGbStreamService { List parentPlatforms = platformGbStreamMapper.selectByAppAndStream(gs.getApp(), gs.getStream()); if (parentPlatforms.size() > 0) { for (ParentPlatform parentPlatform : parentPlatforms) { - eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), gs, type); + if (parentPlatform != null) { + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), gs, type); + } } } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java index 80ea97e2..8a015d9d 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java @@ -28,6 +28,7 @@ import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import java.util.*; +import java.util.stream.Collectors; @Service public class StreamPushServiceImpl implements IStreamPushService { @@ -359,23 +360,63 @@ public class StreamPushServiceImpl implements IStreamPushService { } @Override - public void batchAddForUpload(String platformId, String catalogId, List streamPushItems) { + public void batchAddForUpload(List streamPushItems, Map> streamPushItemsForAll ) { + // 存储数据到stream_push表 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; + List streamPushItemForGbStream = streamPushItems.stream() + .filter(streamPushItem-> streamPushItem.getId() != null) + .collect(Collectors.toList()); + // 存储数据到gb_stream表， id会返回到streamPushItemForGbStream里 + if (streamPushItemForGbStream.size() > 0) { + gbStreamMapper.batchAdd(streamPushItemForGbStream); + } + // 去除没有ID也就是没有存储到数据库的数据 + List streamPushItemsForPlatform = streamPushItemForGbStream.stream() + .filter(streamPushItem-> streamPushItem.getGbStreamId() != null) + .collect(Collectors.toList()); + + if (streamPushItemsForPlatform.size() > 0) { + List streamPushItemListFroPlatform = new ArrayList<>(); + Map> platformForEvent = new HashMap<>(); + // 遍历存储结果，查找app+Stream->platformId+catalogId的对应关系，然后执行批量写入 + for (StreamPushItem streamPushItem : streamPushItemsForPlatform) { + List platFormInfoList = streamPushItemsForAll.get(streamPushItem.getApp() + streamPushItem.getStream()); + if (platFormInfoList != null) { + if (platFormInfoList.size() > 0) { + for (String[] platFormInfoArray : platFormInfoList) { + StreamPushItem streamPushItemForPlatform = new StreamPushItem(); + streamPushItemForPlatform.setGbStreamId(streamPushItem.getGbStreamId()); + if (platFormInfoArray.length > 0) { + // 数组 platFormInfoArray 0 为平台ID。 1为目录ID + streamPushItemForPlatform.setPlatformId(platFormInfoArray[0]); + + List streamPushItemsInPlatform = platformForEvent.get(streamPushItem.getPlatformId()); + if (streamPushItemsInPlatform == null) { + streamPushItemsInPlatform = new ArrayList<>(); + platformForEvent.put(platFormInfoArray[0], streamPushItemsInPlatform); + } + // 为发送通知整理数据 + streamPushItemForPlatform.setApp(streamPushItem.getApp()); + streamPushItemForPlatform.setStream(streamPushItem.getStream()); + streamPushItemForPlatform.setGbId(streamPushItem.getGbId()); + streamPushItemsInPlatform.add(streamPushItemForPlatform); + } + if (platFormInfoArray.length > 1) { + streamPushItemForPlatform.setCatalogId(platFormInfoArray[1]); + } + streamPushItemListFroPlatform.add(streamPushItemForPlatform); + + + } } + } - List gbStreamList = gbStreamMapper.selectAllForAppAndStream(streamPushItems); - platformGbStreamMapper.batchAdd(platformId, catalogId, gbStreamList); - eventPublisher.catalogEventPublishForStream(platformId, streamPushItems.toArray(new GbStream[0]), CatalogEvent.ADD); + } + platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform); + // 发送通知 + for (String platformId : platformForEvent.keySet()) { + eventPublisher.catalogEventPublishForStream( + platformId, platformForEvent.get(platformId).toArray(new GbStream[0]), CatalogEvent.ADD); } } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushUploadFileHandler.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushUploadFileHandler.java index bafbb0e8..87e084b7 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushUploadFileHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushUploadFileHandler.java @@ -5,21 +5,45 @@ import com.alibaba.excel.event.AnalysisEventListener; import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto; +import com.google.common.collect.BiMap; +import com.google.common.collect.HashBiMap; import org.springframework.util.StringUtils; import java.util.*; public class StreamPushUploadFileHandler extends AnalysisEventListener { + // 错误数据的回调，用于将错误数据发送给页面 private ErrorDataHandler errorDataHandler; + + // 推流的业务类用于存储数据 private IStreamPushService pushService; + + // 默认流媒体节点ID private String defaultMediaServerId; + + // 用于存储不加过滤的所有数据 private List streamPushItems = new ArrayList<>(); - private Map streamPushItemsForPlatform = new HashMap<>(); + + // 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表 + private Map streamPushItemForSave = new HashMap<>(); + + // 用于存储按照APP+Stream为KEY， 平台ID+目录Id 为value的数据，用于存储到gb_stream表后获取app+Stream对应的平台与目录信息，然后存入关联表 + private Map> streamPushItemsForPlatform = new HashMap<>(); + + // 用于判断文件是否存在重复的app+Stream+平台ID private Set streamPushStreamSet = new HashSet<>(); - private Map streamPushGBMap = new HashMap<>(); + + // 用于存储APP+Stream->国标ID 的数据结构, 数据一一对应，全局判断APP+Stream->国标ID是否存在不对应 + private BiMap gBMap = HashBiMap.create(); + + // 记录错误的APP+Stream private List errorStreamList = new ArrayList<>(); + + + // 记录错误的国标ID private List errorGBList = new ArrayList<>(); + // 读取数量计数器 private int loadedSize = 0; @@ -33,16 +57,6 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener streams, List gbId); } - private class UploadData{ - public String platformId; - public Map> catalogData = new HashMap<>(); - public List streamPushItems = new ArrayList<>(); - - public UploadData(String platformId) { - this.platformId = platformId; - } - } - @Override public void invoke(StreamPushExcelDto streamPushExcelDto, AnalysisContext analysisContext) { if (StringUtils.isEmpty(streamPushExcelDto.getApp()) @@ -50,18 +64,28 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener platformList = streamPushItemsForPlatform.get(streamPushItem.getApp() + streamPushItem.getStream()); + if (platformList == null) { + platformList = new ArrayList<>(); + streamPushItemsForPlatform.put(streamPushItem.getApp() + streamPushItem.getStream(), platformList); } - if (!StringUtils.isEmpty(streamPushExcelDto.getCatalogId())) { - List 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); + String platformId = streamPushExcelDto.getPlatformId(); + String catalogId = streamPushExcelDto.getCatalogId(); + if (StringUtils.isEmpty(streamPushExcelDto.getCatalogId())) { + catalogId = null; } - + String[] platFormInfoArray = new String[]{platformId, catalogId}; + platformList.add(platFormInfoArray); } - streamPushStreamSet.add(streamPushExcelDto.getApp()+streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId()); loadedSize ++; if (loadedSize > 1000) { saveData(); streamPushItems.clear(); + streamPushItemForSave.clear(); streamPushItemsForPlatform.clear(); loadedSize = 0; } @@ -114,30 +137,18 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener 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)); - } - } - } - } + if (streamPushItemForSave.size() > 0) { + // 向数据库查询是否存在重复的app + pushService.batchAddForUpload(new ArrayList<>(streamPushItemForSave.values()), streamPushItemsForPlatform); } } } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java index f2e9b6f0..55b53793 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java @@ -41,15 +41,15 @@ public interface GbStreamMapper { "SELECT gs.* FROM gb_stream gs " + "WHERE " + "1=1 " + - " AND gs.id in" + + " AND gs.gbStreamId in" + "(select pgs.gbStreamId from platform_gb_stream pgs where pgs.platformId = #{platformId} and pgs.catalogId=#{catalogId}) " + - " AND gs.id not in" + + " AND gs.gbStreamId not in" + "(select pgs.gbStreamId from platform_gb_stream pgs where pgs.platformId = #{platformId}) " + " AND (gs.app LIKE '%${query}%' OR gs.stream LIKE '%${query}%' OR gs.gbId LIKE '%${query}%' OR gs.name LIKE '%${query}%') " + " AND gs.status=1" + " AND gs.status=0" + " AND gs.mediaServerId=#{mediaServerId} " + - " order by gs.id asc " + + " order by gs.gbStreamId asc " + "") List selectAll(String platformId, String catalogId, String query, Boolean pushing, String mediaServerId); @@ -60,18 +60,18 @@ public interface GbStreamMapper { List selectByGBId(String gbId); @Select("SELECT gs.*, pgs.platformId as platformId, pgs.catalogId as catalogId FROM gb_stream gs " + - "LEFT JOIN platform_gb_stream pgs ON gs.id = pgs.catalogId " + + "LEFT JOIN platform_gb_stream pgs ON gs.gbStreamId = pgs.catalogId " + "WHERE gs.gbId = '${gbId}' AND pgs.platformId = '${platformId}'") GbStream queryStreamInPlatform(String platformId, String gbId); @Select("SELECT gs.*, pgs.platformId as platformId, pgs.catalogId as catalogId FROM gb_stream gs " + - "LEFT JOIN platform_gb_stream pgs ON gs.id = pgs.gbStreamId " + + "LEFT JOIN platform_gb_stream pgs ON gs.gbStreamId = pgs.gbStreamId " + "WHERE pgs.platformId = #{platformId}") List queryGbStreamListInPlatform(String platformId); @Select("SELECT gs.* FROM gb_stream gs LEFT JOIN platform_gb_stream pgs " + - "ON gs.id = pgs.gbStreamId WHERE pgs.gbStreamId is NULL") + "ON gs.gbStreamId = pgs.gbStreamId WHERE pgs.gbStreamId is NULL") List queryStreamNotInPlatform(); @Update("UPDATE gb_stream " + @@ -104,7 +104,7 @@ public interface GbStreamMapper { void batchDelForGbStream(List gbStreams); @Insert("") + @Options(useGeneratedKeys = true, keyProperty = "gbStreamId", keyColumn = "gbStreamId") void batchAdd(List subList); @Update({"") - int batchAdd(String platformId, String catalogId, List streamPushItems); + int batchAdd(List streamPushItems); - @Delete("DELETE FROM platform_gb_stream WHERE gbStreamId = (select id from gb_stream where app=#{app} AND stream=#{stream})") + @Delete("DELETE FROM platform_gb_stream WHERE gbStreamId = (select gbStreamId from gb_stream where app=#{app} AND stream=#{stream})") int delByAppAndStream(String app, String stream); @Delete("DELETE FROM platform_gb_stream WHERE platformId=#{platformId}") @@ -42,28 +42,28 @@ public interface PlatformGbStreamMapper { "FROM " + "platform_gb_stream pgs " + "LEFT JOIN parent_platform pp ON pp.serverGBId = pgs.platformId " + - "LEFT JOIN gb_stream gs ON gs.id = pgs.gbStreamId " + + "LEFT JOIN gb_stream gs ON gs.gbStreamId = pgs.gbStreamId " + "WHERE " + "gs.app =#{app} " + "AND gs.stream =#{stream} ") List selectByAppAndStream(String app, String stream); @Select("SELECT pgs.*, gs.gbId FROM platform_gb_stream pgs " + - "LEFT JOIN gb_stream gs ON pgs.gbStreamId = gs.id " + + "LEFT JOIN gb_stream gs ON pgs.gbStreamId = gs.gbStreamId " + "WHERE gs.app=#{app} AND gs.stream=#{stream} AND pgs.platformId=#{serverGBId}") StreamProxyItem selectOne(String app, String stream, String serverGBId); @Select("select gs.* \n" + "from gb_stream gs\n" + " left join platform_gb_stream pgs\n" + - " on gs.id = pgs.gbStreamId\n" + + " on gs.gbStreamId = pgs.gbStreamId\n" + "where pgs.platformId=#{platformId} and pgs.catalogId=#{catalogId}") List queryChannelInParentPlatformAndCatalog(String platformId, String catalogId); @Select("select gs.gbId as id, gs.name as name, pgs.platformId as platformId, pgs.catalogId as catalogId , 0 as childrenCount, 2 as type\n" + "from gb_stream gs\n" + " left join platform_gb_stream pgs\n" + - " on gs.id = pgs.gbStreamId\n" + + " on gs.gbStreamId = pgs.gbStreamId\n" + "where pgs.platformId=#{platformId} and pgs.catalogId=#{catalogId}") List queryChannelInParentPlatformAndCatalogForCatalog(String platformId, String catalogId); @@ -78,7 +78,7 @@ public interface PlatformGbStreamMapper { "left join platform_gb_stream pgs on " + "pp.serverGBId = pgs.platformId " + "left join gb_stream gs " + - "gs.id = pgs.gbStreamId " + + "gs.gbStreamId = pgs.gbStreamId " + "WHERE " + "gs.app = #{app} " + "AND gs.stream = #{stream}" + @@ -93,7 +93,7 @@ public interface PlatformGbStreamMapper { @Delete("") void delByGbStreams(List gbStreams); @@ -101,7 +101,7 @@ public interface PlatformGbStreamMapper { @Delete("") void delByAppAndStreamsByPlatformId(List gbStreams, String platformId); diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java index 66e85d09..594c319e 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java @@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.storager.dao; import com.genersoft.iot.vmp.gb28181.bean.GbStream; import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import org.apache.ibatis.annotations.*; +import org.omg.PortableInterceptor.INACTIVE; import org.springframework.stereotype.Repository; import java.util.Collection; @@ -61,39 +62,37 @@ public interface StreamPushMapper { @Select(value = {" "}) List selectAllForList(String query, Boolean pushing, String mediaServerId); - @Select("SELECT st.*, pgs.gbId, pgs.status, pgs.name, pgs.longitude, pgs.latitude FROM stream_push st LEFT JOIN gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream order by st.createStamp desc") + @Select("SELECT st.*, gs.gbId, gs.status, gs.name, gs.longitude, gs.latitude FROM stream_push st LEFT JOIN gb_stream gs on st.app = gs.app AND st.stream = gs.stream order by st.createStamp desc") List selectAll(); - @Select("SELECT st.*, pgs.gbId, pgs.status, pgs.name, pgs.longitude, pgs.latitude FROM stream_push st LEFT JOIN gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream WHERE st.enable=${enable}") - List selectForEnable(boolean enable); - - @Select("SELECT st.*, pgs.gbId, pgs.status, pgs.name, pgs.longitude, pgs.latitude FROM stream_push st LEFT JOIN gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream WHERE st.app=#{app} AND st.stream=#{stream}") + @Select("SELECT st.*, gs.gbId, gs.status, gs.name, gs.longitude, gs.latitude FROM stream_push st LEFT JOIN gb_stream gs on st.app = gs.app AND st.stream = gs.stream WHERE st.app=#{app} AND st.stream=#{stream}") StreamPushItem selectOne(String app, String stream); @Insert("") - void addAll(List streamPushItems); + @Options(useGeneratedKeys = true, keyProperty = "id", keyColumn = "id") + int addAll(List streamPushItems); @Delete("DELETE FROM stream_push") void clear(); diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java index fabf267b..1d04c82d 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java @@ -748,7 +748,7 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager { return false; } } - }else { + } else { //事务回滚 dataSourceTransactionManager.rollback(transactionStatus); return false;