From 385fdb95f18c1f0b4e38f8a6e481176ac31ee987 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期一, 17 一月 2022 15:48:08 +0800
Subject: [PATCH] 优化级联选择通道逻辑
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 159 +++++++++++++++++++++++++++++++++++++++++++++-------
1 files changed, 136 insertions(+), 23 deletions(-)
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 63596b2..ebfa124 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
@@ -5,24 +5,29 @@
import com.alibaba.fastjson.TypeReference;
import com.genersoft.iot.vmp.common.StreamInfo;
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.event.EventPublisher;
+import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
-import com.genersoft.iot.vmp.media.zlm.dto.OriginType;
-import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+import com.genersoft.iot.vmp.media.zlm.dto.*;
+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.vmanager.bean.StreamPushExcelDto;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
import java.util.*;
@@ -36,7 +41,16 @@
private StreamPushMapper streamPushMapper;
@Autowired
+ private ParentPlatformMapper parentPlatformMapper;
+
+ @Autowired
private PlatformGbStreamMapper platformGbStreamMapper;
+
+ @Autowired
+ private IGbStreamService gbStreamService;
+
+ @Autowired
+ private EventPublisher eventPublisher;
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@@ -70,7 +84,6 @@
result.put(key, streamPushItem);
}
}
-
}
return new ArrayList<>(result.values());
@@ -113,12 +126,38 @@
stream.setStreamType("push");
stream.setStatus(true);
int add = gbStreamMapper.add(stream);
+
+ // 鏌ユ壘寮�鍚簡鍏ㄩ儴鐩存挱娴佸叡浜殑涓婄骇骞冲彴
+ List<ParentPlatform> parentPlatforms = parentPlatformMapper.selectAllAhareAllLiveStream();
+ if (parentPlatforms.size() > 0) {
+ for (ParentPlatform parentPlatform : parentPlatforms) {
+ stream.setCatalogId(parentPlatform.getCatalogId());
+ stream.setPlatformId(parentPlatform.getServerGBId());
+ String streamId = stream.getStream();
+ StreamProxyItem streamProxyItem = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId());
+ if (streamProxyItem == null) {
+ platformGbStreamMapper.add(stream);
+ eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD);
+ }else {
+ if (!streamProxyItem.getGbId().equals(stream.getGbId())) {
+ // 姝ゆ祦浣跨敤鍙︿竴涓浗鏍嘔d宸茬粡涓庤骞冲彴鍏宠仈锛岀Щ闄ゆ璁板綍
+ platformGbStreamMapper.delByAppAndStreamAndPlatform(stream.getApp(), streamId, parentPlatform.getServerGBId());
+ platformGbStreamMapper.add(stream);
+ eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD);
+ }
+ }
+ }
+ }
+
return add > 0;
}
@Override
public boolean removeFromGB(GbStream stream) {
+ // 鍒ゆ柇鏄惁闇�瑕佸彂閫佷簨浠�
+ gbStreamService.sendCatalogMsg(stream, CatalogEvent.DEL);
int del = gbStreamMapper.del(stream.getApp(), stream.getStream());
+ platformGbStreamMapper.delByAppAndStream(stream.getApp(), stream.getStream());
MediaServerItem mediaInfo = mediaServerService.getOne(stream.getMediaServerId());
JSONObject mediaList = zlmresTfulUtils.getMediaList(mediaInfo, stream.getApp(), stream.getStream());
if (mediaList == null) {
@@ -137,6 +176,8 @@
@Override
public boolean stop(String app, String streamId) {
StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId);
+ gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
+
int delStream = streamPushMapper.del(app, streamId);
gbStreamMapper.del(app, streamId);
platformGbStreamMapper.delByAppAndStream(app, streamId);
@@ -158,16 +199,16 @@
List<StreamPushItem> pushList = getPushList(mediaServerId);
Map<String, StreamPushItem> pushItemMap = new HashMap<>();
// redis璁板綍
- List<StreamInfo> streamInfoPushList = redisCatchStorage.getStreams(mediaServerId, "PUSH");
- Map<String, StreamInfo> streamInfoPushItemMap = new HashMap<>();
+ List<MediaItem> mediaItems = redisCatchStorage.getStreams(mediaServerId, "PUSH");
+ Map<String, MediaItem> streamInfoPushItemMap = new HashMap<>();
if (pushList.size() > 0) {
for (StreamPushItem streamPushItem : pushList) {
pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem);
}
}
- if (streamInfoPushList.size() > 0) {
- for (StreamInfo streamInfo : streamInfoPushList) {
- streamInfoPushItemMap.put(streamInfo.getApp() + streamInfo.getStreamId(), streamInfo);
+ if (mediaItems.size() > 0) {
+ for (MediaItem mediaItem : mediaItems) {
+ streamInfoPushItemMap.put(mediaItem.getApp() + mediaItem.getStream(), mediaItem);
}
}
zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
@@ -188,24 +229,37 @@
streamInfoPushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
}
}
- Collection<StreamPushItem> offlinePushItems = pushItemMap.values();
+ List<StreamPushItem> offlinePushItems = new ArrayList<>(pushItemMap.values());
if (offlinePushItems.size() > 0) {
String type = "PUSH";
- streamPushMapper.delAll(new ArrayList<>(offlinePushItems));
+ int runLimit = 300;
+ if (offlinePushItems.size() > runLimit) {
+ for (int i = 0; i < offlinePushItems.size(); i += runLimit) {
+ int toIndex = i + runLimit;
+ if (i + runLimit > offlinePushItems.size()) {
+ toIndex = offlinePushItems.size();
+ }
+ List<StreamPushItem> streamPushItemsSub = offlinePushItems.subList(i, toIndex);
+ streamPushMapper.delAll(streamPushItemsSub);
+ }
+ }else {
+ streamPushMapper.delAll(offlinePushItems);
+ }
+
}
- Collection<StreamInfo> offlineStreamInfoItems = streamInfoPushItemMap.values();
- if (offlineStreamInfoItems.size() > 0) {
+ Collection<MediaItem> offlineMediaItemList = streamInfoPushItemMap.values();
+ if (offlineMediaItemList.size() > 0) {
String type = "PUSH";
- for (StreamInfo offlineStreamInfoItem : offlineStreamInfoItems) {
+ for (MediaItem offlineMediaItem : offlineMediaItemList) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetup.getServerId());
- jsonObject.put("app", offlineStreamInfoItem.getApp());
- jsonObject.put("stream", offlineStreamInfoItem.getStreamId());
+ jsonObject.put("app", offlineMediaItem.getApp());
+ jsonObject.put("stream", offlineMediaItem.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 绉婚櫎redis鍐呮祦鐨勪俊鎭�
- redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineStreamInfoItem.getApp(), offlineStreamInfoItem.getStreamId());
+ redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineMediaItem.getApp(), offlineMediaItem.getStream());
}
}
}));
@@ -222,15 +276,15 @@
// 鍙戦�佹祦鍋滄娑堟伅
String type = "PUSH";
// 鍙戦�乺edis娑堟伅
- List<StreamInfo> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+ List<MediaItem> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
if (streamInfoList.size() > 0) {
- for (StreamInfo streamInfo : streamInfoList) {
+ for (MediaItem mediaItem : streamInfoList) {
// 绉婚櫎redis鍐呮祦鐨勪俊鎭�
- redisCatchStorage.removeStream(mediaServerId, type, streamInfo.getApp(), streamInfo.getStreamId());
+ redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetup.getServerId());
- jsonObject.put("app", streamInfo.getApp());
- jsonObject.put("stream", streamInfo.getStreamId());
+ jsonObject.put("app", mediaItem.getApp());
+ jsonObject.put("stream", mediaItem.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
@@ -242,4 +296,63 @@
public void clean() {
}
+
+ @Override
+ public boolean saveToRandomGB() {
+ List<StreamPushItem> streamPushItems = streamPushMapper.selectAll();
+ long gbId = 100001;
+ for (StreamPushItem streamPushItem : streamPushItems) {
+ streamPushItem.setStreamType("push");
+ streamPushItem.setStatus(true);
+ streamPushItem.setGbId("34020000004111" + gbId);
+ gbId ++;
+ }
+ int limitCount = 30;
+
+ if (streamPushItems.size() > limitCount) {
+ for (int i = 0; i < streamPushItems.size(); i += limitCount) {
+ int toIndex = i + limitCount;
+ if (i + limitCount > streamPushItems.size()) {
+ toIndex = streamPushItems.size();
+ }
+ gbStreamMapper.batchAdd(streamPushItems.subList(i, toIndex));
+ }
+ }else {
+ gbStreamMapper.batchAdd(streamPushItems);
+ }
+ return true;
+ }
+
+ @Override
+ public void batchAdd(List<StreamPushItem> streamPushItems) {
+ streamPushMapper.addAll(streamPushItems);
+ gbStreamMapper.batchAdd(streamPushItems);
+ // 鏌ユ壘寮�鍚簡鍏ㄩ儴鐩存挱娴佸叡浜殑涓婄骇骞冲彴
+ List<ParentPlatform> parentPlatforms = parentPlatformMapper.selectAllAhareAllLiveStream();
+ if (parentPlatforms.size() > 0) {
+ for (StreamPushItem stream : streamPushItems) {
+ for (ParentPlatform parentPlatform : parentPlatforms) {
+ stream.setCatalogId(parentPlatform.getCatalogId());
+ stream.setPlatformId(parentPlatform.getServerGBId());
+ String streamId = stream.getStream();
+ StreamProxyItem streamProxyItem = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId());
+ if (streamProxyItem == null) {
+ platformGbStreamMapper.add(stream);
+ eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD);
+ }else {
+ if (!streamProxyItem.getGbId().equals(stream.getGbId())) {
+ // 姝ゆ祦浣跨敤鍙︿竴涓浗鏍嘔d宸茬粡涓庤骞冲彴鍏宠仈锛岀Щ闄ゆ璁板綍
+ platformGbStreamMapper.delByAppAndStreamAndPlatform(stream.getApp(), streamId, parentPlatform.getServerGBId());
+ platformGbStreamMapper.add(stream);
+ eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD);
+ stream.setGbId(streamProxyItem.getGbId());
+ eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.DEL);
+ }
+ }
+ }
+ }
+
+
+ }
+ }
}
--
Gitblit v1.8.0