From 22efd0f56c20e1aedb18fa7b9bcdc48007fcc954 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期一, 20 十二月 2021 09:51:51 +0800 Subject: [PATCH] 添加心跳到期时验证zlm是否真的离线。 --- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 146 +++++++++++++++++++++++++++++++----------------- 1 files changed, 95 insertions(+), 51 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..b6af955 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,22 +178,34 @@ @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()); + JSONObject ret = new JSONObject(); + ret.put("code", 0); + ret.put("msg", "success"); + ret.put("enableHls", true); + ret.put("enableMP4", userSetup.isRecordPushLive()); String mediaServerId = json.getString("mediaServerId"); ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, json); if (subscribe != null) { MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); if (mediaInfo != null) { subscribe.response(mediaInfo, json); + }else { + ret.put("code", 1); + ret.put("msg", "zlm not register"); } } - JSONObject ret = new JSONObject(); - ret.put("code", 0); - ret.put("msg", "success"); - ret.put("enableHls", true); - ret.put("enableMP4", userSetup.isRecordPushLive()); - ret.put("enableRtxp", true); + String app = json.getString("app"); + String stream = json.getString("stream"); + StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(stream); + + // 褰曞儚鍥炴斁鏃朵笉杩涜褰曞儚涓嬭浇 + if (streamInfo != null) { + ret.put("enableMP4", false); + }else { + ret.put("enableMP4", userSetup.isRecordPushLive()); + } + return new ResponseEntity<String>(ret.toString(), HttpStatus.OK); } @@ -182,7 +218,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 +236,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 +255,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 +274,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 +304,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,25 +342,31 @@ } }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); + if (mediaServerItem != null){ + if (regist) { StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks); - redisCatchStorage.addPushStream(mediaServerItem, 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.addPush(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.getId(), type, app, streamId); } - }else { - int result = zlmMediaListManager.removeMedia( app, streamId); - redisCatchStorage.removePushStream(mediaServerItem, app, streamId); - if (result > 0) { - pushChange = true; - } - } - if(pushChange) { // 鍙戦�佹祦鍙樺寲redis娑堟伅 JSONObject jsonObject = new JSONObject(); jsonObject.put("serverId", userSetup.getServerId()); @@ -332,7 +374,7 @@ jsonObject.put("stream", streamId); jsonObject.put("register", regist); jsonObject.put("mediaServerId", mediaServerId); - redisCatchStorage.sendStreamChangeMsg(jsonObject); + redisCatchStorage.sendStreamChangeMsg(type, jsonObject); } } } @@ -353,19 +395,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 +419,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 +433,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); @@ -399,7 +444,6 @@ } return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); } - } /** @@ -410,7 +454,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); @@ -422,7 +466,7 @@ if (s.length == 2) { String deviceId = s[0]; String channelId = s[1]; - Device device = storager.queryVideoDevice(deviceId); + Device device = redisCatchStorage.getDevice(deviceId); if (device != null) { UUID uuid = UUID.randomUUID(); SSRCInfo ssrcInfo; @@ -457,7 +501,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