From 1af77ab5f7c11a4b3d59c1989b51b9fca29679ce Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 18 十月 2022 22:18:49 +0800
Subject: [PATCH] Merge pull request #645 from IKangXu/wvp-28181-2.0
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 279 ++++++++++++++++++++++++++++++++++---------------------
1 files changed, 174 insertions(+), 105 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 8a015d9..00980f9 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
@@ -1,30 +1,33 @@
package com.genersoft.iot.vmp.service.impl;
import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
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.conf.MediaConfig;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.gb28181.bean.*;
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.service.bean.StreamPushItemFromRedis;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.*;
-import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto;
+import com.genersoft.iot.vmp.utils.DateUtil;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import java.util.*;
@@ -33,11 +36,16 @@
@Service
public class StreamPushServiceImpl implements IStreamPushService {
+ private final static Logger logger = LoggerFactory.getLogger(StreamPushServiceImpl.class);
+
@Autowired
private GbStreamMapper gbStreamMapper;
@Autowired
private StreamPushMapper streamPushMapper;
+
+ @Autowired
+ private StreamProxyMapper streamProxyMapper;
@Autowired
private ParentPlatformMapper parentPlatformMapper;
@@ -61,14 +69,26 @@
private IRedisCatchStorage redisCatchStorage;
@Autowired
- private UserSetup userSetup;
+ private UserSetting userSetting;
@Autowired
private IMediaServerService mediaServerService;
+ @Autowired
+ DataSourceTransactionManager dataSourceTransactionManager;
+
+ @Autowired
+ TransactionDefinition transactionDefinition;
+
+ @Autowired
+ private MediaConfig mediaConfig;
+
+
@Override
public List<StreamPushItem> handleJSON(String jsonData, MediaServerItem mediaServerItem) {
- if (jsonData == null) return null;
+ if (jsonData == null) {
+ return null;
+ }
Map<String, StreamPushItem> result = new HashMap<>();
@@ -102,11 +122,12 @@
streamPushItem.setOriginType(item.getOriginType());
streamPushItem.setOriginTypeStr(item.getOriginTypeStr());
streamPushItem.setOriginUrl(item.getOriginUrl());
- streamPushItem.setCreateStamp(item.getCreateStamp() * 1000);
+ streamPushItem.setCreateTime(DateUtil.getNow());
streamPushItem.setAliveSecond(item.getAliveSecond());
streamPushItem.setStatus(true);
streamPushItem.setStreamType("push");
streamPushItem.setVhost(item.getVhost());
+ streamPushItem.setServerId(item.getSeverId());
return streamPushItem;
}
@@ -126,31 +147,10 @@
public boolean saveToGB(GbStream stream) {
stream.setStreamType("push");
stream.setStatus(true);
- stream.setCreateStamp(System.currentTimeMillis());
+ stream.setCreateTime(DateUtil.getNow());
+ stream.setStreamType("push");
+ stream.setMediaServerId(mediaConfig.getId());
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;
}
@@ -158,12 +158,17 @@
public boolean removeFromGB(GbStream stream) {
// 鍒ゆ柇鏄惁闇�瑕佸彂閫佷簨浠�
gbStreamService.sendCatalogMsg(stream, CatalogEvent.DEL);
- int del = gbStreamMapper.del(stream.getApp(), stream.getStream());
platformGbStreamMapper.delByAppAndStream(stream.getApp(), stream.getStream());
+ int del = gbStreamMapper.del(stream.getApp(), stream.getStream());
MediaServerItem mediaInfo = mediaServerService.getOne(stream.getMediaServerId());
JSONObject mediaList = zlmresTfulUtils.getMediaList(mediaInfo, stream.getApp(), stream.getStream());
- if (mediaList == null) {
- streamPushMapper.del(stream.getApp(), stream.getStream());
+ if (mediaList != null) {
+ if (mediaList.getInteger("code") == 0) {
+ JSONArray data = mediaList.getJSONArray("data");
+ if (data == null) {
+ streamPushMapper.del(stream.getApp(), stream.getStream());
+ }
+ }
}
return del > 0;
}
@@ -171,7 +176,6 @@
@Override
public StreamPushItem getPush(String app, String streamId) {
-
return streamPushMapper.selectOne(app, streamId);
}
@@ -180,9 +184,9 @@
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);
+ gbStreamMapper.del(app, streamId);
+ int delStream = streamPushMapper.del(app, streamId);
if (delStream > 0) {
MediaServerItem mediaServerItem = mediaServerService.getOne(streamPushItem.getMediaServerId());
zlmresTfulUtils.closeStreams(mediaServerItem,app, streamId);
@@ -205,7 +209,7 @@
Map<String, MediaItem> streamInfoPushItemMap = new HashMap<>();
if (pushList.size() > 0) {
for (StreamPushItem streamPushItem : pushList) {
- if (StringUtils.isEmpty(streamPushItem.getGbId())) {
+ if (ObjectUtils.isEmpty(streamPushItem.getGbId())) {
pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem);
}
}
@@ -216,7 +220,9 @@
}
}
zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
- if (mediaList == null) return;
+ if (mediaList == null) {
+ return;
+ }
String dataStr = mediaList.getString("data");
Integer code = mediaList.getInteger("code");
@@ -256,7 +262,7 @@
String type = "PUSH";
for (MediaItem offlineMediaItem : offlineMediaItemList) {
JSONObject jsonObject = new JSONObject();
- jsonObject.put("serverId", userSetup.getServerId());
+ jsonObject.put("serverId", userSetting.getServerId());
jsonObject.put("app", offlineMediaItem.getApp());
jsonObject.put("stream", offlineMediaItem.getStream());
jsonObject.put("register", false);
@@ -276,7 +282,8 @@
streamPushMapper.deleteWithoutGBId(mediaServerId);
gbStreamMapper.deleteWithoutGBId("push", mediaServerId);
// 鍏朵粬鐨勬祦璁剧疆鏈惎鐢�
- gbStreamMapper.updateStatusByMediaServerId(mediaServerId, false);
+ streamPushMapper.updateStatusByMediaServerId(mediaServerId, false);
+ streamProxyMapper.updateStatusByMediaServerId(mediaServerId, false);
// 鍙戦�佹祦鍋滄娑堟伅
String type = "PUSH";
// 鍙戦�乺edis娑堟伅
@@ -286,7 +293,7 @@
// 绉婚櫎redis鍐呮祦鐨勪俊鎭�
redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
JSONObject jsonObject = new JSONObject();
- jsonObject.put("serverId", userSetup.getServerId());
+ jsonObject.put("serverId", userSetting.getServerId());
jsonObject.put("app", mediaItem.getApp());
jsonObject.put("stream", mediaItem.getStream());
jsonObject.put("register", false);
@@ -309,7 +316,7 @@
streamPushItem.setStreamType("push");
streamPushItem.setStatus(true);
streamPushItem.setGbId("34020000004111" + gbId);
- streamPushItem.setCreateStamp(System.currentTimeMillis());
+ streamPushItem.setCreateTime(DateUtil.getNow());
gbId ++;
}
int limitCount = 30;
@@ -332,32 +339,8 @@
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 ) {
@@ -376,47 +359,74 @@
.collect(Collectors.toList());
if (streamPushItemsForPlatform.size() > 0) {
+ // 鑾峰彇鎵�鏈夊钩鍙帮紝骞冲彴鍜岀洰褰曚俊鎭竴鑸笉浼氱壒鍒ぇ閲忋��
+ List<ParentPlatform> parentPlatformList = parentPlatformMapper.getParentPlatformList();
+ Map<String, Map<String, PlatformCatalog>> platformInfoMap = new HashMap<>();
+ if (parentPlatformList.size() == 0) {
+ return;
+ }
+ for (ParentPlatform platform : parentPlatformList) {
+ Map<String, PlatformCatalog> catalogMap = new HashMap<>();
+
+ // 鍒涘缓鏍硅妭鐐�
+ PlatformCatalog platformCatalog = new PlatformCatalog();
+ platformCatalog.setId(platform.getServerGBId());
+ catalogMap.put(platform.getServerGBId(), platformCatalog);
+
+ // 鏌ヨ鎵�鏈夎妭鐐逛俊鎭�
+ List<PlatformCatalog> platformCatalogs = platformCatalogMapper.selectByPlatForm(platform.getServerGBId());
+ if (platformCatalogs.size() > 0) {
+ for (PlatformCatalog catalog : platformCatalogs) {
+ catalogMap.put(catalog.getId(), catalog);
+ }
+ }
+ platformInfoMap.put(platform.getServerGBId(), catalogMap);
+ }
List<StreamPushItem> streamPushItemListFroPlatform = new ArrayList<>();
- Map<String, List<StreamPushItem>> platformForEvent = new HashMap<>();
+ 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<StreamPushItem> 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 (platFormInfoList != null && platFormInfoList.size() > 0) {
+ for (String[] platFormInfoArray : platFormInfoList) {
+ StreamPushItem streamPushItemForPlatform = new StreamPushItem();
+ streamPushItemForPlatform.setGbStreamId(streamPushItem.getGbStreamId());
+ if (platFormInfoArray.length > 0) {
+ // 鏁扮粍 platFormInfoArray 0 涓哄钩鍙癐D銆� 1涓虹洰褰旾D
+ // 涓嶅瓨鍦ㄨ繖涓钩鍙帮紝鍒欏拷鐣ュ鍏ユ鍏宠仈鍏崇郴
+ if (platformInfoMap.get(platFormInfoArray[0]) == null
+ || platformInfoMap.get(platFormInfoArray[0]).get(platFormInfoArray[1]) == null) {
+ logger.info("瀵煎叆鏁版嵁鏃朵笉瀛樺湪骞冲彴鎴栫洰褰晎}/{},宸插鍏ユ湭鍒嗛厤", platFormInfoArray[0], platFormInfoArray[1] );
+ continue;
}
- if (platFormInfoArray.length > 1) {
- streamPushItemForPlatform.setCatalogId(platFormInfoArray[1]);
+ streamPushItemForPlatform.setPlatformId(platFormInfoArray[0]);
+ List<GbStream> gbStreamList = platformForEvent.get(platFormInfoArray[0]);
+ if (gbStreamList == null) {
+ gbStreamList = new ArrayList<>();
+ platformForEvent.put(platFormInfoArray[0], gbStreamList);
}
- streamPushItemListFroPlatform.add(streamPushItemForPlatform);
-
-
+ // 涓哄彂閫侀�氱煡鏁寸悊鏁版嵁
+ 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);
}
}
}
- platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
- // 鍙戦�侀�氱煡
- for (String platformId : platformForEvent.keySet()) {
- eventPublisher.catalogEventPublishForStream(
- platformId, platformForEvent.get(platformId).toArray(new GbStream[0]), CatalogEvent.ADD);
+ if (streamPushItemListFroPlatform.size() > 0) {
+ platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
+ // 鍙戦�侀�氱煡
+ for (String platformId : platformForEvent.keySet()) {
+ eventPublisher.catalogEventPublishForStream(
+ platformId, platformForEvent.get(platformId), CatalogEvent.ADD);
+ }
}
}
}
@@ -428,9 +438,9 @@
}
gbStreamService.sendCatalogMsgs(gbStreams, CatalogEvent.DEL);
- int delStream = streamPushMapper.delAllForGbStream(gbStreams);
- gbStreamMapper.batchDelForGbStream(gbStreams);
platformGbStreamMapper.delByGbStreams(gbStreams);
+ gbStreamMapper.batchDelForGbStream(gbStreams);
+ int delStream = streamPushMapper.delAllForGbStream(gbStreams);
if (delStream > 0) {
for (GbStream gbStream : gbStreams) {
MediaServerItem mediaServerItem = mediaServerService.getOne(gbStream.getMediaServerId());
@@ -440,4 +450,63 @@
}
return true;
}
+
+ @Override
+ public void allStreamOffline() {
+ List<GbStream> onlinePushers = streamPushMapper.getOnlinePusherForGb();
+ if (onlinePushers.size() == 0) {
+ return;
+ }
+ streamPushMapper.setAllStreamOffline();
+
+ // 鍙戦�侀�氱煡
+ eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.OFF);
+ }
+
+ @Override
+ public void offline(List<StreamPushItemFromRedis> offlineStreams) {
+ // 鏇存柊閮ㄥ垎璁惧绂荤嚎
+ List<GbStream> onlinePushers = streamPushMapper.getOnlinePusherForGbInList(offlineStreams);
+ streamPushMapper.offline(offlineStreams);
+ // 鍙戦�侀�氱煡
+ eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.OFF);
+ }
+
+ @Override
+ public void online(List<StreamPushItemFromRedis> onlineStreams) {
+ // 鏇存柊閮ㄥ垎璁惧涓婄嚎streamPushService
+ List<GbStream> onlinePushers = streamPushMapper.getOfflinePusherForGbInList(onlineStreams);
+ streamPushMapper.online(onlineStreams);
+ // 鍙戦�侀�氱煡
+ eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.ON);
+ }
+
+ @Override
+ public boolean add(StreamPushItem stream) {
+ stream.setUpdateTime(DateUtil.getNow());
+ stream.setCreateTime(DateUtil.getNow());
+ stream.setServerId(userSetting.getServerId());
+
+ // 鏀惧湪浜嬪姟鍐呮墽琛�
+ boolean result = false;
+ TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
+ try {
+ int addStreamResult = streamPushMapper.add(stream);
+ if (!ObjectUtils.isEmpty(stream.getGbId())) {
+ stream.setStreamType("push");
+ gbStreamMapper.add(stream);
+ }
+ dataSourceTransactionManager.commit(transactionStatus);
+ result = true;
+ }catch (Exception e) {
+ logger.error("鎵归噺绉婚櫎娴佷笌骞冲彴鐨勫叧绯绘椂閿欒", e);
+ dataSourceTransactionManager.rollback(transactionStatus);
+ }
+ return result;
+ }
+
+ @Override
+ public List<String> getAllAppAndStream() {
+ return streamPushMapper.getAllAppAndStream();
+ }
}
--
Gitblit v1.8.0