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

Merge branch 'wvp-28181-2.0'

Browse Source 
# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
This commit is contained in:
648540858
2022-09-01 14:50:28 +08:00
parent 8f7e8efc98 703c2e292a
commit 03d6ad289b
76 changed files with 979 additions and 904 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/main/java/com/genersoft/iot/vmp/service/IPlatformService.java Normal file
View File

@@ -0,0 +1,45 @@
package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.github.pagehelper.PageInfo;
/**
* 国标平台的业务类
* @author lin
*/
public interface IPlatformService {
ParentPlatform queryPlatformByServerGBId(String platformGbId);
/**
* 分页获取上级平台
* @param page
* @param count
* @return
*/
PageInfo<ParentPlatform> queryParentPlatformList(int page, int count);
/**
* 添加级联平台
* @param parentPlatform 级联平台
*/
boolean add(ParentPlatform parentPlatform);
/**
* 平台上线
* @param parentPlatform 平台信息
*/
void online(ParentPlatform parentPlatform);
/**
* 平台离线
* @param parentPlatform 平台信息
*/
void offline(ParentPlatform parentPlatform);
/**
* 向上级平台发起注册
* @param parentPlatform
*/
void login(ParentPlatform parentPlatform);
}

7
src/main/java/com/genersoft/iot/vmp/service/IPlayService.java
View File

@@ -6,7 +6,7 @@ import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
@@ -14,7 +14,6 @@ import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.AudioBroadcastEvent;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import org.springframework.http.ResponseEntity;
import org.springframework.web.context.request.async.DeferredResult;
/**
@@ -25,9 +24,9 @@ public interface IPlayService {
void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid);
void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
InviteTimeOutCallback timeoutCallback, String uuid);
PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent, Runnable timeoutCallback);
PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZlmHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent, Runnable timeoutCallback);
MediaServerItem getNewMediaServerItem(Device device);

61
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
View File

@@ -11,6 +11,7 @@ import java.util.Set;
import com.genersoft.iot.vmp.media.zlm.ZLMRunner;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import org.slf4j.Logger;
@@ -22,7 +23,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
@@ -42,7 +42,6 @@ import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.dao.MediaServerMapper;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import okhttp3.OkHttpClient;
import okhttp3.Request;
@@ -58,6 +57,8 @@ public class MediaServerServiceImpl implements IMediaServerService {
private final static Logger logger = LoggerFactory.getLogger(MediaServerServiceImpl.class);
private final String zlmKeepaliveKeyPrefix = "zlm-keepalive_";
@Autowired
private SipConfig sipConfig;
@@ -88,10 +89,12 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private EventPublisher publisher;
@Autowired
private DynamicTask dynamicTask;
/**
* 初始化
*/
@@ -135,7 +138,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
logger.info("media server [ {} ] ssrcConfig is null", mediaServerItem.getId());
return null;
}else {
String ssrc = null;
String ssrc;
if (presetSsrc != null) {
ssrc = presetSsrc;
}else {
@@ -404,15 +407,43 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (serverItem.isAutoConfig()) {
setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable()));
}
final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + serverItem.getId();
dynamicTask.stop(zlmKeepaliveKey);
dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (serverItem.getHookAliveInterval() + 5) * 1000);
publisher.zlmOnlineEventPublish(serverItem.getId());
logger.info("[ZLM] 连接成功 {} - {}:{} ",
zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort());
}
class KeepAliveTimeoutRunnable implements Runnable{
private MediaServerItem serverItem;
public KeepAliveTimeoutRunnable(MediaServerItem serverItem) {
this.serverItem = serverItem;
}
@Override
public void run() {
logger.info("[zlm心跳到期]：" + serverItem.getId());
// 发起http请求验证zlm是否确实无法连接，如果确实无法连接则发送离线事件，否则不作处理
JSONObject mediaServerConfig = zlmresTfulUtils.getMediaServerConfig(serverItem);
if (mediaServerConfig != null && mediaServerConfig.getInteger("code") == 0) {
logger.info("[zlm心跳到期]：{}验证后zlm仍在线，恢复心跳信息,请检查zlm是否可以正常向wvp发送心跳", serverItem.getId());
// 添加zlm信息
updateMediaServerKeepalive(serverItem.getId(), mediaServerConfig);
}else {
publisher.zlmOfflineEventPublish(serverItem.getId());
}
}
}
@Override
public void zlmServerOffline(String mediaServerId) {
delete(mediaServerId);
final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerId;
dynamicTask.stop(zlmKeepaliveKey);
}
@Override
@@ -423,7 +454,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (RedisUtil.zScore(key, serverItem.getId()) == null) { // 不存在则设置默认值 已存在则重置
RedisUtil.zAdd(key, serverItem.getId(), 0L);
// 查询服务流数量
zlmresTfulUtils.getMediaList(serverItem, null, null, "rtmp",(mediaList ->{
zlmresTfulUtils.getMediaList(serverItem, null, null, "rtsp",(mediaList ->{
Integer code = mediaList.getInteger("code");
if (code == 0) {
JSONArray data = mediaList.getJSONArray("data");
@@ -435,7 +466,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
}else {
clearRTPServer(serverItem);
}
}
@@ -471,7 +501,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
// 获取分数最低的，及并发最低的
Set<Object> objects = RedisUtil.ZRange(key, 0, -1);
Set<Object> objects = RedisUtil.zRange(key, 0, -1);
ArrayList<Object> mediaServerObjectS = new ArrayList<>(objects);
String mediaServerId = (String)mediaServerObjectS.get(0);
@@ -489,10 +519,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
String protocol = sslEnabled ? "https" : "http";
String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
String recordHookPrex = null;
if (mediaServerItem.getRecordAssistPort() != 0) {
recordHookPrex = String.format("http://127.0.0.1:%s/api/record", mediaServerItem.getRecordAssistPort());
}
Map<String, Object> param = new HashMap<>();
param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
param.put("ffmpeg.cmd","%s -fflags nobuffer -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s");
@@ -501,7 +528,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_play",String.format("%s/on_play", hookPrex));
param.put("hook.on_http_access",String.format("%s/on_http_access", hookPrex));
param.put("hook.on_publish", String.format("%s/on_publish", hookPrex));
param.put("hook.on_record_mp4",recordHookPrex != null? String.format("%s/on_record_mp4", recordHookPrex): "");
param.put("hook.on_record_ts",String.format("%s/on_record_ts", hookPrex));
param.put("hook.on_rtsp_auth",String.format("%s/on_rtsp_auth", hookPrex));
param.put("hook.on_rtsp_realm",String.format("%s/on_rtsp_realm", hookPrex));
@@ -511,6 +537,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
if (mediaServerItem.getRecordAssistPort() > 0) {
param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort()));
}else {
param.put("hook.on_record_mp4","");
}
param.put("hook.timeoutSec","20");
param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS()==-1?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() );
// 推流断开后可以在超时时间内重新连接上继续推流，这样播放器会接着播放。
@@ -631,9 +662,9 @@ public class MediaServerServiceImpl implements IMediaServerService {
return;
}
}
String key = VideoManagerConstants.MEDIA_SERVER_KEEPALIVE_PREFIX + userSetting.getServerId() + "_" + mediaServerId;
int hookAliveInterval = mediaServerItem.getHookAliveInterval() + 2;
RedisUtil.set(key, data, hookAliveInterval);
final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerItem.getId();
dynamicTask.stop(zlmKeepaliveKey);
dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(mediaServerItem), (mediaServerItem.getHookAliveInterval() + 5) * 1000);
}
private MediaServerItem getOneFromDatabase(String mediaServerId) {

3
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
View File

@@ -108,6 +108,7 @@ public class MediaServiceImpl implements IMediaService {
streamInfoResult.setWs_fmp4(String.format("ws://%s:%s/%s/%s.live.mp4%s", addr, mediaInfo.getHttpPort(), app, stream, callIdParam));
streamInfoResult.setTs(String.format("http://%s:%s/%s/%s.live.ts%s", addr, mediaInfo.getHttpPort(), app, stream, callIdParam));
streamInfoResult.setWs_ts(String.format("ws://%s:%s/%s/%s.live.ts%s", addr, mediaInfo.getHttpPort(), app, stream, callIdParam));
streamInfoResult.setRtc(String.format("http://%s:%s/index/api/webrtc?app=%s&stream=%s&type=play%s", mediaInfo.getStreamIp(), mediaInfo.getHttpPort(), app, stream, ObjectUtils.isEmpty(callId)?"":"&callId=" + callId));
if (mediaInfo.getHttpSSlPort() != 0) {
streamInfoResult.setHttps_flv(String.format("https://%s:%s/%s/%s.live.flv%s", addr, mediaInfo.getHttpSSlPort(), app, stream, callIdParam));
streamInfoResult.setWss_flv(String.format("wss://%s:%s/%s/%s.live.flv%s", addr, mediaInfo.getHttpSSlPort(), app, stream, callIdParam));
@@ -118,7 +119,7 @@ public class MediaServiceImpl implements IMediaService {
streamInfoResult.setHttps_ts(String.format("https://%s:%s/%s/%s.live.ts%s", addr, mediaInfo.getHttpSSlPort(), app, stream, callIdParam));
streamInfoResult.setWss_ts(String.format("wss://%s:%s/%s/%s.live.ts%s", addr, mediaInfo.getHttpSSlPort(), app, stream, callIdParam));
streamInfoResult.setWss_ts(String.format("wss://%s:%s/%s/%s.live.ts%s", addr, mediaInfo.getHttpSSlPort(), app, stream, callIdParam));
streamInfoResult.setRtc(String.format("https://%s:%s/index/api/webrtc?app=%s&stream=%s&type=%s%s", mediaInfo.getStreamIp(), mediaInfo.getHttpSSlPort(), app, stream, isPlay?"play":"push", ObjectUtils.isEmpty(callId)?"":"&callId=" + callId));
streamInfoResult.setRtcs(String.format("https://%s:%s/index/api/webrtc?app=%s&stream=%s&type=play%s", mediaInfo.getStreamIp(), mediaInfo.getHttpSSlPort(), app, stream, ObjectUtils.isEmpty(callId)?"":"&callId=" + callId));
}
streamInfoResult.setTracks(tracks);

232
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java Normal file
View File

@@ -0,0 +1,232 @@
package com.genersoft.iot.vmp.service.impl;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.sip.TimeoutEvent;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author lin
*/
@Service
public class PlatformServiceImpl implements IPlatformService {
private final static String REGISTER_KEY_PREFIX = "platform_register_";
private final static String KEEPALIVE_KEY_PREFIX = "platform_keepalive_";
private final static Logger logger = LoggerFactory.getLogger(PlatformServiceImpl.class);
@Autowired
private ParentPlatformMapper platformMapper;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private SIPCommanderFroPlatform commanderForPlatform;
@Autowired
private DynamicTask dynamicTask;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private SubscribeHolder subscribeHolder;
@Override
public ParentPlatform queryPlatformByServerGBId(String platformGbId) {
return platformMapper.getParentPlatByServerGBId(platformGbId);
}
@Override
public PageInfo<ParentPlatform> queryParentPlatformList(int page, int count) {
PageHelper.startPage(page, count);
List<ParentPlatform> all = platformMapper.getParentPlatformList();
return new PageInfo<>(all);
}
@Override
public boolean add(ParentPlatform parentPlatform) {
if (parentPlatform.getCatalogGroup() == 0) {
// 每次发送目录的数量默认为1
parentPlatform.setCatalogGroup(1);
}
if (parentPlatform.getAdministrativeDivision() == null) {
// 行政区划默认去编号的前6位
parentPlatform.setAdministrativeDivision(parentPlatform.getServerGBId().substring(0,6));
}
parentPlatform.setCatalogId(parentPlatform.getDeviceGBId());
int result = platformMapper.addParentPlatform(parentPlatform);
// 添加缓存
ParentPlatformCatch parentPlatformCatch = new ParentPlatformCatch();
parentPlatformCatch.setParentPlatform(parentPlatform);
parentPlatformCatch.setId(parentPlatform.getServerGBId());
parentPlatformCatch.setParentPlatform(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
if (parentPlatform.isEnable()) {
// 保存时启用就发送注册
// 注册成功时由程序直接调用了online方法
commanderForPlatform.register(parentPlatform, eventResult -> {
logger.info("[国标级联] {},添加向上级注册失败，请确定上级平台可用时重新保存", parentPlatform.getServerGBId());
}, null);
}
return result > 0;
}
@Override
public void online(ParentPlatform parentPlatform) {
logger.info("[国标级联]：{}, 平台上线/更新注册", parentPlatform.getServerGBId());
platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), true);
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
if (parentPlatformCatch != null) {
parentPlatformCatch.getParentPlatform().setStatus(true);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}else {
parentPlatformCatch = new ParentPlatformCatch();
parentPlatformCatch.setParentPlatform(parentPlatform);
parentPlatformCatch.setId(parentPlatform.getServerGBId());
parentPlatform.setStatus(true);
parentPlatformCatch.setParentPlatform(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(registerTaskKey)) {
dynamicTask.stop(registerTaskKey);
}
// 添加注册任务
dynamicTask.startDelay(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->commanderForPlatform.register(parentPlatform, eventResult -> offline(parentPlatform),null),
parentPlatform.getExpires()*1000);
final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId();
if (!dynamicTask.contains(keepaliveTaskKey)) {
// 添加心跳任务
dynamicTask.startCron(keepaliveTaskKey,
()-> commanderForPlatform.keepalive(parentPlatform, eventResult -> {
// 心跳失败
if (eventResult.type == SipSubscribe.EventResultType.timeout) {
// 心跳超时
ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
// 此时是第三次心跳超时， 平台离线
if (platformCatch.getKeepAliveReply() == 2) {
// 设置平台离线，并重新注册
offline(parentPlatform);
logger.info("[国标级联] {}，三次心跳超时后再次发起注册", parentPlatform.getServerGBId());
commanderForPlatform.register(parentPlatform, eventResult1 -> {
logger.info("[国标级联] {}，三次心跳超时后再次发起注册仍然失败，开始定时发起注册，间隔为1分钟", parentPlatform.getServerGBId());
// 添加注册任务
dynamicTask.startCron(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->logger.info("[国标级联] {},平台离线后持续发起注册，失败", parentPlatform.getServerGBId()),
60*1000);
}, null);
}
}else {
logger.warn("[国标级联]发送心跳收到错误，code： {}, msg: {}", eventResult.statusCode, eventResult.msg);
}
}, eventResult -> {
// 心跳成功
// 清空之前的心跳超时计数
ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
if (platformCatch.getKeepAliveReply() > 0) {
platformCatch.setKeepAliveReply(0);
redisCatchStorage.updatePlatformCatchInfo(platformCatch);
}
}),
parentPlatform.getExpires()*1000);
}
}
@Override
public void offline(ParentPlatform parentPlatform) {
logger.info("[平台离线]：{}", parentPlatform.getServerGBId());
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
parentPlatformCatch.setKeepAliveReply(0);
parentPlatformCatch.setRegisterAliveReply(0);
ParentPlatform parentPlatformInCatch = parentPlatformCatch.getParentPlatform();
parentPlatformInCatch.setStatus(false);
parentPlatformCatch.setParentPlatform(parentPlatformInCatch);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), false);
// 停止所有推流
logger.info("[平台离线] {}, 停止所有推流", parentPlatform.getServerGBId());
stopAllPush(parentPlatform.getServerGBId());
// 清除注册定时
logger.info("[平台离线] {}, 停止定时注册任务", parentPlatform.getServerGBId());
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(registerTaskKey)) {
dynamicTask.stop(registerTaskKey);
}
// 清除心跳定时
logger.info("[平台离线] {}, 停止定时发送心跳任务", parentPlatform.getServerGBId());
final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(keepaliveTaskKey)) {
// 添加心跳任务
dynamicTask.stop(keepaliveTaskKey);
}
// 停止目录订阅回复
logger.info("[平台离线] {}, 停止订阅回复", parentPlatform.getServerGBId());
subscribeHolder.removeAllSubscribe(parentPlatform.getServerGBId());
}
private void stopAllPush(String platformId) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(platformId);
if (sendRtpItems != null && sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(), null, null);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
Map<String, Object> param = new HashMap<>(3);
param.put("vhost", "__defaultVhost__");
param.put("app", sendRtpItem.getApp());
param.put("stream", sendRtpItem.getStreamId());
zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
}
}
}
@Override
public void login(ParentPlatform parentPlatform) {
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
commanderForPlatform.register(parentPlatform, eventResult1 -> {
logger.info("[国标级联] {}，开始定时发起注册，间隔为1分钟", parentPlatform.getServerGBId());
// 添加注册任务
dynamicTask.startCron(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->logger.info("[国标级联] {},平台离线后持续发起注册，失败", parentPlatform.getServerGBId()),
60*1000);
}, null);
}
}

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

@@ -37,7 +37,7 @@ import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
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.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -128,7 +128,7 @@ public class PlayServiceImpl implements IPlayService {
private DynamicTask dynamicTask;
@Autowired
private ZLMHttpHookSubscribe subscribe;
private ZlmHttpHookSubscribe subscribe;
@Qualifier("taskExecutor")
@@ -139,7 +139,7 @@ public class PlayServiceImpl implements IPlayService {
@Override
public PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId,
ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
Runnable timeoutCallback) {
if (mediaServerItem == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的zlm");
@@ -222,6 +222,7 @@ public class PlayServiceImpl implements IPlayService {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false);
logger.info(JSONObject.toJSONString(ssrcInfo));
play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response)->{
if (hookEvent != null) {
hookEvent.response(mediaServerItem, response);
@@ -257,8 +258,8 @@ public class PlayServiceImpl implements IPlayService {
@Override
public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
InviteTimeOutCallback timeoutCallback, String uuid) {
ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
InviteTimeOutCallback timeoutCallback, String uuid) {
String streamId = null;
if (mediaServerItem.isRtpEnable()) {
@@ -333,7 +334,7 @@ public class PlayServiceImpl implements IPlayService {
// 单端口模式streamId也有变化，需要重新设置监听
if (!mediaServerItem.isRtpEnable()) {
// 添加订阅
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtmp", mediaServerItem.getId());
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{
@@ -609,7 +610,7 @@ public class PlayServiceImpl implements IPlayService {
logger.warn("查询录像信息时发现节点已离线");
return null;
}
if (mediaServerItem.getRecordAssistPort() != 0) {
if (mediaServerItem.getRecordAssistPort() > 0) {
JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, streamInfo.getApp(), streamInfo.getStream(), null);
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
long duration = jsonObject.getLong("data");
@@ -802,7 +803,7 @@ public class PlayServiceImpl implements IPlayService {
// for (SendRtpItem sendRtpItem : sendRtpItems) {
// if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
// if (mediaListMap.get(sendRtpItem.getStreamId()) == null) {
// ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
// ParentPlatform platform = storager.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
// sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
// }
// }

32
src/main/java/com/genersoft/iot/vmp/service/impl/RedisAlarmMsgListener.java
View File

@@ -4,6 +4,8 @@ import com.alibaba.fastjson.JSON;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.service.IPlatformChannelService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import org.slf4j.Logger;
@@ -12,6 +14,9 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import java.util.List;
@Component
@@ -37,8 +42,6 @@ public class RedisAlarmMsgListener implements MessageListener {
return;
}
String gbId = alarmChannelMessage.getGbId();
Device device = storage.queryVideoDevice(gbId);
ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId);
DeviceAlarm deviceAlarm = new DeviceAlarm();
deviceAlarm.setCreateTime(DateUtil.getNow());
@@ -46,18 +49,29 @@ public class RedisAlarmMsgListener implements MessageListener {
deviceAlarm.setAlarmDescription(alarmChannelMessage.getAlarmDescription());
deviceAlarm.setAlarmMethod("" + alarmChannelMessage.getAlarmSn());
deviceAlarm.setAlarmPriority("1");
deviceAlarm.setAlarmTime(DateUtil.getNow());
deviceAlarm.setAlarmTime(DateUtil.getNowForISO8601());
deviceAlarm.setAlarmType("1");
deviceAlarm.setLongitude(0);
deviceAlarm.setLatitude(0);
if (device != null && platform == null) {
commander.sendAlarmMessage(device, deviceAlarm);
}else if (device == null && platform != null){
commanderForPlatform.sendAlarmMessage(platform, deviceAlarm);
if (ObjectUtils.isEmpty(gbId)) {
// 发送给所有的上级
List<ParentPlatform> parentPlatforms = storage.queryEnableParentPlatformList(true);
if (parentPlatforms.size() > 0) {
for (ParentPlatform parentPlatform : parentPlatforms) {
commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm);
}
}
}else {
logger.warn("无法确定" + gbId + "是平台还是设备");
Device device = storage.queryVideoDevice(gbId);
ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId);
if (device != null && platform == null) {
commander.sendAlarmMessage(device, deviceAlarm);
}else if (device == null && platform != null){
commanderForPlatform.sendAlarmMessage(platform, deviceAlarm);
}else {
logger.warn("无法确定" + gbId + "是平台还是设备");
}
}
}
}

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

@@ -5,12 +5,11 @@ import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
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.*;
@@ -24,9 +23,6 @@ import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@@ -86,7 +82,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
private ZLMMediaListManager mediaListManager;
@Autowired
private ZLMHttpHookSubscribe subscribe;
private ZlmHttpHookSubscribe subscribe;
public interface PlayMsgCallback{
@@ -271,7 +267,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
}, userSetting.getPlatformPlayTimeout());
// 添加订阅
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(content.getApp(), content.getStream(), true, "rtmp", mediaServerItem.getId());
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(content.getApp(), content.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
dynamicTask.stop(taskKey);

3
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
View File

@@ -295,7 +295,6 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
if (jsonObject == null) {
return false;
}
System.out.println(jsonObject);
if (jsonObject.getInteger("code") == 0) {
result = true;
streamProxy.setEnable(true);
@@ -421,7 +420,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
if(data != null && data.size() > 0) {
for (int i = 0; i < data.size(); i++) {
JSONObject streamJSONObj = data.getJSONObject(i);
if ("rtmp".equals(streamJSONObj.getString("schema"))) {
if ("rtsp".equals(streamJSONObj.getString("schema"))) {
StreamInfo streamInfo = new StreamInfo();
String app = streamJSONObj.getString("app");
String stream = streamJSONObj.getString("stream");