From c21d973977a9f1d00d26179de764687ddd0ec56c Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 24 四月 2024 14:59:41 +0800
Subject: [PATCH] 修复收到catalog消息是更新导致是否有音频的设置失效的BUG
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 121 ++++++++++++++++++++++++++++++----------
1 files changed, 91 insertions(+), 30 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 ed59230..bb51edc
--- 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,9 +1,10 @@
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.JSON;
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import com.alibaba.fastjson2.TypeReference;
+import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
@@ -11,6 +12,8 @@
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.*;
+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;
@@ -18,6 +21,7 @@
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.*;
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;
@@ -27,12 +31,13 @@
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
-import org.springframework.util.StringUtils;
+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);
@@ -91,8 +96,8 @@
Map<String, StreamPushItem> result = new HashMap<>();
- List<MediaItem> mediaItems = JSON.parseObject(jsonData, new TypeReference<List<MediaItem>>() {});
- for (MediaItem item : mediaItems) {
+ List<OnStreamChangedHookParam> onStreamChangedHookParams = JSON.parseObject(jsonData, new TypeReference<List<OnStreamChangedHookParam>>() {});
+ for (OnStreamChangedHookParam item : onStreamChangedHookParams) {
// 涓嶄繚瀛樺浗鏍囨帹鐞嗕互鍙婃媺娴佷唬鐞嗙殑娴�
if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
@@ -110,7 +115,7 @@
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());
@@ -180,8 +185,11 @@
@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);
@@ -204,19 +212,25 @@
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<OnStreamChangedHookParam> onStreamChangedHookParams = redisCatchStorage.getStreams(mediaServerId, "PUSH");
+ Map<String, OnStreamChangedHookParam> 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 (onStreamChangedHookParams.size() > 0) {
+ for (OnStreamChangedHookParam onStreamChangedHookParam : onStreamChangedHookParams) {
+ streamInfoPushItemMap.put(onStreamChangedHookParam.getApp() + onStreamChangedHookParam.getStream(), onStreamChangedHookParam);
}
+ }
+ // 鑾峰彇鎵�鏈夋帹娴侀壌鏉冧俊鎭紝娓呯悊杩囨湡鐨�
+ List<StreamAuthorityInfo> allStreamAuthorityInfo = redisCatchStorage.getAllStreamAuthorityInfo();
+ Map<String, StreamAuthorityInfo> streamAuthorityInfoInfoMap = new HashMap<>();
+ for (StreamAuthorityInfo streamAuthorityInfo : allStreamAuthorityInfo) {
+ streamAuthorityInfoInfoMap.put(streamAuthorityInfo.getApp() + streamAuthorityInfo.getStream(), streamAuthorityInfo);
}
zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
if (mediaList == null) {
@@ -236,6 +250,7 @@
for (StreamPushItem streamPushItem : streamPushItems) {
pushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
streamInfoPushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
+ streamAuthorityInfoInfoMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
}
}
List<StreamPushItem> offlinePushItems = new ArrayList<>(pushItemMap.values());
@@ -256,19 +271,29 @@
}
}
- Collection<MediaItem> offlineMediaItemList = streamInfoPushItemMap.values();
- if (offlineMediaItemList.size() > 0) {
+ Collection<OnStreamChangedHookParam> offlineOnStreamChangedHookParamList = streamInfoPushItemMap.values();
+ if (offlineOnStreamChangedHookParamList.size() > 0) {
String type = "PUSH";
- for (MediaItem offlineMediaItem : offlineMediaItemList) {
+ for (OnStreamChangedHookParam offlineOnStreamChangedHookParam : offlineOnStreamChangedHookParamList) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetting.getServerId());
- jsonObject.put("app", offlineMediaItem.getApp());
- jsonObject.put("stream", offlineMediaItem.getStream());
+ jsonObject.put("app", offlineOnStreamChangedHookParam.getApp());
+ jsonObject.put("stream", offlineOnStreamChangedHookParam.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 绉婚櫎redis鍐呮祦鐨勪俊鎭�
- redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineMediaItem.getApp(), offlineMediaItem.getStream());
+ redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineOnStreamChangedHookParam.getApp(), offlineOnStreamChangedHookParam.getStream());
+ // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+ redisCatchStorage.removePushListItem(offlineOnStreamChangedHookParam.getApp(), offlineOnStreamChangedHookParam.getStream(), mediaServerItem.getId());
+ }
+ }
+
+ Collection<StreamAuthorityInfo> streamAuthorityInfos = streamAuthorityInfoInfoMap.values();
+ if (streamAuthorityInfos.size() > 0) {
+ for (StreamAuthorityInfo streamAuthorityInfo : streamAuthorityInfos) {
+ // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+ redisCatchStorage.removeStreamAuthorityInfo(streamAuthorityInfo.getApp(), streamAuthorityInfo.getStream());
}
}
}));
@@ -286,18 +311,21 @@
// 鍙戦�佹祦鍋滄娑堟伅
String type = "PUSH";
// 鍙戦�乺edis娑堟伅
- List<MediaItem> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+ List<OnStreamChangedHookParam> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
if (streamInfoList.size() > 0) {
- for (MediaItem mediaItem : streamInfoList) {
+ for (OnStreamChangedHookParam onStreamChangedHookParam : streamInfoList) {
// 绉婚櫎redis鍐呮祦鐨勪俊鎭�
- redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
+ redisCatchStorage.removeStream(mediaServerId, type, onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream());
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetting.getServerId());
- jsonObject.put("app", mediaItem.getApp());
- jsonObject.put("stream", mediaItem.getStream());
+ jsonObject.put("app", onStreamChangedHookParam.getApp());
+ jsonObject.put("stream", onStreamChangedHookParam.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+
+ // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+ redisCatchStorage.removePushListItem(onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream(), mediaServerId);
}
}
}
@@ -346,7 +374,7 @@
// 瀛樺偍鏁版嵁鍒皊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) {
@@ -419,7 +447,7 @@
}
}
- if (streamPushItemListFroPlatform.size() > 0) {
+ if (!streamPushItemListFroPlatform.isEmpty()) {
platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
// 鍙戦�侀�氱煡
for (String platformId : platformForEvent.keySet()) {
@@ -427,7 +455,6 @@
platformId, platformForEvent.get(platformId), CatalogEvent.ADD);
}
}
-
}
}
@@ -486,13 +513,16 @@
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 (!StringUtils.isEmpty(stream.getGbId())) {
+ if (!ObjectUtils.isEmpty(stream.getGbId())) {
stream.setStreamType("push");
gbStreamMapper.add(stream);
}
@@ -507,6 +537,37 @@
@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