Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'wvp-28181-2.0' into wvp-28181-2.0

Browse Source
This commit is contained in:
648540858 2022-07-28 16:24:51 +08:00 committed by GitHub
commit 646c3e41b7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 319 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
View File

@ -10,6 +10,9 @@ import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderProvider;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.HookType;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
@ -348,25 +351,19 @@ public class SIPCommander implements ISIPCommander {
@Override
public void playStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
ZLMHttpHookSubscribe.Event event, SipSubscribe.Event okEvent, SipSubscribe.Event errorEvent) {
String streamId = ssrcInfo.getStream();
String stream = ssrcInfo.getStream();
try {
if (device == null) {
return;
}
String streamMode = device.getStreamMode().toUpperCase();
logger.info("{} 分配的ZLM为: {} [{}:{}]", streamId, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", streamId);
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
logger.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtmp", mediaServerItem.getId());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
if (event != null) {
event.response(mediaServerItemInUse, json);
subscribe.removeSubscribe(hookSubscribe);
}
});
//
@ -440,7 +437,7 @@ public class SIPCommander implements ISIPCommander {
errorEvent.response(e);
}), e ->{
// 这里为例避免一个通道的点播只有一个callID这个参数使用一个固定值
streamSession.put(device.getDeviceId(), channelId ,"play", streamId, ssrcInfo.getSsrc(), mediaServerItem.getId(), ((ResponseEvent)e.event).getClientTransaction(), VideoStreamSessionManager.SessionType.play);
streamSession.put(device.getDeviceId(), channelId ,"play", stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), ((ResponseEvent)e.event).getClientTransaction(), VideoStreamSessionManager.SessionType.play);
streamSession.put(device.getDeviceId(), channelId ,"play", e.dialog);
okEvent.response(e);
});
@ -530,21 +527,14 @@ public class SIPCommander implements ISIPCommander {
CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtmp", mediaServerItem.getId());
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey);
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
if (hookEvent != null) {
InviteStreamInfo inviteStreamInfo = new InviteStreamInfo(mediaServerItemInUse, json, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream());
hookEvent.call(inviteStreamInfo);
}
subscribe.removeSubscribe(hookSubscribe);
});
Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());
@ -643,21 +633,15 @@ public class SIPCommander implements ISIPCommander {
CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, null, mediaServerItem.getId());
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", ssrcInfo.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("mediaServerId", mediaServerItem.getId());
logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
hookEvent.call(new InviteStreamInfo(mediaServerItem, json, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream()));
subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
subscribeKey.put("regist", false);
subscribeKey.put("schema", "rtmp");
subscribe.removeSubscribe(hookSubscribe);
hookSubscribe.getContent().put("regist", false);
hookSubscribe.getContent().put("schema", "rtmp");
// 添加流注销的订阅，注销了后向设备发送bye
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
subscribe.addSubscribe(hookSubscribe,
(MediaServerItem mediaServerItemForEnd, JSONObject jsonForEnd)->{
ClientTransaction transaction = streamSession.getTransaction(device.getDeviceId(), channelId, ssrcInfo.getStream(), callIdHeader.getCallId());
if (transaction != null) {

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);

86
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,51 +20,39 @@ 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));
}
}
if (null != result && result) {
event = eventMap.get(key);
@ -69,26 +61,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 +93,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 +107,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();
}

38
src/main/java/com/genersoft/iot/vmp/service/StreamGPSSubscribeTask.java
View File

@ -1,38 +0,0 @@
package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* 定时查找redis中的GPS推送消息，并保存到对应的流中
*/
@Component
public class StreamGPSSubscribeTask {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IVideoManagerStorage storager;
@Scheduled(fixedRate = 30 * 1000) //每30秒执行一次
public void execute(){
List<GPSMsgInfo> gpsMsgInfo = redisCatchStorage.getAllGpsMsgInfo();
if (gpsMsgInfo.size() > 0) {
storager.updateStreamGPS(gpsMsgInfo);
for (GPSMsgInfo msgInfo : gpsMsgInfo) {
msgInfo.setStored(true);
redisCatchStorage.updateGpsMsgInfo(msgInfo);
}
}
}
}

18
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@ -35,6 +35,10 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.HookType;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
@ -305,16 +309,10 @@ public class PlayServiceImpl implements IPlayService {
// 单端口模式streamId也有变化，需要重新设置监听
if (!mediaServerItem.isRtpEnable()) {
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", "rtp");
subscribeKey.put("stream", stream);
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed,subscribeKey);
subscribeKey.put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject response)->{
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtmp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{
logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + response.toJSONString());
dynamicTask.stop(timeOutTaskKey);
// hook响应

14
src/main/java/com/genersoft/iot/vmp/service/impl/RedisGbPlayMsgListener.java
View File

@ -8,6 +8,9 @@ import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.HookType;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.*;
@ -270,14 +273,9 @@ public class RedisGbPlayMsgListener implements MessageListener {
}, userSetting.getPlatformPlayTimeout());
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", content.getApp());
subscribeKey.put("stream", content.getStream());
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(content.getApp(), content.getStream(), true, "rtmp", mediaServerItem.getId());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
dynamicTask.stop(taskKey);
responseSendItem(mediaServerItem, content, toId, serial);
});

22
src/main/java/com/genersoft/iot/vmp/service/impl/RedisGpsMsgListener.java
View File

@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
import com.genersoft.iot.vmp.gb28181.bean.HandlerCatchData;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -11,9 +12,11 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
@ -30,6 +33,9 @@ public class RedisGpsMsgListener implements MessageListener {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IVideoManagerStorage storager;
private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@ -46,10 +52,26 @@ public class RedisGpsMsgListener implements MessageListener {
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
GPSMsgInfo gpsMsgInfo = JSON.parseObject(msg.getBody(), GPSMsgInfo.class);
// 只是放入redis缓存起来
redisCatchStorage.updateGpsMsgInfo(gpsMsgInfo);
}
taskQueueHandlerRun = false;
});
}
}
/**
* 定时将经纬度更新到数据库
*/
@Scheduled(fixedRate = 2 * 1000) //每2秒执行一次
public void execute(){
List<GPSMsgInfo> gpsMsgInfo = redisCatchStorage.getAllGpsMsgInfo();
if (gpsMsgInfo.size() > 0) {
storager.updateStreamGPS(gpsMsgInfo);
for (GPSMsgInfo msgInfo : gpsMsgInfo) {
msgInfo.setStored(true);
redisCatchStorage.updateGpsMsgInfo(msgInfo);
}
}
}
}

16
src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java
View File

@ -8,6 +8,8 @@ import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.VersionInfo;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.IHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.utils.SpringBeanFactory;
@ -38,7 +40,7 @@ import java.util.Set;
public class ServerController {
@Autowired
private ConfigurableApplicationContext context;
private ZLMHttpHookSubscribe zlmHttpHookSubscribe;
@Autowired
private IMediaServerService mediaServerService;
@ -254,6 +256,18 @@ public class ServerController {
return result;
}
@ApiOperation("获取当前所有hook")
@GetMapping(value = "/hooks")
@ResponseBody
public WVPResult<List<IHookSubscribe>> getHooks(){
WVPResult<List<IHookSubscribe>> result = new WVPResult<>();
result.setCode(0);
result.setMsg("success");
List<IHookSubscribe> all = zlmHttpHookSubscribe.getAll();
result.setData(all);
return result;
}
// @ApiOperation("当前进行中的动态任务")
// @GetMapping(value = "/dynamicTask")
// @ResponseBody