From 4abbf4df07f02c0c94baa2787fdcf4a14dcb3463 Mon Sep 17 00:00:00 2001
From: sc.dm <822878110@qq.com>
Date: 星期四, 10 三月 2022 17:42:02 +0800
Subject: [PATCH] Merge branch '648540858:wvp-28181-2.0' into wvp-28181-2.0
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 207 ++++++++++++++++++++++++++++++++++++++++++++-------
1 files changed, 178 insertions(+), 29 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 5fef8cf..39d37c5 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,19 +5,22 @@
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.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;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
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.storager.dao.*;
+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;
@@ -25,6 +28,7 @@
import org.springframework.util.StringUtils;
import java.util.*;
+import java.util.stream.Collectors;
@Service
public class StreamPushServiceImpl implements IStreamPushService {
@@ -39,7 +43,16 @@
private ParentPlatformMapper parentPlatformMapper;
@Autowired
+ private PlatformCatalogMapper platformCatalogMapper;
+
+ @Autowired
private PlatformGbStreamMapper platformGbStreamMapper;
+
+ @Autowired
+ private IGbStreamService gbStreamService;
+
+ @Autowired
+ private EventPublisher eventPublisher;
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@@ -84,13 +97,12 @@
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");
@@ -99,40 +111,55 @@
}
@Override
- public PageInfo<StreamPushItem> getPushList(Integer page, Integer count) {
+ public PageInfo<StreamPushItem> getPushList(Integer page, Integer count, String query, Boolean pushing, String mediaServerId) {
PageHelper.startPage(page, count);
- List<StreamPushItem> all = streamPushMapper.selectAll();
+ List<StreamPushItem> all = streamPushMapper.selectAllForList(query, pushing, mediaServerId);
return new PageInfo<>(all);
}
@Override
public List<StreamPushItem> getPushList(String mediaServerId) {
- return streamPushMapper.selectAllByMediaServerId(mediaServerId);
+ return streamPushMapper.selectAllByMediaServerIdWithOutGbID(mediaServerId);
}
@Override
public boolean saveToGB(GbStream stream) {
stream.setStreamType("push");
stream.setStatus(true);
+ stream.setCreateStamp(System.currentTimeMillis());
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 streamProxyItems = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId());
- if (streamProxyItems == null) {
+ 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) {
@@ -151,6 +178,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);
@@ -172,16 +201,18 @@
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 (StringUtils.isEmpty(streamPushItem.getGbId())) {
+ 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 ->{
@@ -220,19 +251,19 @@
}
}
- 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());
}
}
}));
@@ -240,7 +271,7 @@
@Override
public void zlmServerOffline(String mediaServerId) {
- List<StreamPushItem> streamPushItems = streamPushMapper.selectAllByMediaServerId(mediaServerId);
+ List<StreamPushItem> streamPushItems = streamPushMapper.selectAllByMediaServerIdWithOutGbID(mediaServerId);
// 绉婚櫎娌℃湁GBId鐨勬帹娴�
streamPushMapper.deleteWithoutGBId(mediaServerId);
gbStreamMapper.deleteWithoutGBId("push", mediaServerId);
@@ -249,15 +280,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);
@@ -278,6 +309,7 @@
streamPushItem.setStreamType("push");
streamPushItem.setStatus(true);
streamPushItem.setGbId("34020000004111" + gbId);
+ streamPushItem.setCreateStamp(System.currentTimeMillis());
gbId ++;
}
int limitCount = 30;
@@ -295,4 +327,121 @@
}
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);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void batchAddForUpload(List<StreamPushItem> streamPushItems, Map<String, List<String[]>> streamPushItemsForAll ) {
+ // 瀛樺偍鏁版嵁鍒皊tream_push琛�
+ streamPushMapper.addAll(streamPushItems);
+ List<StreamPushItem> streamPushItemForGbStream = streamPushItems.stream()
+ .filter(streamPushItem-> streamPushItem.getId() != null)
+ .collect(Collectors.toList());
+ // 瀛樺偍鏁版嵁鍒癵b_stream琛紝 id浼氳繑鍥炲埌streamPushItemForGbStream閲�
+ if (streamPushItemForGbStream.size() > 0) {
+ gbStreamMapper.batchAdd(streamPushItemForGbStream);
+ }
+ // 鍘婚櫎娌℃湁ID涔熷氨鏄病鏈夊瓨鍌ㄥ埌鏁版嵁搴撶殑鏁版嵁
+ List<StreamPushItem> streamPushItemsForPlatform = streamPushItemForGbStream.stream()
+ .filter(streamPushItem-> streamPushItem.getGbStreamId() != null)
+ .collect(Collectors.toList());
+
+ if (streamPushItemsForPlatform.size() > 0) {
+ List<StreamPushItem> streamPushItemListFroPlatform = new ArrayList<>();
+ Map<String, List<GbStream>> platformForEvent = new HashMap<>();
+ // 閬嶅巻瀛樺偍缁撴灉锛屾煡鎵綼pp+Stream->platformId+catalogId鐨勫搴斿叧绯伙紝鐒跺悗鎵ц鎵归噺鍐欏叆
+ for (StreamPushItem streamPushItem : streamPushItemsForPlatform) {
+ List<String[]> 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 涓哄钩鍙癐D銆� 1涓虹洰褰旾D
+ streamPushItemForPlatform.setPlatformId(platFormInfoArray[0]);
+
+ List<GbStream> gbStreamList = platformForEvent.get(streamPushItem.getPlatformId());
+ if (gbStreamList == null) {
+ gbStreamList = new ArrayList<>();
+ platformForEvent.put(platFormInfoArray[0], gbStreamList);
+ }
+ // 涓哄彂閫侀�氱煡鏁寸悊鏁版嵁
+ streamPushItemForPlatform.setName(streamPushItem.getName());
+ streamPushItemForPlatform.setApp(streamPushItem.getApp());
+ streamPushItemForPlatform.setStream(streamPushItem.getStream());
+ streamPushItemForPlatform.setGbId(streamPushItem.getGbId());
+ gbStreamList.add(streamPushItemForPlatform);
+ }
+ if (platFormInfoArray.length > 1) {
+ streamPushItemForPlatform.setCatalogId(platFormInfoArray[1]);
+ }
+ streamPushItemListFroPlatform.add(streamPushItemForPlatform);
+
+
+ }
+ }
+
+ }
+ }
+ if (streamPushItemListFroPlatform.size() > 0) {
+ platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
+ // 鍙戦�侀�氱煡
+ for (String platformId : platformForEvent.keySet()) {
+ eventPublisher.catalogEventPublishForStream(
+ platformId, platformForEvent.get(platformId), CatalogEvent.ADD);
+ }
+ }
+
+ }
+ }
+
+ @Override
+ public boolean batchStop(List<GbStream> gbStreams) {
+ if (gbStreams == null || gbStreams.size() == 0) {
+ return false;
+ }
+ gbStreamService.sendCatalogMsgs(gbStreams, CatalogEvent.DEL);
+
+ int delStream = streamPushMapper.delAllForGbStream(gbStreams);
+ gbStreamMapper.batchDelForGbStream(gbStreams);
+ platformGbStreamMapper.delByGbStreams(gbStreams);
+ if (delStream > 0) {
+ for (GbStream gbStream : gbStreams) {
+ MediaServerItem mediaServerItem = mediaServerService.getOne(gbStream.getMediaServerId());
+ zlmresTfulUtils.closeStreams(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+ }
+
+ }
+ return true;
+ }
}
--
Gitblit v1.8.0