From b6fa459bc3ae287abcb3292a873718fd08297e61 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 10 十二月 2021 18:38:58 +0800
Subject: [PATCH] 修复推流列表大数据量时批量删除的错误，修复推流列表分页查询错误
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 125 ++++++++++++++++++++++++++++++++++++++---
1 files changed, 115 insertions(+), 10 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 7c17c2a..be34fae 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
@@ -3,11 +3,15 @@
import com.alibaba.fastjson.JSON;
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.GbStream;
+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.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamPushService;
@@ -20,10 +24,7 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
@Service
public class StreamPushServiceImpl implements IStreamPushService {
@@ -44,6 +45,9 @@
private IRedisCatchStorage redisCatchStorage;
@Autowired
+ private UserSetup userSetup;
+
+ @Autowired
private IMediaServerService mediaServerService;
@Override
@@ -56,7 +60,9 @@
for (MediaItem item : mediaItems) {
// 涓嶄繚瀛樺浗鏍囨帹鐞嗕互鍙婃媺娴佷唬鐞嗙殑娴�
- if (item.getOriginType() == 1 || item.getOriginType() == 2 || item.getOriginType() == 8) {
+ 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();
StreamPushItem streamPushItem = result.get(key);
if (streamPushItem == null) {
@@ -95,6 +101,11 @@
PageHelper.startPage(page, count);
List<StreamPushItem> all = streamPushMapper.selectAll();
return new PageInfo<>(all);
+ }
+
+ @Override
+ public List<StreamPushItem> getPushList(String mediaServerId) {
+ return streamPushMapper.selectAllByMediaServerId(mediaServerId);
}
@Override
@@ -137,17 +148,111 @@
}
@Override
- public void zlmServerOnline(ZLMServerConfig zlmServerConfig) {
- // 浼间箮娌″暐闇�瑕佸仛鐨�
+ public void zlmServerOnline(String mediaServerId) {
+ // 鍚屾zlm鎺ㄦ祦淇℃伅
+ MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+ if (mediaServerItem == null) {
+ return;
+ }
+ // 鏁版嵁搴撹褰�
+ 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<>();
+ 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);
+ }
+ }
+ 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);
+ }
+ }
+
+ 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);
+ }
+ }else {
+ streamPushMapper.delAll(offlinePushItems);
+ }
+
+ }
+ Collection<StreamInfo> offlineStreamInfoItems = streamInfoPushItemMap.values();
+ if (offlineStreamInfoItems.size() > 0) {
+ String type = "PUSH";
+ for (StreamInfo offlineStreamInfoItem : offlineStreamInfoItems) {
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("serverId", userSetup.getServerId());
+ jsonObject.put("app", offlineStreamInfoItem.getApp());
+ jsonObject.put("stream", offlineStreamInfoItem.getStreamId());
+ jsonObject.put("register", false);
+ jsonObject.put("mediaServerId", mediaServerId);
+ redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+ // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+ redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineStreamInfoItem.getApp(), offlineStreamInfoItem.getStreamId());
+ }
+ }
+ }));
}
@Override
public void zlmServerOffline(String mediaServerId) {
- // 绉婚櫎娌℃湁serverId鐨勬帹娴�
+ List<StreamPushItem> streamPushItems = streamPushMapper.selectAllByMediaServerId(mediaServerId);
+ // 绉婚櫎娌℃湁GBId鐨勬帹娴�
streamPushMapper.deleteWithoutGBId(mediaServerId);
+ gbStreamMapper.deleteWithoutGBId("push", mediaServerId);
// 鍏朵粬鐨勬祦璁剧疆鏈惎鐢�
gbStreamMapper.updateStatusByMediaServerId(mediaServerId, false);
- // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
- redisCatchStorage.removeStream(mediaServerId, "PUSH");
+ // 鍙戦�佹祦鍋滄娑堟伅
+ String type = "PUSH";
+ // 鍙戦�乺edis娑堟伅
+ List<StreamInfo> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+ if (streamInfoList.size() > 0) {
+ for (StreamInfo streamInfo : streamInfoList) {
+ // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+ redisCatchStorage.removeStream(mediaServerId, type, streamInfo.getApp(), streamInfo.getStreamId());
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("serverId", userSetup.getServerId());
+ jsonObject.put("app", streamInfo.getApp());
+ jsonObject.put("stream", streamInfo.getStreamId());
+ jsonObject.put("register", false);
+ jsonObject.put("mediaServerId", mediaServerId);
+ redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+ }
+ }
+ }
+
+ @Override
+ public void clean() {
+
}
}
--
Gitblit v1.8.0