From bd5b9ca1f777e0991f37f6b0865fce873ed87072 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期四, 08 九月 2022 20:20:43 +0800 Subject: [PATCH] 获取实际内容长度,不使用header中的长度信息 --- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 168 ++++++++++++++++++++++++++++++++++---------------------- 1 files changed, 102 insertions(+), 66 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 7afe0fd..b789974 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 @@ -19,12 +19,8 @@ import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.http.HttpStatus; -import org.springframework.http.ResponseEntity; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.ObjectUtils; -import org.springframework.util.StringUtils; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; @@ -81,7 +77,7 @@ private ZLMMediaListManager zlmMediaListManager; @Autowired - private ZLMHttpHookSubscribe subscribe; + private ZlmHttpHookSubscribe subscribe; @Autowired private UserSetting userSetting; @@ -105,13 +101,13 @@ */ @ResponseBody @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onServerKeepalive(@RequestBody JSONObject json){ + public JSONObject onServerKeepalive(@RequestBody JSONObject json){ logger.info("[ ZLM HOOK ] on_server_keepalive API璋冪敤锛屽弬鏁帮細" + json.toString()); String mediaServerId = json.getString("mediaServerId"); - List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(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) { + for (ZlmHttpHookSubscribe.Event subscribe : subscribes) { subscribe.response(null, json); } } @@ -120,7 +116,8 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + + return ret; } /** @@ -129,16 +126,15 @@ */ @ResponseBody @PostMapping(value = "/on_flow_report", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onFlowReport(@RequestBody JSONObject json){ + public JSONObject onFlowReport(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_flow_report API璋冪敤锛屽弬鏁帮細" + json.toString()); } - String mediaServerId = json.getString("mediaServerId"); JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -147,7 +143,7 @@ */ @ResponseBody @PostMapping(value = "/on_http_access", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onHttpAccess(@RequestBody JSONObject json){ + public JSONObject onHttpAccess(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_http_access API 璋冪敤锛屽弬鏁帮細" + json.toString()); @@ -158,7 +154,7 @@ ret.put("err", ""); ret.put("path", ""); ret.put("second", 600); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -167,7 +163,7 @@ */ @ResponseBody @PostMapping(value = "/on_play", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onPlay(@RequestBody OnPlayHookParam param){ + public JSONObject onPlay(@RequestBody OnPlayHookParam param){ JSONObject json = (JSONObject)JSON.toJSON(param); @@ -175,7 +171,7 @@ logger.debug("[ ZLM HOOK ]on_play API璋冪敤锛屽弬鏁帮細" + JSON.toJSONString(param)); } String mediaServerId = param.getMediaServerId(); - ZLMHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(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) { @@ -186,17 +182,16 @@ if (!"rtp".equals(param.getApp())) { Map<String, String> paramMap = urlParamToMap(param.getParams()); StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream()); - if (streamAuthorityInfo == null - || (streamAuthorityInfo.getCallId() != null && !streamAuthorityInfo.getCallId().equals(paramMap.get("callId")))) { + if (streamAuthorityInfo != null && streamAuthorityInfo.getCallId() != null && !streamAuthorityInfo.getCallId().equals(paramMap.get("callId"))) { ret.put("code", 401); ret.put("msg", "Unauthorized"); - return new ResponseEntity<>(ret.toString(),HttpStatus.OK); + return ret; } } ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -205,7 +200,7 @@ */ @ResponseBody @PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onPublish(@RequestBody OnPublishHookParam param) { + public JSONObject onPublish(@RequestBody OnPublishHookParam param) { JSONObject json = (JSONObject) JSON.toJSON(param); @@ -219,7 +214,7 @@ logger.info("鎺ㄦ祦閴存潈澶辫触锛� 缂哄皯涓嶈鍙傛暟锛歴ign=md5(user琛ㄧ殑pushKey)"); ret.put("code", 401); ret.put("msg", "Unauthorized"); - return new ResponseEntity<>(ret.toString(), HttpStatus.OK); + return ret; } Map<String, String> paramMap = urlParamToMap(param.getParams()); String sign = paramMap.get("sign"); @@ -227,7 +222,7 @@ logger.info("鎺ㄦ祦閴存潈澶辫触锛� 缂哄皯涓嶈鍙傛暟锛歴ign=md5(user琛ㄧ殑pushKey)"); ret.put("code", 401); ret.put("msg", "Unauthorized"); - return new ResponseEntity<>(ret.toString(), HttpStatus.OK); + return ret; } // 鎺ㄦ祦鑷畾涔夋挱鏀鹃壌鏉冪爜 String callId = paramMap.get("callId"); @@ -237,7 +232,7 @@ logger.info("鎺ㄦ祦閴存潈澶辫触锛� sign 鏃犳潈闄�: callId={}. sign={}", callId, sign); ret.put("code", 401); ret.put("msg", "Unauthorized"); - return new ResponseEntity<>(ret.toString(), HttpStatus.OK); + return ret; } StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param); streamAuthorityInfo.setCallId(callId); @@ -245,12 +240,11 @@ // 閴存潈閫氳繃 redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo); // 閫氱煡assist鏂扮殑callId - taskExecutor.execute(()->{ - if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) { + if (mediaInfo != null && mediaInfo.getRecordAssistPort() > 0) { + taskExecutor.execute(()->{ assistRESTfulUtils.addStreamCallInfo(mediaInfo, param.getApp(), param.getStream(), callId, null); - } - }); - + }); + } }else { zlmMediaListManager.sendStreamEvent(param.getApp(),param.getStream(), param.getMediaServerId()); } @@ -263,7 +257,7 @@ } - ZLMHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json); + ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json); if (subscribe != null) { if (mediaInfo != null) { subscribe.response(mediaInfo, json); @@ -294,10 +288,7 @@ } } - - - - return new ResponseEntity<String>(ret.toString(), HttpStatus.OK); + return ret; } @@ -308,7 +299,7 @@ */ @ResponseBody @PostMapping(value = "/on_record_mp4", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onRecordMp4(@RequestBody JSONObject json){ + public JSONObject onRecordMp4(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_record_mp4 API璋冪敤锛屽弬鏁帮細" + json.toString()); @@ -317,7 +308,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** * 褰曞埗hls瀹屾垚鍚庨�氱煡浜嬩欢锛涙浜嬩欢瀵瑰洖澶嶄笉鏁忔劅銆� @@ -325,7 +316,7 @@ */ @ResponseBody @PostMapping(value = "/on_record_ts", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onRecordTs(@RequestBody JSONObject json){ + public JSONObject onRecordTs(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_record_ts API璋冪敤锛屽弬鏁帮細" + json.toString()); @@ -334,7 +325,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -343,7 +334,7 @@ */ @ResponseBody @PostMapping(value = "/on_rtsp_realm", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onRtspRealm(@RequestBody JSONObject json){ + public JSONObject onRtspRealm(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_rtsp_realm API璋冪敤锛屽弬鏁帮細" + json.toString()); @@ -352,7 +343,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("realm", ""); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } @@ -362,7 +353,7 @@ */ @ResponseBody @PostMapping(value = "/on_rtsp_auth", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onRtspAuth(@RequestBody JSONObject json){ + public JSONObject onRtspAuth(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_rtsp_auth API璋冪敤锛屽弬鏁帮細" + json.toString()); @@ -372,7 +363,7 @@ ret.put("code", 0); ret.put("encrypted", false); ret.put("passwd", "test"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -381,13 +372,13 @@ */ @ResponseBody @PostMapping(value = "/on_shell_login", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onShellLogin(@RequestBody JSONObject json){ + public JSONObject onShellLogin(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_shell_login API璋冪敤锛屽弬鏁帮細" + json.toString()); } String mediaServerId = json.getString("mediaServerId"); - ZLMHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(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) { @@ -399,7 +390,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -408,12 +399,12 @@ */ @ResponseBody @PostMapping(value = "/on_stream_changed", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onStreamChanged(@RequestBody MediaItem item){ + public JSONObject onStreamChanged(@RequestBody MediaItem item){ 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.sendNotify(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) { @@ -478,8 +469,12 @@ if (mediaServerItem != null){ if (regist) { StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream); + String callId = null; + if (streamAuthorityInfo != null) { + callId = streamAuthorityInfo.getCallId(); + } StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem, - app, stream, tracks, streamAuthorityInfo.getCallId()); + app, stream, tracks, callId); item.setStreamInfo(streamInfoByAppAndStream); redisCatchStorage.addStream(mediaServerItem, type, app, stream, item); if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal() @@ -519,7 +514,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -528,7 +523,7 @@ */ @ResponseBody @PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onStreamNoneReader(@RequestBody JSONObject json){ + public JSONObject onStreamNoneReader(@RequestBody JSONObject json){ logger.info("[ ZLM HOOK ]on_stream_none_reader API璋冪敤锛屽弬鏁帮細" + json.toString()); String mediaServerId = json.getString("mediaServerId"); @@ -547,6 +542,8 @@ for (SendRtpItem sendRtpItem : sendRtpItems) { ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId()); + redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(), + sendRtpItem.getCallId(), sendRtpItem.getStreamId()); } } } @@ -573,18 +570,24 @@ if (mediaServerItem != null && mediaServerItem.getStreamNoneReaderDelayMS() == -1) { ret.put("close", false); } - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; }else { StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId); - if (streamProxyItem != null && streamProxyItem.isEnable_remove_none_reader()) { - ret.put("close", true); - streamProxyService.del(app, streamId); - String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url(); - logger.info("[{}/{}]<-[{}] 鎷夋祦浠g悊鏃犱汉瑙傜湅宸茬粡绉婚櫎", app, streamId, url); - }else { - ret.put("close", false); + if (streamProxyItem != null ) { + if (streamProxyItem.isEnable_remove_none_reader()) { + // 鏃犱汉瑙傜湅鑷姩绉婚櫎 + ret.put("close", true); + streamProxyService.del(app, streamId); + String url = streamProxyItem.getUrl() != null?streamProxyItem.getUrl():streamProxyItem.getSrc_url(); + logger.info("[{}/{}]<-[{}] 鎷夋祦浠g悊鏃犱汉瑙傜湅宸茬粡绉婚櫎", app, streamId, url); + }else if (streamProxyItem.isEnable_disable_none_reader()) { + // 鏃犱汉瑙傜湅鍋滅敤 + ret.put("close", true); + }else { + ret.put("close", false); + } } - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } } @@ -594,7 +597,7 @@ */ @ResponseBody @PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onStreamNotFound(@RequestBody JSONObject json){ + public JSONObject onStreamNotFound(@RequestBody JSONObject json){ if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_stream_not_found API璋冪敤锛屽弬鏁帮細" + json.toString()); } @@ -619,7 +622,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -628,16 +631,16 @@ */ @ResponseBody @PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8") - public ResponseEntity<String> onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){ - + public JSONObject onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){ + if (logger.isDebugEnabled()) { logger.debug("[ ZLM HOOK ]on_server_started API璋冪敤锛屽弬鏁帮細" + jsonObject.toString()); } String remoteAddr = request.getRemoteAddr(); jsonObject.put("ip", remoteAddr); - List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(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) { + for (ZlmHttpHookSubscribe.Event subscribe : subscribes) { subscribe.response(null, jsonObject); } } @@ -649,7 +652,40 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<>(ret.toString(),HttpStatus.OK); + return ret; + } + + /** + * 鍙戦�乺tp(startSendRtp)琚姩鍏抽棴鏃跺洖璋� + */ + @ResponseBody + @PostMapping(value = "/on_send_rtp_stopped", produces = "application/json;charset=UTF-8") + public JSONObject onSendRtpStopped(HttpServletRequest request, @RequestBody JSONObject jsonObject){ + + logger.info("[ ZLM HOOK ]on_send_rtp_stopped API璋冪敤锛屽弬鏁帮細" + jsonObject); + + JSONObject ret = new JSONObject(); + ret.put("code", 0); + ret.put("msg", "success"); + + // 鏌ユ壘瀵瑰簲鐨勪笂绾ф帹娴侊紝鍙戦�佸仠姝� + String app = jsonObject.getString("app"); + if (!"rtp".equals(app)) { + return ret; + } + String stream = jsonObject.getString("stream"); + List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream); + if (sendRtpItems.size() > 0) { + for (SendRtpItem sendRtpItem : sendRtpItems) { + ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); + commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId()); + redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(), + sendRtpItem.getCallId(), sendRtpItem.getStreamId()); + } + } + + + return ret; } private Map<String, String> urlParamToMap(String params) { -- Gitblit v1.8.0