From ab81136765f1b641223b982b2baef13e06307fe4 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 08 十二月 2021 16:45:50 +0800
Subject: [PATCH] 优化适配zlm的hook保活
---
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 196 ++++++++++++++++++++++++++++++++++++++++++------
1 files changed, 169 insertions(+), 27 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 ce7f02b..dcca0e5 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,22 +3,28 @@
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.StreamProxyItem;
+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;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
+import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.StreamPushMapper;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
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 {
@@ -29,8 +35,23 @@
@Autowired
private StreamPushMapper streamPushMapper;
+ @Autowired
+ private PlatformGbStreamMapper platformGbStreamMapper;
+
+ @Autowired
+ private ZLMRESTfulUtils zlmresTfulUtils;
+
+ @Autowired
+ private IRedisCatchStorage redisCatchStorage;
+
+ @Autowired
+ private UserSetup userSetup;
+
+ @Autowired
+ private IMediaServerService mediaServerService;
+
@Override
- public List<StreamPushItem> handleJSON(String jsonData) {
+ public List<StreamPushItem> handleJSON(String jsonData, MediaServerItem mediaServerItem) {
if (jsonData == null) return null;
Map<String, StreamPushItem> result = new HashMap<>();
@@ -39,31 +60,40 @@
for (MediaItem item : mediaItems) {
// 涓嶄繚瀛樺浗鏍囨帹鐞嗕互鍙婃媺娴佷唬鐞嗙殑娴�
- if (item.getOriginType() == 3 || item.getOriginType() == 4 || item.getOriginType() == 5) {
- continue;
+ 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) {
+ streamPushItem = transform(item);
+ result.put(key, streamPushItem);
+ }
}
- String key = item.getApp() + "_" + item.getStream();
- StreamPushItem streamPushItem = result.get(key);
- if (streamPushItem == null) {
- streamPushItem = new StreamPushItem();
- streamPushItem.setApp(item.getApp());
- 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.setAliveSecond(item.getAliveSecond());
- streamPushItem.setStatus(true);
- streamPushItem.setVhost(item.getVhost());
- result.put(key, streamPushItem);
- }
+
}
return new ArrayList<>(result.values());
+ }
+ @Override
+ public StreamPushItem transform(MediaItem item) {
+ StreamPushItem streamPushItem = new StreamPushItem();
+ streamPushItem.setApp(item.getApp());
+ 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.setAliveSecond(item.getAliveSecond());
+ streamPushItem.setStatus(true);
+ streamPushItem.setStreamType("push");
+ streamPushItem.setVhost(item.getVhost());
+ return streamPushItem;
}
@Override
@@ -71,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
@@ -84,6 +119,113 @@
@Override
public boolean removeFromGB(GbStream stream) {
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());
+ }
return del > 0;
}
+
+
+ @Override
+ public StreamPushItem getPush(String app, String streamId) {
+
+ return streamPushMapper.selectOne(app, streamId);
+ }
+
+ @Override
+ public boolean stop(String app, String streamId) {
+ StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId);
+ int delStream = streamPushMapper.del(app, streamId);
+ gbStreamMapper.del(app, streamId);
+ platformGbStreamMapper.delByAppAndStream(app, streamId);
+ if (delStream > 0) {
+ MediaServerItem mediaServerItem = mediaServerService.getOne(streamPushItem.getMediaServerId());
+ zlmresTfulUtils.closeStreams(mediaServerItem,app, streamId);
+ }
+ return true;
+ }
+
+ @Override
+ public void zlmServerOnline(String mediaServerId) {
+ // 鍚屾zlm鎺ㄦ祦淇℃伅
+ MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+ if (mediaServerItem == null) {
+ return;
+ }
+ List<StreamPushItem> pushList = getPushList(mediaServerId);
+ if (pushList.size() > 0) {
+ Map<String, StreamPushItem> pushItemMap = new HashMap<>();
+ for (StreamPushItem streamPushItem : pushList) {
+ pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem);
+ }
+ 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());
+ }
+ }
+ Collection<StreamPushItem> offlinePushItems = pushItemMap.values();
+ if (offlinePushItems.size() > 0) {
+ String type = "PUSH";
+ streamPushMapper.delAll(new ArrayList<>(offlinePushItems));
+ for (StreamPushItem offlinePushItem : offlinePushItems) {
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("serverId", userSetup.getServerId());
+ jsonObject.put("app", offlinePushItem.getApp());
+ jsonObject.put("stream", offlinePushItem.getStream());
+ jsonObject.put("register", false);
+ jsonObject.put("mediaServerId", mediaServerId);
+ redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+ // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+ redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlinePushItem.getApp(), offlinePushItem.getStream());
+ }
+ }
+ }));
+ }
+ }
+
+ @Override
+ public void zlmServerOffline(String mediaServerId) {
+ List<StreamPushItem> streamPushItems = streamPushMapper.selectAllByMediaServerId(mediaServerId);
+ // 绉婚櫎娌℃湁GBId鐨勬帹娴�
+ streamPushMapper.deleteWithoutGBId(mediaServerId);
+ gbStreamMapper.deleteWithoutGBId("push", mediaServerId);
+ // 鍏朵粬鐨勬祦璁剧疆鏈惎鐢�
+ gbStreamMapper.updateStatusByMediaServerId(mediaServerId, false);
+ // 鍙戦�佹祦鍋滄娑堟伅
+ String type = "PUSH";
+ // 鍙戦�乺edis娑堟伅
+ List<StreamInfo> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+ if (streamInfoList.size() > 0) {
+ for (StreamInfo streamInfo : streamInfoList) {
+ 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);
+ // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+ redisCatchStorage.removeStream(mediaServerId, type, streamInfo.getApp(), streamInfo.getStreamId());
+ }
+ }
+ }
+
+ @Override
+ public void clean() {
+
+ }
}
--
Gitblit v1.8.0