XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化hook订阅机制

Browse Source
This commit is contained in:
648540858
2022-07-28 16:18:41 +08:00
parent f84eebdb75
commit fd091e545b
12 changed files with 296 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
View File

@@ -102,7 +102,7 @@ public class ZLMHttpHookListener {
logger.debug("[ ZLM HOOK ] on_server_keepalive API调用，参数：" + json.toString());
}
String mediaServerId = json.getString("mediaServerId");
List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(ZLMHttpHookSubscribe.HookType.on_server_keepalive);
List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive);
if (subscribes != null && subscribes.size() > 0) {
for (ZLMHttpHookSubscribe.Event subscribe : subscribes) {
subscribe.response(null, json);
@@ -168,7 +168,7 @@ public class ZLMHttpHookListener {
logger.debug("[ ZLM HOOK ]on_play API调用，参数：" + JSON.toJSONString(param));
}
String mediaServerId = param.getMediaServerId();
ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_play, json);
ZLMHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_play, json);
if (subscribe != null ) {
MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
if (mediaInfo != null) {
@@ -253,7 +253,7 @@ public class ZLMHttpHookListener {
}
ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, json);
ZLMHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json);
if (subscribe != null) {
if (mediaInfo != null) {
subscribe.response(mediaInfo, json);
@@ -377,7 +377,7 @@ public class ZLMHttpHookListener {
logger.debug("[ ZLM HOOK ]on_shell_login API调用，参数：" + json.toString());
}
String mediaServerId = json.getString("mediaServerId");
ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_shell_login, json);
ZLMHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_shell_login, json);
if (subscribe != null ) {
MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
if (mediaInfo != null) {
@@ -403,7 +403,7 @@ public class ZLMHttpHookListener {
logger.info("[ ZLM HOOK ]on_stream_changed API调用，参数：" + JSONObject.toJSONString(item));
String mediaServerId = item.getMediaServerId();
JSONObject json = (JSONObject) JSON.toJSON(item);
ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, json);
ZLMHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json);
if (subscribe != null ) {
MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
if (mediaInfo != null) {
@@ -614,7 +614,7 @@ public class ZLMHttpHookListener {
}
String remoteAddr = request.getRemoteAddr();
jsonObject.put("ip", remoteAddr);
List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(ZLMHttpHookSubscribe.HookType.on_server_started);
List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_started);
if (subscribes != null && subscribes.size() > 0) {
for (ZLMHttpHookSubscribe.Event subscribe : subscribes) {
subscribe.response(null, jsonObject);

85
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookSubscribe.java
View File

@@ -1,12 +1,16 @@
package com.genersoft.iot.vmp.media.zlm;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.media.zlm.dto.HookType;
import com.genersoft.iot.vmp.media.zlm.dto.IHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
* @description:针对 ZLMediaServer的hook事件订阅
@@ -16,49 +20,38 @@ import java.util.concurrent.ConcurrentHashMap;
@Component
public class ZLMHttpHookSubscribe {
public enum HookType{
on_flow_report,
on_http_access,
on_play,
on_publish,
on_record_mp4,
on_rtsp_auth,
on_rtsp_realm,
on_shell_login,
on_stream_changed,
on_stream_none_reader,
on_stream_not_found,
on_server_started,
on_server_keepalive
}
@FunctionalInterface
public interface Event{
void response(MediaServerItem mediaServerItem, JSONObject response);
}
private Map<HookType, Map<JSONObject, ZLMHttpHookSubscribe.Event>> allSubscribes = new ConcurrentHashMap<>();
private Map<HookType, Map<IHookSubscribe, ZLMHttpHookSubscribe.Event>> allSubscribes = new ConcurrentHashMap<>();
public void addSubscribe(HookType type, JSONObject hookResponse, ZLMHttpHookSubscribe.Event event) {
allSubscribes.computeIfAbsent(type, k -> new ConcurrentHashMap<>()).put(hookResponse, event);
public void addSubscribe(IHookSubscribe hookSubscribe, ZLMHttpHookSubscribe.Event event) {
if (hookSubscribe.getExpires() == null) {
// 默认5分钟过期
Instant expiresInstant = Instant.now().plusSeconds(TimeUnit.MINUTES.toSeconds(5));
hookSubscribe.setExpires(expiresInstant);
}
allSubscribes.computeIfAbsent(hookSubscribe.getHookType(), k -> new ConcurrentHashMap<>()).put(hookSubscribe, event);
}
public ZLMHttpHookSubscribe.Event getSubscribe(HookType type, JSONObject hookResponse) {
public ZLMHttpHookSubscribe.Event sendNotify(HookType type, JSONObject hookResponse) {
ZLMHttpHookSubscribe.Event event= null;
Map<JSONObject, Event> eventMap = allSubscribes.get(type);
Map<IHookSubscribe, Event> eventMap = allSubscribes.get(type);
if (eventMap == null) {
return null;
}
for (JSONObject key : eventMap.keySet()) {
for (IHookSubscribe key : eventMap.keySet()) {
Boolean result = null;
for (String s : key.keySet()) {
for (String s : key.getContent().keySet()) {
if (result == null) {
result = key.getString(s).equals(hookResponse.getString(s));
result = key.getContent().getString(s).equals(hookResponse.getString(s));
}else {
if (key.getString(s) == null) {
if (key.getContent().getString(s) == null) {
continue;
}
result = result && key.getString(s).equals(hookResponse.getString(s));
result = result && key.getContent().getString(s).equals(hookResponse.getString(s));
}
}
@@ -69,26 +62,30 @@ public class ZLMHttpHookSubscribe {
return event;
}
public void removeSubscribe(HookType type, JSONObject hookResponse) {
Map<JSONObject, Event> eventMap = allSubscribes.get(type);
public void removeSubscribe(IHookSubscribe hookSubscribe) {
Map<IHookSubscribe, Event> eventMap = allSubscribes.get(hookSubscribe.getHookType());
if (eventMap == null) {
return;
}
Set<Map.Entry<JSONObject, Event>> entries = eventMap.entrySet();
Set<Map.Entry<IHookSubscribe, Event>> entries = eventMap.entrySet();
if (entries.size() > 0) {
List<Map.Entry<JSONObject, ZLMHttpHookSubscribe.Event>> entriesToRemove = new ArrayList<>();
for (Map.Entry<JSONObject, ZLMHttpHookSubscribe.Event> entry : entries) {
JSONObject key = entry.getKey();
List<Map.Entry<IHookSubscribe, ZLMHttpHookSubscribe.Event>> entriesToRemove = new ArrayList<>();
for (Map.Entry<IHookSubscribe, ZLMHttpHookSubscribe.Event> entry : entries) {
JSONObject content = entry.getKey().getContent();
if (content == null || content.size() == 0) {
entriesToRemove.add(entry);
continue;
}
Boolean result = null;
for (String s : key.keySet()) {
for (String s : content.keySet()) {
if (result == null) {
result = key.getString(s).equals(hookResponse.getString(s));
result = content.getString(s).equals(hookSubscribe.getContent().getString(s));
}else {
if (key.getString(s) == null) {
if (content.getString(s) == null) {
continue;
}
result = result && key.getString(s).equals(hookResponse.getString(s));
result = result && content.getString(s).equals(hookSubscribe.getContent().getString(s));
}
}
if (null != result && result){
@@ -97,7 +94,7 @@ public class ZLMHttpHookSubscribe {
}
if (!CollectionUtils.isEmpty(entriesToRemove)) {
for (Map.Entry<JSONObject, ZLMHttpHookSubscribe.Event> entry : entriesToRemove) {
for (Map.Entry<IHookSubscribe, ZLMHttpHookSubscribe.Event> entry : entriesToRemove) {
entries.remove(entry);
}
}
@@ -111,17 +108,25 @@ public class ZLMHttpHookSubscribe {
* @return
*/
public List<ZLMHttpHookSubscribe.Event> getSubscribes(HookType type) {
// ZLMHttpHookSubscribe.Event event= null;
Map<JSONObject, Event> eventMap = allSubscribes.get(type);
Map<IHookSubscribe, Event> eventMap = allSubscribes.get(type);
if (eventMap == null) {
return null;
}
List<ZLMHttpHookSubscribe.Event> result = new ArrayList<>();
for (JSONObject key : eventMap.keySet()) {
for (IHookSubscribe key : eventMap.keySet()) {
result.add(eventMap.get(key));
}
return result;
}
public List<IHookSubscribe> getAll(){
ArrayList<IHookSubscribe> result = new ArrayList<>();
Collection<Map<IHookSubscribe, Event>> values = allSubscribes.values();
for (Map<IHookSubscribe, Event> value : values) {
result.addAll(value.keySet());
}
return result;
}
}

33
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java
View File

@@ -6,22 +6,22 @@ import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForServerStarted;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.TimeUnit;
@Component
@Order(value=1)
@@ -37,18 +37,12 @@ public class ZLMRunner implements CommandLineRunner {
@Autowired
private ZLMHttpHookSubscribe hookSubscribe;
@Autowired
private IStreamProxyService streamProxyService;
@Autowired
private EventPublisher publisher;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private MediaConfig mediaConfig;
@@ -67,17 +61,25 @@ public class ZLMRunner implements CommandLineRunner {
mediaServerService.updateToDatabase(mediaSerItem);
}
mediaServerService.syncCatchFromDatabase();
HookSubscribeForServerStarted hookSubscribeForServerStarted = HookSubscribeFactory.on_server_started();
// Instant expiresInstant = Instant.now().plusSeconds(TimeUnit.SECONDS.toSeconds(60));
// hookSubscribeForStreamChange.setExpires(expiresInstant);
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_server_started,new JSONObject(),
hookSubscribe.addSubscribe(hookSubscribeForServerStarted,
(MediaServerItem mediaServerItem, JSONObject response)->{
ZLMServerConfig zlmServerConfig = JSONObject.toJavaObject(response, ZLMServerConfig.class);
if (zlmServerConfig !=null ) {
if (startGetMedia != null) {
startGetMedia.remove(zlmServerConfig.getGeneralMediaServerId());
if (startGetMedia.size() == 0) {
hookSubscribe.removeSubscribe(HookSubscribeFactory.on_server_started());
}
}
}
});
// 获取zlm信息
logger.info("[zlm] 等待默认zlm中...");
@@ -103,7 +105,6 @@ public class ZLMRunner implements CommandLineRunner {
}
startGetMedia = null;
}
hookSubscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_server_started, new JSONObject());
// TODO 清理数据库中与redis不匹配的zlm
}, 60 * 1000 );
}
@@ -116,6 +117,9 @@ public class ZLMRunner implements CommandLineRunner {
zlmServerConfigFirst.setIp(mediaServerItem.getIp());
zlmServerConfigFirst.setHttpPort(mediaServerItem.getHttpPort());
startGetMedia.remove(mediaServerItem.getId());
if (startGetMedia.size() == 0) {
hookSubscribe.removeSubscribe(HookSubscribeFactory.on_server_started());
}
mediaServerService.zlmServerOnline(zlmServerConfigFirst);
}else {
logger.info("[ {} ]-[ {}:{} ]主动连接失败, 清理相关资源， 开始尝试重试连接",
@@ -130,6 +134,9 @@ public class ZLMRunner implements CommandLineRunner {
zlmServerConfig.setIp(mediaServerItem.getIp());
zlmServerConfig.setHttpPort(mediaServerItem.getHttpPort());
startGetMedia.remove(mediaServerItem.getId());
if (startGetMedia.size() == 0) {
hookSubscribe.removeSubscribe(HookSubscribeFactory.on_server_started());
}
mediaServerService.zlmServerOnline(zlmServerConfig);
}
}, 2000);

33
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeFactory.java Normal file
View File

@@ -0,0 +1,33 @@
package com.genersoft.iot.vmp.media.zlm.dto;
import com.alibaba.fastjson.JSONObject;
/**
* hook 订阅工厂
* @author lin
*/
public class HookSubscribeFactory {
public static HookSubscribeForStreamChange on_stream_changed(String app, String stream, boolean regist, String scheam, String mediaServerId) {
HookSubscribeForStreamChange hookSubscribe = new HookSubscribeForStreamChange();
JSONObject subscribeKey = new com.alibaba.fastjson.JSONObject();
subscribeKey.put("app", app);
subscribeKey.put("stream", stream);
subscribeKey.put("regist", regist);
if (scheam != null) {
subscribeKey.put("schema", scheam);
}
subscribeKey.put("mediaServerId", mediaServerId);
hookSubscribe.setContent(subscribeKey);
return hookSubscribe;
}
public static HookSubscribeForServerStarted on_server_started() {
HookSubscribeForServerStarted hookSubscribe = new HookSubscribeForServerStarted();
hookSubscribe.setContent(new JSONObject());
return hookSubscribe;
}
}

44
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeForServerStarted.java Normal file
View File

@@ -0,0 +1,44 @@
package com.genersoft.iot.vmp.media.zlm.dto;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.annotation.JSONField;
import java.time.Instant;
/**
* hook订阅-流变化
* @author lin
*/
public class HookSubscribeForServerStarted implements IHookSubscribe{
private HookType hookType = HookType.on_server_started;
private JSONObject content;
@JSONField(format="yyyy-MM-dd HH:mm:ss")
private Instant expires;
@Override
public HookType getHookType() {
return hookType;
}
@Override
public JSONObject getContent() {
return content;
}
public void setContent(JSONObject content) {
this.content = content;
}
@Override
public Instant getExpires() {
return expires;
}
@Override
public void setExpires(Instant expires) {
this.expires = expires;
}
}

43
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeForStreamChange.java Normal file
View File

@@ -0,0 +1,43 @@
package com.genersoft.iot.vmp.media.zlm.dto;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.annotation.JSONField;
import java.time.Instant;
/**
* hook订阅-流变化
* @author lin
*/
public class HookSubscribeForStreamChange implements IHookSubscribe{
private HookType hookType = HookType.on_stream_changed;
private JSONObject content;
private Instant expires;
@Override
public HookType getHookType() {
return hookType;
}
@Override
public JSONObject getContent() {
return content;
}
public void setContent(JSONObject content) {
this.content = content;
}
@Override
public Instant getExpires() {
return expires;
}
@Override
public void setExpires(Instant expires) {
this.expires = expires;
}
}

23
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookType.java Normal file
View File

@@ -0,0 +1,23 @@
package com.genersoft.iot.vmp.media.zlm.dto;
/**
* hook类型
* @author lin
*/
public enum HookType {
on_flow_report,
on_http_access,
on_play,
on_publish,
on_record_mp4,
on_rtsp_auth,
on_rtsp_realm,
on_shell_login,
on_stream_changed,
on_stream_none_reader,
on_stream_not_found,
on_server_started,
on_server_keepalive
}

36
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/IHookSubscribe.java Normal file
View File

@@ -0,0 +1,36 @@
package com.genersoft.iot.vmp.media.zlm.dto;
import com.alibaba.fastjson.JSONObject;
import java.time.Instant;
/**
* zlm hook事件的参数
* @author lin
*/
public interface IHookSubscribe {
/**
* 获取hook类型
* @return hook类型
*/
HookType getHookType();
/**
* 获取hook的具体内容
* @return hook的具体内容
*/
JSONObject getContent();
/**
* 设置过期时间
* @param instant 过期时间
*/
void setExpires(Instant instant);
/**
* 获取过期时间
* @return 过期时间
*/
Instant getExpires();
}