From 381b3d15caa30e99c75f58a366b3b75cbbf01450 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期二, 07 十二月 2021 11:40:17 +0800 Subject: [PATCH] 修复未结束的推流无法下载 --- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 136 +++++++++++++++++++++++++++++---------------- 1 files changed, 88 insertions(+), 48 deletions(-) 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 d6e77c7..caa68ef 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,16 +9,11 @@ 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.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 +60,9 @@ private IStreamProxyService streamProxyService; @Autowired + private IStreamPushService streamPushService; + + @Autowired private IMediaService mediaService; @Autowired @@ -83,6 +81,32 @@ private MediaConfig mediaConfig; /** + * 鏈嶅姟鍣ㄥ畾鏃朵笂鎶ユ椂闂达紝涓婃姤闂撮殧鍙厤缃紝榛樿10s涓婃姤涓�娆� + * + */ + @ResponseBody + @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8") + public ResponseEntity<String> onServerKeepalive(@RequestBody JSONObject json){ + + if (logger.isDebugEnabled()) { + 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); + 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<String>(ret.toString(),HttpStatus.OK); + } + + /** * 娴侀噺缁熻浜嬩欢锛屾挱鏀惧櫒鎴栨帹娴佸櫒鏂紑鏃跺苟涓旇�楃敤娴侀噺瓒呰繃鐗瑰畾闃堝�兼椂浼氳Е鍙戞浜嬩欢锛岄槇鍊奸�氳繃閰嶇疆鏂囦欢general.flowThreshold閰嶇疆锛涙浜嬩欢瀵瑰洖澶嶄笉鏁忔劅銆� * */ @@ -91,7 +115,7 @@ public ResponseEntity<String> 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(); @@ -109,7 +133,7 @@ public ResponseEntity<String> 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(); @@ -129,7 +153,7 @@ public ResponseEntity<String> 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); @@ -154,7 +178,7 @@ @PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8") public ResponseEntity<String> 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); @@ -164,12 +188,20 @@ subscribe.response(mediaInfo, json); } } + String app = json.getString("app"); + String stream = json.getString("stream"); + StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(stream); JSONObject ret = new JSONObject(); + // 褰曞儚鍥炴斁鏃朵笉杩涜褰曞儚涓嬭浇 + if (streamInfo != null) { + ret.put("enableMP4", false); + }else { + ret.put("enableMP4", userSetup.isRecordPushLive()); + } ret.put("code", 0); ret.put("msg", "success"); ret.put("enableHls", true); ret.put("enableMP4", userSetup.isRecordPushLive()); - ret.put("enableRtxp", true); return new ResponseEntity<String>(ret.toString(), HttpStatus.OK); } @@ -182,7 +214,7 @@ public ResponseEntity<String> 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(); @@ -200,7 +232,7 @@ public ResponseEntity<String> 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(); @@ -219,7 +251,7 @@ public ResponseEntity<String> 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(); @@ -238,7 +270,7 @@ public ResponseEntity<String> 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 濡傛灉鏄甫鏈塺tpstream鍒欏紑鍚寜闇�鎷夋祦 // String app = json.getString("app"); @@ -268,7 +300,7 @@ public ResponseEntity<String> 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); @@ -306,34 +338,39 @@ } }else { if (!"rtp".equals(app)){ - - boolean pushChange = false; - + String type = OriginType.values()[item.getOriginType()].getType(); MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); if (regist) { - if ((item.getOriginType() == 1 || item.getOriginType() == 2 || item.getOriginType() == 8)) { - pushChange = true; - zlmMediaListManager.addMedia(item); - StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks); - redisCatchStorage.addPushStream(mediaServerItem, app, streamId, streamInfo); + StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks); + 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.addPush(item); } }else { - int result = zlmMediaListManager.removeMedia( app, streamId); - redisCatchStorage.removePushStream(mediaServerItem, app, streamId); - if (result > 0) { - pushChange = true; + // 鍏煎娴佹敞閿�鏃剁被鍨嬮敊璇殑闂锛岀瓑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(pushChange) { - // 鍙戦�佹祦鍙樺寲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); } } } @@ -353,19 +390,18 @@ public ResponseEntity<String> 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"); String app = json.getString("app"); - - // TODO 濡傛灉鍦ㄧ粰涓婄骇鎺ㄦ祦锛屼篃涓嶅仠姝€�� + JSONObject ret = new JSONObject(); + ret.put("code", 0); if ("rtp".equals(app)){ - JSONObject ret = new JSONObject(); - ret.put("code", 0); ret.put("close", true); StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(streamId); if (streamInfoForPlayCatch != null) { + // 濡傛灉鍦ㄧ粰涓婄骇鎺ㄦ祦锛屼篃涓嶅仠姝€�� if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) { ret.put("close", false); } else { @@ -378,6 +414,12 @@ if (streamInfoForPlayBackCatch != null) { cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(), streamInfoForPlayBackCatch.getChannelId()); redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch); + }else { + StreamInfo streamInfoForDownload = redisCatchStorage.queryDownloadByStreamId(streamId); + // 杩涜褰曞儚涓嬭浇鏃舵棤浜鸿鐪嬩笉鏂祦 + if (streamInfoForDownload != null) { + ret.put("close", false); + } } } MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); @@ -386,8 +428,6 @@ } return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); }else { - JSONObject ret = new JSONObject(); - ret.put("code", 0); StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId); if (streamProxyItem != null && streamProxyItem.isEnable_remove_none_reader()) { ret.put("close", true); @@ -410,7 +450,7 @@ @PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8") public ResponseEntity<String> 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); @@ -457,7 +497,7 @@ public ResponseEntity<String> 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); -- Gitblit v1.8.0