From 92022f8a8b7bccc865c47bd8cfafaffb91c2799c Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期四, 13 十月 2022 22:18:06 +0800 Subject: [PATCH] Merge pull request #640 from duzeng/wvp-28181-2.0 --- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 231 ++++++++++++++++++++++++++++++++++++++++----------------- 1 files changed, 161 insertions(+), 70 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 ce46393..e1689c9 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 @@ -1,5 +1,6 @@ package com.genersoft.iot.vmp.media.zlm; +import java.text.ParseException; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -7,6 +8,7 @@ import com.alibaba.fastjson.JSON; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.UserSetting; +import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; @@ -19,8 +21,6 @@ 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.util.ObjectUtils; import org.springframework.web.bind.annotation.PostMapping; @@ -33,6 +33,8 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import javax.servlet.http.HttpServletRequest; +import javax.sip.InvalidArgumentException; +import javax.sip.SipException; /** * @description:閽堝 ZLMediaServer鐨刪ook浜嬩欢鐩戝惉 @@ -59,6 +61,9 @@ @Autowired private IRedisCatchStorage redisCatchStorage; + + @Autowired + private IDeviceService deviceService; @Autowired private IMediaServerService mediaServerService; @@ -103,9 +108,9 @@ */ @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()); + 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); if (subscribes != null && subscribes.size() > 0) { @@ -118,7 +123,8 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + + return ret; } /** @@ -127,16 +133,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; } /** @@ -145,7 +150,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()); @@ -156,7 +161,7 @@ ret.put("err", ""); ret.put("path", ""); ret.put("second", 600); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -165,7 +170,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); @@ -187,13 +192,13 @@ 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; } /** @@ -202,7 +207,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); @@ -216,7 +221,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"); @@ -224,7 +229,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"); @@ -234,7 +239,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); @@ -242,12 +247,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()); } @@ -291,10 +295,7 @@ } } - - - - return new ResponseEntity<String>(ret.toString(), HttpStatus.OK); + return ret; } @@ -305,7 +306,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()); @@ -314,7 +315,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** * 褰曞埗hls瀹屾垚鍚庨�氱煡浜嬩欢锛涙浜嬩欢瀵瑰洖澶嶄笉鏁忔劅銆� @@ -322,7 +323,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()); @@ -331,7 +332,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -340,7 +341,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()); @@ -349,7 +350,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("realm", ""); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } @@ -359,7 +360,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()); @@ -369,7 +370,7 @@ ret.put("code", 0); ret.put("encrypted", false); ret.put("passwd", "test"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -378,7 +379,7 @@ */ @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()); @@ -396,7 +397,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -405,7 +406,7 @@ */ @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(); @@ -423,10 +424,11 @@ String schema = item.getSchema(); List<MediaItem.MediaTrack> tracks = item.getTracks(); boolean regist = item.isRegist(); - if (item.getOriginType() == OriginType.RTMP_PUSH.ordinal() - || item.getOriginType() == OriginType.RTSP_PUSH.ordinal() - || item.getOriginType() == OriginType.RTC_PUSH.ordinal()) { - if (regist) { + if (regist) { + if (item.getOriginType() == OriginType.RTMP_PUSH.ordinal() + || item.getOriginType() == OriginType.RTSP_PUSH.ordinal() + || item.getOriginType() == OriginType.RTC_PUSH.ordinal()) { + StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream); if (streamAuthorityInfo == null) { streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(item); @@ -435,9 +437,9 @@ streamAuthorityInfo.setOriginTypeStr(item.getOriginTypeStr()); } redisCatchStorage.updateStreamAuthorityInfo(app, stream, streamAuthorityInfo); - }else { - redisCatchStorage.removeStreamAuthorityInfo(app, stream); } + }else { + redisCatchStorage.removeStreamAuthorityInfo(app, stream); } if ("rtsp".equals(schema)){ @@ -457,15 +459,12 @@ if (streamInfo!=null){ redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); - // 濡傛灉姝e湪缁欎笂绾ф帹閫侊紝鍒欏彂閫乥ye - }else{ streamInfo = redisCatchStorage.queryPlayback(null, null, stream, null); if (streamInfo != null) { redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream(), null); } - // 濡傛灉姝e湪缁欎笂绾ф帹閫侊紝鍒欏彂閫乥ye } }else { if (!"rtp".equals(app)){ @@ -515,12 +514,34 @@ } } } + if (!regist) { + List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream); + if (sendRtpItems.size() > 0) { + for (SendRtpItem sendRtpItem : sendRtpItems) { + if (sendRtpItem.getApp().equals(app)) { + String platformId = sendRtpItem.getPlatformId(); + ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId); + Device device = deviceService.queryDevice(platformId); + + try { + if (platform != null) { + commanderFroPlatform.streamByeCmd(platform, sendRtpItem); + }else { + cmder.streamByeCmd(device, sendRtpItem.getChannelId(), stream, sendRtpItem.getCallId()); + } + } catch (SipException | InvalidArgumentException | ParseException | SsrcTransactionNotFoundException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage()); + } + } + } + } + } } JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<String>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -529,7 +550,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,21 +568,48 @@ if (sendRtpItems.size() > 0) { for (SendRtpItem sendRtpItem : sendRtpItems) { ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); - commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId()); + try { + commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId()); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage()); + } + redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(), + sendRtpItem.getCallId(), sendRtpItem.getStreamId()); } } } - cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId(), - streamInfoForPlayCatch.getStream(), null); + Device device = deviceService.queryDevice(streamInfoForPlayCatch.getDeviceID()); + if (device != null) { + try { + cmder.streamByeCmd(device, streamInfoForPlayCatch.getChannelId(), + streamInfoForPlayCatch.getStream(), null); + } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { + logger.error("[鏃犱汉瑙傜湅]鐐规挱锛� 鍙戦�丅YE澶辫触 {}", e.getMessage()); + } + } + redisCatchStorage.stopPlay(streamInfoForPlayCatch); storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId()); }else{ StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, streamId, null); - if (streamInfoForPlayBackCatch != null) { - cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(), - streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null); - redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(), - streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null); + if (streamInfoForPlayBackCatch != null ) { + if (streamInfoForPlayBackCatch.isPause()) { + ret.put("close", false); + }else { + Device device = deviceService.queryDevice(streamInfoForPlayBackCatch.getDeviceID()); + if (device != null) { + try { + cmder.streamByeCmd(device,streamInfoForPlayBackCatch.getChannelId(), + streamInfoForPlayBackCatch.getStream(), null); + } catch (InvalidArgumentException | ParseException | SipException | + SsrcTransactionNotFoundException e) { + logger.error("[鏃犱汉瑙傜湅]鍥炴斁锛� 鍙戦�丅YE澶辫触 {}", e.getMessage()); + } + } + redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(), + streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null); + } + }else { StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, streamId, null); // 杩涜褰曞儚涓嬭浇鏃舵棤浜鸿鐪嬩笉鏂祦 @@ -574,18 +622,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; } } @@ -595,7 +649,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()); } @@ -620,7 +674,7 @@ JSONObject ret = new JSONObject(); ret.put("code", 0); ret.put("msg", "success"); - return new ResponseEntity<>(ret.toString(),HttpStatus.OK); + return ret; } /** @@ -629,8 +683,8 @@ */ @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()); } @@ -650,7 +704,44 @@ 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()); + try { + commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId()); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage()); + } + redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(), + sendRtpItem.getCallId(), sendRtpItem.getStreamId()); + } + } + + + return ret; } private Map<String, String> urlParamToMap(String params) { -- Gitblit v1.8.0