diff --git a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index 03b003b9..386eb1ca 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java @@ -56,5 +56,5 @@ public class VideoManagerConstants { public static final String MEDIA_TRANSACTION_USED_PREFIX = "VMP_media_transaction_"; //************************** redis 消息********************************* - public static final String WVP_MSG_STREAM_PUSH_CHANGE_PREFIX = "WVP_MSG_STREAM_PUSH_CHANGE"; + public static final String WVP_MSG_STREAM_PUSH_CHANGE_PREFIX = "WVP_MSG_STREAM_CHANGE_"; } diff --git a/src/main/java/com/genersoft/iot/vmp/conf/ThreadPoolTaskConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/ThreadPoolTaskConfig.java index bde6b3fc..1bc85218 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/ThreadPoolTaskConfig.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/ThreadPoolTaskConfig.java @@ -38,7 +38,7 @@ public class ThreadPoolTaskConfig { /** * 线程池名前缀 */ - private static final String threadNamePrefix = "wvp-sip-"; + private static final String threadNamePrefix = "wvp-"; @Bean("taskExecutor") // bean的名称，默认为首字母小写的方法名 public ThreadPoolTaskExecutor taskExecutor() { diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java index f211fd62..26ce6287 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -9,17 +9,11 @@ import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.MediaConfig; import com.genersoft.iot.vmp.conf.UserSetup; import com.genersoft.iot.vmp.gb28181.bean.Device; -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.StreamProxyItem; -import com.genersoft.iot.vmp.service.IMediaServerService; -import com.genersoft.iot.vmp.service.IMediaService; -import com.genersoft.iot.vmp.service.IStreamProxyService; +import com.genersoft.iot.vmp.media.zlm.dto.*; +import com.genersoft.iot.vmp.service.*; import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorager; -import com.genersoft.iot.vmp.service.IPlayService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -65,6 +59,9 @@ public class ZLMHttpHookListener { @Autowired private IStreamProxyService streamProxyService; + @Autowired + private IStreamPushService streamPushService; + @Autowired private IMediaService mediaService; @@ -83,6 +80,32 @@ public class ZLMHttpHookListener { @Autowired private MediaConfig mediaConfig; + /** + * 服务器定时上报时间，上报间隔可配置，默认10s上报一次 + * + */ + @ResponseBody + @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8") + public ResponseEntity onServerKeepalive(@RequestBody JSONObject json){ + + if (logger.isDebugEnabled()) { + logger.debug("[ ZLM HOOK ]on_server_keepalive API调用，参数：" + json.toString()); + } + String mediaServerId = json.getString("mediaServerId"); + + List subscribes = this.subscribe.getSubscribes(ZLMHttpHookSubscribe.HookType.on_server_keepalive); + if (subscribes != null && subscribes.size() > 0) { + for (ZLMHttpHookSubscribe.Event subscribe : subscribes) { + subscribe.response(null, json); + } + } + + JSONObject ret = new JSONObject(); + ret.put("code", 0); + ret.put("msg", "success"); + return new ResponseEntity(ret.toString(),HttpStatus.OK); + } + /** * 流量统计事件，播放器或推流器断开时并且耗用流量超过特定阈值时会触发此事件，阈值通过配置文件general.flowThreshold配置；此事件对回复不敏感。 * @@ -92,7 +115,7 @@ public class ZLMHttpHookListener { public ResponseEntity onFlowReport(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_flow_report API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_flow_report API调用，参数：" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); JSONObject ret = new JSONObject(); @@ -110,7 +133,7 @@ public class ZLMHttpHookListener { public ResponseEntity onHttpAccess(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_http_access API 调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_http_access API 调用，参数：" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); JSONObject ret = new JSONObject(); @@ -130,7 +153,7 @@ public class ZLMHttpHookListener { public ResponseEntity onPlay(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_play API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_play API调用，参数：" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_play, json); @@ -155,7 +178,7 @@ public class ZLMHttpHookListener { @PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8") public ResponseEntity onPublish(@RequestBody JSONObject json) { - logger.debug("ZLM HOOK on_publish API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_publish API调用，参数：" + json.toString()); String mediaServerId = json.getString("mediaServerId"); ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, json); @@ -191,7 +214,7 @@ public class ZLMHttpHookListener { public ResponseEntity onRecordMp4(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_record_mp4 API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_record_mp4 API调用，参数：" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); JSONObject ret = new JSONObject(); @@ -209,7 +232,7 @@ public class ZLMHttpHookListener { public ResponseEntity onRtspRealm(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_rtsp_realm API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_rtsp_realm API调用，参数：" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); JSONObject ret = new JSONObject(); @@ -228,7 +251,7 @@ public class ZLMHttpHookListener { public ResponseEntity onRtspAuth(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_rtsp_auth API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_rtsp_auth API调用，参数：" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); JSONObject ret = new JSONObject(); @@ -247,7 +270,7 @@ public class ZLMHttpHookListener { public ResponseEntity onShellLogin(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_shell_login API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_shell_login API调用，参数：" + json.toString()); } // TODO 如果是带有rtpstream则开启按需拉流 // String app = json.getString("app"); @@ -277,7 +300,7 @@ public class ZLMHttpHookListener { public ResponseEntity onStreamChanged(@RequestBody MediaItem item){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_stream_changed API调用，参数：" + JSONObject.toJSONString(item)); + logger.debug("[ ZLM HOOK ]on_stream_changed API调用，参数：" + JSONObject.toJSONString(item)); } String mediaServerId = item.getMediaServerId(); JSONObject json = (JSONObject) JSON.toJSON(item); @@ -315,33 +338,39 @@ public class ZLMHttpHookListener { } }else { if (!"rtp".equals(app)){ - + String type = OriginType.values()[item.getOriginType()].getType(); MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); if (regist) { StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks); - redisCatchStorage.addStream(mediaServerItem, OriginType.values()[item.getOriginType()].getType(), app, streamId, streamInfo); + redisCatchStorage.addStream(mediaServerItem, type, app, streamId, streamInfo); if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal() || item.getOriginType() == OriginType.RTMP_PUSH.ordinal() || item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) { zlmMediaListManager.addMedia(item); } }else { + // 兼容流注销时类型错误的问题，等zlm更新后删除 + StreamPushItem streamPushItem = streamPushService.getPush(app, streamId); + if (streamPushItem != null) { + type = "PUSH"; + }else { + StreamProxyItem streamProxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(app, streamId); + if (streamProxyByAppAndStream != null) { + type = "PULL"; + } + } zlmMediaListManager.removeMedia( app, streamId); redisCatchStorage.removeStream(mediaServerItem, OriginType.values()[item.getOriginType()].getType(), app, streamId); - - } - if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal() - || item.getOriginType() == OriginType.RTMP_PUSH.ordinal() - || item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) { - // 发送流变化redis消息 - JSONObject jsonObject = new JSONObject(); - jsonObject.put("serverId", userSetup.getServerId()); - jsonObject.put("app", app); - jsonObject.put("stream", streamId); - jsonObject.put("register", regist); - jsonObject.put("mediaServerId", mediaServerId); - redisCatchStorage.sendStreamChangeMsg(jsonObject); } + + // 发送流变化redis消息 + JSONObject jsonObject = new JSONObject(); + jsonObject.put("serverId", userSetup.getServerId()); + jsonObject.put("app", app); + jsonObject.put("stream", streamId); + jsonObject.put("register", regist); + jsonObject.put("mediaServerId", mediaServerId); + redisCatchStorage.sendStreamChangeMsg(type, jsonObject); } } } @@ -361,7 +390,7 @@ public class ZLMHttpHookListener { public ResponseEntity onStreamNoneReader(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_stream_none_reader API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_stream_none_reader API调用，参数：" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); String streamId = json.getString("stream"); @@ -421,7 +450,7 @@ public class ZLMHttpHookListener { @PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8") public ResponseEntity onStreamNotFound(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_stream_not_found API调用，参数：" + json.toString()); + logger.debug("[ ZLM HOOK ]on_stream_not_found API调用，参数：" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); @@ -468,7 +497,7 @@ public class ZLMHttpHookListener { public ResponseEntity onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){ if (logger.isDebugEnabled()) { - logger.debug("ZLM HOOK on_server_started API调用，参数：" + jsonObject.toString()); + logger.debug("[ ZLM HOOK ]on_server_started API调用，参数：" + jsonObject.toString()); } String remoteAddr = request.getRemoteAddr(); jsonObject.put("ip", remoteAddr); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookSubscribe.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookSubscribe.java index c47c394e..c8cca53e 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookSubscribe.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookSubscribe.java @@ -27,7 +27,8 @@ public class ZLMHttpHookSubscribe { on_stream_changed, on_stream_none_reader, on_stream_not_found, - on_server_started + on_server_started, + on_server_keepalive } public interface Event{ diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java index b0960df8..f53c1677 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java @@ -3,7 +3,6 @@ package com.genersoft.iot.vmp.media.zlm; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; -import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.MediaConfig; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IMediaServerService; @@ -11,9 +10,11 @@ import com.genersoft.iot.vmp.service.IStreamProxyService; 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 java.util.*; @@ -41,8 +42,13 @@ public class ZLMRunner implements CommandLineRunner { @Autowired private MediaConfig mediaConfig; + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + @Override public void run(String... strings) throws Exception { + mediaServerService.clearMediaServerForOnline(); if (mediaServerService.getDefaultMediaServer() == null) { mediaServerService.addToDatabase(mediaConfig.getMediaSerItem()); } @@ -59,6 +65,9 @@ public class ZLMRunner implements CommandLineRunner { } }); + // TODO 订阅 zlm保活事件, 当zlm离线时做业务的处理 + + // 获取zlm信息 logger.info("等待默认zlm接入..."); @@ -70,7 +79,9 @@ public class ZLMRunner implements CommandLineRunner { for (MediaServerItem mediaServerItem : all) { if (startGetMedia == null) startGetMedia = new HashMap<>(); startGetMedia.put(mediaServerItem.getId(), true); - connectZlmServer(mediaServerItem); + taskExecutor.execute(()->{ + connectZlmServer(mediaServerItem); + }); } Timer timer = new Timer(); // 2分钟后未连接到则不再去主动连接, TODO 并对重启前使用此在zlm的通道发送bye diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/OriginType.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/OriginType.java index cb41c9a3..630e8257 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/OriginType.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/OriginType.java @@ -1,15 +1,16 @@ package com.genersoft.iot.vmp.media.zlm.dto; public enum OriginType { + // 不可调整顺序 UNKNOWN("UNKNOWN"), RTMP_PUSH("PUSH"), RTSP_PUSH("PUSH"), RTP_PUSH("RTP"), - RTC_PUSH("PUSH"), PULL("PULL"), FFMPEG_PULL("PULL"), MP4_VOD("MP4_VOD"), - DEVICE_CHN("DEVICE_CHN"); + DEVICE_CHN("DEVICE_CHN"), + RTC_PUSH("PUSH"); private final String type; OriginType(String type) { diff --git a/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java b/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java index 94e7d691..2b45ce7d 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java +++ b/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java @@ -35,4 +35,6 @@ public interface IStreamPushService { PageInfo getPushList(Integer page, Integer count); StreamPushItem transform(MediaItem item); + + StreamPushItem getPush(String app, String streamId); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java index 228e622e..3e7d5c4b 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java @@ -310,7 +310,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR */ @Override public void handLeZLMServerConfig(ZLMServerConfig zlmServerConfig) { - logger.info("[ {} ]-[ {}:{} ]已连接", + logger.info("[ ZLM：{} ]-[ {}:{} ]已连接", zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort()); MediaServerItem serverItem = mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId()); @@ -469,7 +469,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR */ @Override public void setZLMConfig(MediaServerItem mediaServerItem) { - logger.info("[ {} ]-[ {}:{} ]设置zlm", + logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); String protocol = sslEnabled ? "https" : "http"; String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort); @@ -494,16 +494,17 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrex)); 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)); param.put("hook.timeoutSec","20"); param.put("general.streamNoneReaderDelayMS","-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() ); JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param); if (responseJSON != null && responseJSON.getInteger("code") == 0) { - logger.info("[ {} ]-[ {}:{} ]设置zlm成功", + logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm成功", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); }else { - logger.info("[ {} ]-[ {}:{} ]设置zlm失败", + logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm失败", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); } } 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 28207212..634f670d 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 @@ -109,4 +109,11 @@ public class StreamPushServiceImpl implements IStreamPushService { } return del > 0; } + + + @Override + public StreamPushItem getPush(String app, String streamId) { + + return streamPushMapper.selectOne(app, streamId); + } } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index 543f9324..172213d2 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -127,7 +127,7 @@ public interface IRedisCatchStorage { * 发送推流生成与推流消失消息 * @param jsonObject 消息内容 */ - void sendStreamChangeMsg(JSONObject jsonObject); + void sendStreamChangeMsg(String type, JSONObject jsonObject); /** * 添加流信息到redis diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java index dd581185..835f942d 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java @@ -9,6 +9,8 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper; import com.genersoft.iot.vmp.utils.redis.RedisUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @@ -19,6 +21,8 @@ import java.util.*; @Component public class RedisCatchStorageImpl implements IRedisCatchStorage { + private Logger logger = LoggerFactory.getLogger(RedisCatchStorageImpl.class); + @Autowired private RedisUtil redis; @@ -311,8 +315,9 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { } @Override - public void sendStreamChangeMsg(JSONObject jsonObject) { - String key = VideoManagerConstants.WVP_MSG_STREAM_PUSH_CHANGE_PREFIX; + public void sendStreamChangeMsg(String type, JSONObject jsonObject) { + String key = VideoManagerConstants.WVP_MSG_STREAM_PUSH_CHANGE_PREFIX + type; + logger.debug("[redis 流变化事件] {}: {}", key, jsonObject.toString()); redis.convertAndSend(key, jsonObject); }