From b6e604f2449bb65dfaafb0f0741ba54ff0d2f9c2 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 30 四月 2024 15:20:03 +0800
Subject: [PATCH] 修改云端录像详情页使用直接访问zlm的方式播放录像
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 552 ++++++++++++++++++++++++++++++++++++-------------------
1 files changed, 362 insertions(+), 190 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
old mode 100644
new mode 100755
index e59a090..a64b71a
--- 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,36 +1,49 @@
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.alibaba.fastjson2.JSONObject;
+import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.common.StreamInfo;
-import com.genersoft.iot.vmp.conf.UserSetup;
-import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.conf.MediaConfig;
+import com.genersoft.iot.vmp.conf.UserSetting;
+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.media.bean.MediaInfo;
+import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
+import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
+import com.genersoft.iot.vmp.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
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.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
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.context.event.EventListener;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
-import org.springframework.util.StringUtils;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.util.ObjectUtils;
import java.util.*;
import java.util.stream.Collectors;
@Service
+@DS("master")
public class StreamPushServiceImpl implements IStreamPushService {
private final static Logger logger = LoggerFactory.getLogger(StreamPushServiceImpl.class);
@@ -40,6 +53,9 @@
@Autowired
private StreamPushMapper streamPushMapper;
+
+ @Autowired
+ private StreamProxyMapper streamProxyMapper;
@Autowired
private ParentPlatformMapper parentPlatformMapper;
@@ -57,43 +73,144 @@
private EventPublisher eventPublisher;
@Autowired
- private ZLMRESTfulUtils zlmresTfulUtils;
-
- @Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
- private UserSetup userSetup;
+ private UserSetting userSetting;
@Autowired
private IMediaServerService mediaServerService;
- @Override
- public List<StreamPushItem> handleJSON(String jsonData, MediaServerItem mediaServerItem) {
- if (jsonData == null) return null;
+ @Autowired
+ DataSourceTransactionManager dataSourceTransactionManager;
+ @Autowired
+ TransactionDefinition transactionDefinition;
+
+ @Autowired
+ private MediaConfig mediaConfig;
+
+ /**
+ * 娴佸埌鏉ョ殑澶勭悊
+ */
+ @Async("taskExecutor")
+ @EventListener
+ public void onApplicationEvent(MediaArrivalEvent event) {
+ MediaInfo mediaInfo = event.getMediaInfo();
+ if (mediaInfo == null) {
+ return;
+ }
+ if (mediaInfo.getOriginType() != OriginType.RTMP_PUSH.ordinal()
+ && mediaInfo.getOriginType() != OriginType.RTSP_PUSH.ordinal()
+ && mediaInfo.getOriginType() != OriginType.RTC_PUSH.ordinal()) {
+ return;
+ }
+
+ StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(event.getApp(), event.getStream());
+ if (streamAuthorityInfo == null) {
+ streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(event);
+ } else {
+ streamAuthorityInfo.setOriginType(mediaInfo.getOriginType());
+ }
+ redisCatchStorage.updateStreamAuthorityInfo(event.getApp(), event.getStream(), streamAuthorityInfo);
+ StreamPushItem transform = StreamPushItem.getInstance(event, userSetting.getServerId());
+ transform.setPushIng(true);
+ transform.setUpdateTime(DateUtil.getNow());
+ transform.setPushTime(DateUtil.getNow());
+ transform.setSelf(true);
+ StreamPushItem pushInDb = getPush(event.getApp(), event.getStream());
+ if (pushInDb == null) {
+ transform.setCreateTime(DateUtil.getNow());
+ streamPushMapper.add(transform);
+ }else {
+ streamPushMapper.update(transform);
+ gbStreamMapper.updateMediaServer(event.getApp(), event.getStream(), event.getMediaServer().getId());
+ }
+ // TODO 鐩稿叧鐨勪簨浠惰嚜琛岀鐞嗭紝涓嶉渶瑕佸啓鍏LMMediaListManager
+// ChannelOnlineEvent channelOnlineEventLister = getChannelOnlineEventLister(transform.getApp(), transform.getStream());
+// if ( channelOnlineEventLister != null) {
+// try {
+// channelOnlineEventLister.run(transform.getApp(), transform.getStream(), transform.getServerId());;
+// } catch (ParseException e) {
+// logger.error("addPush: ", e);
+// }
+// removedChannelOnlineEventLister(transform.getApp(), transform.getStream());
+// }
+ // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+ redisCatchStorage.addPushListItem(event.getApp(), event.getStream(), event);
+
+ // 鍙戦�佹祦鍙樺寲redis娑堟伅
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("serverId", userSetting.getServerId());
+ jsonObject.put("app", event.getApp());
+ jsonObject.put("stream", event.getStream());
+ jsonObject.put("register", true);
+ jsonObject.put("mediaServerId", event.getMediaServer().getId());
+ redisCatchStorage.sendStreamChangeMsg(OriginType.values()[event.getMediaInfo().getOriginType()].getType(), jsonObject);
+ }
+
+ /**
+ * 娴佺寮�鐨勫鐞�
+ */
+ @Async("taskExecutor")
+ @EventListener
+ public void onApplicationEvent(MediaDepartureEvent event) {
+ // 鍏煎娴佹敞閿�鏃剁被鍨嬩粠redis璁板綍鑾峰彇
+ MediaInfo mediaInfo = redisCatchStorage.getStreamInfo(
+ event.getApp(), event.getStream(), event.getMediaServer().getId());
+ if (mediaInfo != null) {
+ String type = OriginType.values()[mediaInfo.getOriginType()].getType();
+ redisCatchStorage.removeStream(event.getMediaServer().getId(), type, event.getApp(), event.getStream());
+ if ("PUSH".equalsIgnoreCase(type)) {
+ // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+ redisCatchStorage.removePushListItem(event.getApp(), event.getStream(), event.getMediaServer().getId());
+ }
+ if (type != null) {
+ // 鍙戦�佹祦鍙樺寲redis娑堟伅
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("serverId", userSetting.getServerId());
+ jsonObject.put("app", event.getApp());
+ jsonObject.put("stream", event.getStream());
+ jsonObject.put("register", false);
+ jsonObject.put("mediaServerId", event.getMediaServer().getId());
+ redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+ }
+ }
+ GbStream gbStream = gbStreamMapper.selectOne(event.getApp(), event.getStream());
+ if (gbStream != null) {
+ if (userSetting.isUsePushingAsStatus()) {
+ streamPushMapper.updatePushStatus(event.getApp(), event.getStream(), false);
+ eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
+ }
+ }else {
+ streamPushMapper.del(event.getApp(), event.getStream());
+ }
+ }
+
+
+ private List<StreamPushItem> handleJSON(List<StreamInfo> streamInfoList) {
+ if (streamInfoList == null || streamInfoList.isEmpty()) {
+ return null;
+ }
Map<String, StreamPushItem> result = new HashMap<>();
-
- List<MediaItem> mediaItems = JSON.parseObject(jsonData, new TypeReference<List<MediaItem>>() {});
- for (MediaItem item : mediaItems) {
-
+ for (StreamInfo streamInfo : streamInfoList) {
// 涓嶄繚瀛樺浗鏍囨帹鐞嗕互鍙婃媺娴佷唬鐞嗙殑娴�
- if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
- || item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
- || item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
- String key = item.getApp() + "_" + item.getStream();
+ if (streamInfo.getOriginType() == OriginType.RTSP_PUSH.ordinal()
+ || streamInfo.getOriginType() == OriginType.RTMP_PUSH.ordinal()
+ || streamInfo.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
+ String key = streamInfo.getApp() + "_" + streamInfo.getStream();
StreamPushItem streamPushItem = result.get(key);
if (streamPushItem == null) {
- streamPushItem = transform(item);
+ streamPushItem = streamPushItem.getInstance(streamInfo);
result.put(key, streamPushItem);
}
}
}
-
return new ArrayList<>(result.values());
}
+
@Override
- public StreamPushItem transform(MediaItem item) {
+ public StreamPushItem transform(OnStreamChangedHookParam item) {
StreamPushItem streamPushItem = new StreamPushItem();
streamPushItem.setApp(item.getApp());
streamPushItem.setMediaServerId(item.getMediaServerId());
@@ -104,11 +221,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;
}
@@ -128,31 +246,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;
}
@@ -162,15 +259,10 @@
gbStreamService.sendCatalogMsg(stream, CatalogEvent.DEL);
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) {
- if (mediaList.getInteger("code") == 0) {
- JSONArray data = mediaList.getJSONArray("data");
- if (data == null) {
- streamPushMapper.del(stream.getApp(), stream.getStream());
- }
- }
+ MediaServer mediaInfo = mediaServerService.getOne(stream.getMediaServerId());
+ List<StreamInfo> mediaList = mediaServerService.getMediaList(mediaInfo, stream.getApp(), stream.getStream(), null);
+ if (mediaList != null && mediaList.isEmpty()) {
+ streamPushMapper.del(stream.getApp(), stream.getStream());
}
return del > 0;
}
@@ -178,21 +270,23 @@
@Override
public StreamPushItem getPush(String app, String streamId) {
-
return streamPushMapper.selectOne(app, streamId);
}
@Override
public boolean stop(String app, String streamId) {
+ logger.info("[鎺ㄦ祦 ] 鍋滄娴侊細 {}/{}", app, streamId);
StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId);
- gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
+ if (streamPushItem != null) {
+ gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
+ }
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);
+ MediaServer mediaServerItem = mediaServerService.getOne(streamPushItem.getMediaServerId());
+ mediaServerService.closeStreams(mediaServerItem,app, streamId);
}
return true;
}
@@ -200,7 +294,7 @@
@Override
public void zlmServerOnline(String mediaServerId) {
// 鍚屾zlm鎺ㄦ祦淇℃伅
- MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+ MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId);
if (mediaServerItem == null) {
return;
}
@@ -208,72 +302,81 @@
List<StreamPushItem> pushList = getPushList(mediaServerId);
Map<String, StreamPushItem> pushItemMap = new HashMap<>();
// redis璁板綍
- List<MediaItem> mediaItems = redisCatchStorage.getStreams(mediaServerId, "PUSH");
- Map<String, MediaItem> streamInfoPushItemMap = new HashMap<>();
+ List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, "PUSH");
+ Map<String, MediaInfo> 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);
}
}
}
- if (mediaItems.size() > 0) {
- for (MediaItem mediaItem : mediaItems) {
- streamInfoPushItemMap.put(mediaItem.getApp() + mediaItem.getStream(), mediaItem);
+ if (mediaInfoList.size() > 0) {
+ for (MediaInfo mediaInfo : mediaInfoList) {
+ streamInfoPushItemMap.put(mediaInfo.getApp() + mediaInfo.getStream(), mediaInfo);
}
}
- zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
- if (mediaList == null) return;
- String dataStr = mediaList.getString("data");
-
- Integer code = mediaList.getInteger("code");
- List<StreamPushItem> streamPushItems = null;
- if (code == 0 ) {
- if (dataStr != null) {
- streamPushItems = handleJSON(dataStr, mediaServerItem);
- }
+ // 鑾峰彇鎵�鏈夋帹娴侀壌鏉冧俊鎭紝娓呯悊杩囨湡鐨�
+ List<StreamAuthorityInfo> allStreamAuthorityInfo = redisCatchStorage.getAllStreamAuthorityInfo();
+ Map<String, StreamAuthorityInfo> streamAuthorityInfoInfoMap = new HashMap<>();
+ for (StreamAuthorityInfo streamAuthorityInfo : allStreamAuthorityInfo) {
+ streamAuthorityInfoInfoMap.put(streamAuthorityInfo.getApp() + streamAuthorityInfo.getStream(), streamAuthorityInfo);
+ }
+ List<StreamInfo> mediaList = mediaServerService.getMediaList(mediaServerItem, null, null, null);
+ if (mediaList == null) {
+ return;
+ }
+ List<StreamPushItem> streamPushItems = handleJSON(mediaList);
+ if (streamPushItems != null) {
+ for (StreamPushItem streamPushItem : streamPushItems) {
+ pushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
+ streamInfoPushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
+ streamAuthorityInfoInfoMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
}
-
- if (streamPushItems != null) {
- for (StreamPushItem streamPushItem : streamPushItems) {
- pushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
- streamInfoPushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
- }
- }
- List<StreamPushItem> offlinePushItems = new ArrayList<>(pushItemMap.values());
- if (offlinePushItems.size() > 0) {
- String type = "PUSH";
- 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);
+ }
+ List<StreamPushItem> offlinePushItems = new ArrayList<>(pushItemMap.values());
+ if (offlinePushItems.size() > 0) {
+ String type = "PUSH";
+ 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();
}
- }else {
- streamPushMapper.delAll(offlinePushItems);
+ List<StreamPushItem> streamPushItemsSub = offlinePushItems.subList(i, toIndex);
+ streamPushMapper.delAll(streamPushItemsSub);
}
+ }else {
+ streamPushMapper.delAll(offlinePushItems);
+ }
+ }
+ Collection<MediaInfo> mediaInfos = streamInfoPushItemMap.values();
+ if (mediaInfos.size() > 0) {
+ String type = "PUSH";
+ for (MediaInfo mediaInfo : mediaInfos) {
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("serverId", userSetting.getServerId());
+ jsonObject.put("app", mediaInfo.getApp());
+ jsonObject.put("stream", mediaInfo.getStream());
+ jsonObject.put("register", false);
+ jsonObject.put("mediaServerId", mediaServerId);
+ redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+ // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+ redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", mediaInfo.getApp(), mediaInfo.getStream());
+ // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+ redisCatchStorage.removePushListItem(mediaInfo.getApp(), mediaInfo.getStream(), mediaServerItem.getId());
}
- Collection<MediaItem> offlineMediaItemList = streamInfoPushItemMap.values();
- if (offlineMediaItemList.size() > 0) {
- String type = "PUSH";
- for (MediaItem offlineMediaItem : offlineMediaItemList) {
- JSONObject jsonObject = new JSONObject();
- jsonObject.put("serverId", userSetup.getServerId());
- 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", offlineMediaItem.getApp(), offlineMediaItem.getStream());
- }
+ }
+
+ Collection<StreamAuthorityInfo> streamAuthorityInfos = streamAuthorityInfoInfoMap.values();
+ if (streamAuthorityInfos.size() > 0) {
+ for (StreamAuthorityInfo streamAuthorityInfo : streamAuthorityInfos) {
+ // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+ redisCatchStorage.removeStreamAuthorityInfo(streamAuthorityInfo.getApp(), streamAuthorityInfo.getStream());
}
- }));
+ }
}
@Override
@@ -283,22 +386,26 @@
streamPushMapper.deleteWithoutGBId(mediaServerId);
gbStreamMapper.deleteWithoutGBId("push", mediaServerId);
// 鍏朵粬鐨勬祦璁剧疆鏈惎鐢�
- gbStreamMapper.updateStatusByMediaServerId(mediaServerId, false);
+ streamPushMapper.updateStatusByMediaServerId(mediaServerId, false);
+ streamProxyMapper.updateStatusByMediaServerId(mediaServerId, false);
// 鍙戦�佹祦鍋滄娑堟伅
String type = "PUSH";
// 鍙戦�乺edis娑堟伅
- List<MediaItem> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
- if (streamInfoList.size() > 0) {
- for (MediaItem mediaItem : streamInfoList) {
+ List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+ if (mediaInfoList.size() > 0) {
+ for (MediaInfo mediaInfo : mediaInfoList) {
// 绉婚櫎redis鍐呮祦鐨勪俊鎭�
- redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
+ redisCatchStorage.removeStream(mediaServerId, type, mediaInfo.getApp(), mediaInfo.getStream());
JSONObject jsonObject = new JSONObject();
- jsonObject.put("serverId", userSetup.getServerId());
- jsonObject.put("app", mediaItem.getApp());
- jsonObject.put("stream", mediaItem.getStream());
+ jsonObject.put("serverId", userSetting.getServerId());
+ jsonObject.put("app", mediaInfo.getApp());
+ jsonObject.put("stream", mediaInfo.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+
+ // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+ redisCatchStorage.removePushListItem(mediaInfo.getApp(), mediaInfo.getStream(), mediaServerId);
}
}
}
@@ -316,7 +423,7 @@
streamPushItem.setStreamType("push");
streamPushItem.setStatus(true);
streamPushItem.setGbId("34020000004111" + gbId);
- streamPushItem.setCreateStamp(System.currentTimeMillis());
+ streamPushItem.setCreateTime(DateUtil.getNow());
gbId ++;
}
int limitCount = 30;
@@ -339,39 +446,15 @@
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)
+ .filter(streamPushItem-> streamPushItem.getGbId() != null)
.collect(Collectors.toList());
// 瀛樺偍鏁版嵁鍒癵b_stream琛紝 id浼氳繑鍥炲埌streamPushItemForGbStream閲�
if (streamPushItemForGbStream.size() > 0) {
@@ -411,43 +494,40 @@
// 閬嶅巻瀛樺偍缁撴灉锛屾煡鎵綼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
- // 涓嶅瓨鍦ㄨ繖涓钩鍙帮紝鍒欏拷鐣ュ鍏ユ鍏宠仈鍏崇郴
- if (platformInfoMap.get(platFormInfoArray[0]) == null
- || platformInfoMap.get(platFormInfoArray[0]).get(platFormInfoArray[1]) == null) {
- logger.info("瀵煎叆鏁版嵁鏃朵笉瀛樺湪骞冲彴鎴栫洰褰晎}/{},宸插鍏ユ湭鍒嗛厤", platFormInfoArray[0], platFormInfoArray[1] );
- continue;
- }
- 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 (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);
}
}
}
- if (streamPushItemListFroPlatform.size() > 0) {
+ if (!streamPushItemListFroPlatform.isEmpty()) {
platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
// 鍙戦�侀�氱煡
for (String platformId : platformForEvent.keySet()) {
@@ -455,7 +535,6 @@
platformId, platformForEvent.get(platformId), CatalogEvent.ADD);
}
}
-
}
}
@@ -471,11 +550,104 @@
int delStream = streamPushMapper.delAllForGbStream(gbStreams);
if (delStream > 0) {
for (GbStream gbStream : gbStreams) {
- MediaServerItem mediaServerItem = mediaServerService.getOne(gbStream.getMediaServerId());
- zlmresTfulUtils.closeStreams(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+ MediaServer mediaServerItem = mediaServerService.getOne(gbStream.getMediaServerId());
+ mediaServerService.closeStreams(mediaServerItem, gbStream.getApp(), gbStream.getStream());
}
}
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());
+ stream.setMediaServerId(mediaConfig.getId());
+ stream.setSelf(true);
+ stream.setPushIng(true);
+
+ // 鏀惧湪浜嬪姟鍐呮墽琛�
+ 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();
+ }
+
+ @Override
+ public ResourceBaseInfo getOverview() {
+ int total = streamPushMapper.getAllCount();
+ int online = streamPushMapper.getAllOnline(userSetting.isUsePushingAsStatus());
+
+ return new ResourceBaseInfo(total, online);
+ }
+
+ @Override
+ public Map<String, StreamPushItem> getAllAppAndStreamMap() {
+ return streamPushMapper.getAllAppAndStreamMap();
+ }
+
+ @Override
+ public void updatePush(OnStreamChangedHookParam param) {
+ StreamPushItem transform = transform(param);
+ StreamPushItem pushInDb = getPush(param.getApp(), param.getStream());
+ transform.setPushIng(param.isRegist());
+ transform.setUpdateTime(DateUtil.getNow());
+ transform.setPushTime(DateUtil.getNow());
+ transform.setSelf(userSetting.getServerId().equals(param.getSeverId()));
+ if (pushInDb == null) {
+ transform.setCreateTime(DateUtil.getNow());
+ streamPushMapper.add(transform);
+ }else {
+ streamPushMapper.update(transform);
+ gbStreamMapper.updateMediaServer(param.getApp(), param.getStream(), param.getMediaServerId());
+ }
+ }
}
--
Gitblit v1.8.0