From d739bfa5976e36ced26f906ab16f83c20c8cb27c Mon Sep 17 00:00:00 2001 From: xiaoxie <hotcoffie@163.com> Date: 星期一, 23 五月 2022 09:44:21 +0800 Subject: [PATCH] 处理冲突 --- src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 204 +++++++++++++++++++++++++++++--------------------- 1 files changed, 118 insertions(+), 86 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java index 00daf10..2521a06 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java @@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.common.StreamInfo; +import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; @@ -12,23 +13,22 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform; -import com.genersoft.iot.vmp.gb28181.utils.DateUtil; +import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IMediaServerService; +import com.genersoft.iot.vmp.service.IMediaService; +import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback; import com.genersoft.iot.vmp.service.bean.PlayBackCallback; import com.genersoft.iot.vmp.service.bean.PlayBackResult; import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.utils.redis.RedisUtil; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult; -import com.genersoft.iot.vmp.service.IMediaService; -import com.genersoft.iot.vmp.service.IPlayService; import gov.nist.javax.sip.stack.SIPDialog; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,6 +42,7 @@ import javax.sip.ResponseEvent; import java.io.FileNotFoundException; import java.math.BigDecimal; +import java.math.RoundingMode; import java.util.*; @SuppressWarnings(value = {"rawtypes", "unchecked"}) @@ -61,9 +62,6 @@ @Autowired private IRedisCatchStorage redisCatchStorage; - - @Autowired - private RedisUtil redis; @Autowired private DeferredResultHolder resultHolder; @@ -86,6 +84,12 @@ @Autowired private UserSetting userSetting; + @Autowired + private DynamicTask dynamicTask; + + @Autowired + private ZLMHttpHookSubscribe subscribe; + @@ -100,7 +104,7 @@ String uuid = UUID.randomUUID().toString(); msg.setId(uuid); playResult.setUuid(uuid); - DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(userSetting.getPlayTimeout()); + DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); playResult.setResult(result); // 褰曞儚鏌ヨ浠hannelId浣滀负deviceId鏌ヨ resultHolder.put(key, uuid, result); @@ -189,7 +193,7 @@ if (mediaServerItem.isRtpEnable()) { streamId = String.format("%s_%s", device.getDeviceId(), channelId); } - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck()); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response)->{ if (hookEvent != null) { hookEvent.response(mediaServerItem, response); @@ -233,34 +237,32 @@ streamId = String.format("%s_%s", device.getDeviceId(), channelId); } if (ssrcInfo == null) { - ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck()); + ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); } // 瓒呮椂澶勭悊 - Timer timer = new Timer(); + String timeOutTaskKey = UUID.randomUUID().toString(); SSRCInfo finalSsrcInfo = ssrcInfo; - timer.schedule(new TimerTask() { - @Override - public void run() { - logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", device.getDeviceId(), channelId)); + dynamicTask.startDelay( timeOutTaskKey,()->{ + logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", device.getDeviceId(), channelId)); - SIPDialog dialog = streamSession.getDialogByStream(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); - if (dialog != null) { - timeoutCallback.run(1, "鏀舵祦瓒呮椂"); - // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� - cmder.streamByeCmd(device.getDeviceId(), channelId, finalSsrcInfo.getStream(), null); - }else { - timeoutCallback.run(0, "鐐规挱瓒呮椂"); - mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); - mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); - streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); - } + SIPDialog dialog = streamSession.getDialogByStream(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); + if (dialog != null) { + timeoutCallback.run(1, "鏀舵祦瓒呮椂"); + // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� + cmder.streamByeCmd(device.getDeviceId(), channelId, finalSsrcInfo.getStream(), null); + }else { + timeoutCallback.run(0, "鐐规挱瓒呮椂"); + mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); + mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); + streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); } - }, userSetting.getPlayTimeout()); + }, userSetting.getPlayTimeout()*1000); final String ssrc = ssrcInfo.getSsrc(); + final String stream = ssrcInfo.getStream(); cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString()); - timer.cancel(); + dynamicTask.stop(timeOutTaskKey); // hook鍝嶅簲 onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId, uuid); hookEvent.response(mediaServerItemInuse, response); @@ -273,9 +275,13 @@ if (ssrcIndex >= 0) { //ssrc瑙勫畾闀垮害涓�10瀛楄妭锛屼笉鍙栦綑涓嬮暱搴︿互閬垮厤鍚庣画杩樻湁鈥渇=鈥濆瓧娈� TODO 鍚庣画瀵逛笉瑙勮寖鐨勯潪10浣峴src鍏煎 String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); - if (!ssrc.equals(ssrcInResponse) && device.isSsrcCheck()) { // 鏌ヨ鍒皊src涓嶄竴鑷翠笖寮�鍚簡ssrc鏍¢獙鍒欓渶瑕侀拡瀵瑰鐞� - // 鏌ヨ ssrcInResponse 鏄惁鍙敤 - if (mediaServerItem.isRtpEnable() && !mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { + // 鏌ヨ鍒皊src涓嶄竴鑷翠笖寮�鍚簡ssrc鏍¢獙鍒欓渶瑕侀拡瀵瑰鐞� + if (ssrc.equals(ssrcInResponse)) { + return; + } + logger.info("[SIP 娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc 寮�鍚慨姝�"); + if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { + if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { // ssrc 涓嶅彲鐢� // 閲婃斁ssrc mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); @@ -285,14 +291,36 @@ errorEvent.response(event); return; } + + // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚� + if (!mediaServerItem.isRtpEnable()) { + // 娣诲姞璁㈤槄 + JSONObject subscribeKey = new JSONObject(); + subscribeKey.put("app", "rtp"); + subscribeKey.put("stream", stream); + subscribeKey.put("regist", true); + subscribeKey.put("schema", "rtmp"); + subscribeKey.put("mediaServerId", mediaServerItem.getId()); + subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed,subscribeKey); + subscribeKey.put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase()); + subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey, + (MediaServerItem mediaServerItemInUse, JSONObject response)->{ + logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); + dynamicTask.stop(timeOutTaskKey); + // hook鍝嶅簲 + onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid); + hookEvent.response(mediaServerItemInUse, response); + }); + } // 鍏抽棴rtp server mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); // 閲嶆柊寮�鍚痵src server mediaServerService.openRTPServer(mediaServerItem, finalSsrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false); + } } }, (event) -> { - timer.cancel(); + dynamicTask.stop(timeOutTaskKey); mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); // 閲婃斁ssrc mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); @@ -334,7 +362,9 @@ @Override public MediaServerItem getNewMediaServerItem(Device device) { - if (device == null) return null; + if (device == null) { + return null; + } String mediaServerId = device.getMediaServerId(); MediaServerItem mediaServerItem; if (mediaServerId == null) { @@ -353,9 +383,11 @@ String endTime,InviteStreamCallback inviteStreamCallback, PlayBackCallback callback) { Device device = storager.queryVideoDevice(deviceId); - if (device == null) return null; + if (device == null) { + return null; + } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true, true); return playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback); } @@ -365,7 +397,9 @@ String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback) { - if (mediaServerItem == null || ssrcInfo == null) return null; + if (mediaServerItem == null || ssrcInfo == null) { + return null; + } String uuid = UUID.randomUUID().toString(); String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId; DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(30000L); @@ -380,34 +414,31 @@ msg.setId(uuid); msg.setKey(key); PlayBackResult<RequestMessage> playBackResult = new PlayBackResult<>(); - - Timer timer = new Timer(); - timer.schedule(new TimerTask() { - @Override - public void run() { - logger.warn(String.format("璁惧鍥炴斁瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); - playBackResult.setCode(-1); - playBackResult.setData(msg); - playBackCallback.call(playBackResult); - SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream()); + String playBackTimeOutTaskKey = UUID.randomUUID().toString(); + dynamicTask.startDelay(playBackTimeOutTaskKey, ()->{ + logger.warn(String.format("璁惧鍥炴斁瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); + playBackResult.setCode(-1); + playBackResult.setData(msg); + playBackCallback.call(playBackResult); + SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream()); + // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� + if (dialog != null) { // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� - if (dialog != null) { - // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� - cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null); - }else { - mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); - mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream()); - streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); - } cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null); - // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰 - playBackCallback.call(playBackResult); + }else { + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream()); + streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); } - }, userSetting.getPlayTimeout()); + cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null); + // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰 + playBackCallback.call(playBackResult); + }, userSetting.getPlayTimeout()*1000); + cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack, (InviteStreamInfo inviteStreamInfo) -> { logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + inviteStreamInfo.getResponse().toJSONString()); - timer.cancel(); + dynamicTask.stop(playBackTimeOutTaskKey); StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); if (streamInfo == null) { logger.warn("璁惧鍥炴斁API璋冪敤澶辫触锛�"); @@ -425,7 +456,7 @@ playBackResult.setResponse(inviteStreamInfo.getResponse()); playBackCallback.call(playBackResult); }, event -> { - timer.cancel(); + dynamicTask.stop(playBackTimeOutTaskKey); msg.setData(String.format("鍥炴斁澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg)); playBackResult.setCode(-1); playBackResult.setData(msg); @@ -439,16 +470,20 @@ @Override public DeferredResult<ResponseEntity<String>> download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) { Device device = storager.queryVideoDevice(deviceId); - if (device == null) return null; + if (device == null) { + return null; + } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true, true); return download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed,infoCallBack, hookCallBack); } @Override public DeferredResult<ResponseEntity<String>> download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) { - if (mediaServerItem == null || ssrcInfo == null) return null; + if (mediaServerItem == null || ssrcInfo == null) { + return null; + } String uuid = UUID.randomUUID().toString(); String key = DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId; DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(30000L); @@ -467,34 +502,31 @@ PlayBackResult<RequestMessage> downloadResult = new PlayBackResult<>(); downloadResult.setData(msg); - Timer timer = new Timer(); - timer.schedule(new TimerTask() { - @Override - public void run() { - logger.warn(String.format("褰曞儚涓嬭浇璇锋眰瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); - wvpResult.setCode(-1); - wvpResult.setMsg("褰曞儚涓嬭浇璇锋眰瓒呮椂"); - downloadResult.setCode(-1); - hookCallBack.call(downloadResult); - SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream()); + String downLoadTimeOutTaskKey = UUID.randomUUID().toString(); + dynamicTask.startDelay(downLoadTimeOutTaskKey, ()->{ + logger.warn(String.format("褰曞儚涓嬭浇璇锋眰瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); + wvpResult.setCode(-1); + wvpResult.setMsg("褰曞儚涓嬭浇璇锋眰瓒呮椂"); + downloadResult.setCode(-1); + hookCallBack.call(downloadResult); + SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream()); + // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� + if (dialog != null) { // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� - if (dialog != null) { - // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� - cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null); - }else { - mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); - mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream()); - streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); - } cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null); - // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰 - hookCallBack.call(downloadResult); + }else { + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream()); + streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); } - }, userSetting.getPlayTimeout()); + cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null); + // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰 + hookCallBack.call(downloadResult); + }, userSetting.getPlayTimeout()*1000); cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack, inviteStreamInfo -> { logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + inviteStreamInfo.getResponse().toJSONString()); - timer.cancel(); + dynamicTask.stop(downLoadTimeOutTaskKey); StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); streamInfo.setStartTime(startTime); streamInfo.setEndTime(endTime); @@ -515,7 +547,7 @@ downloadResult.setResponse(inviteStreamInfo.getResponse()); hookCallBack.call(downloadResult); }, event -> { - timer.cancel(); + dynamicTask.stop(downLoadTimeOutTaskKey); downloadResult.setCode(-1); wvpResult.setCode(-1); wvpResult.setMsg(String.format("褰曞儚涓嬭浇澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg)); @@ -556,7 +588,7 @@ BigDecimal currentCount = new BigDecimal(duration/1000); BigDecimal totalCount = new BigDecimal(end-start); - BigDecimal divide = currentCount.divide(totalCount,2, BigDecimal.ROUND_HALF_UP); + BigDecimal divide = currentCount.divide(totalCount,2, RoundingMode.HALF_UP); double process = divide.doubleValue(); streamInfo.setProgress(process); } -- Gitblit v1.8.0