From ba8fffd907b5c3a7acfd3348c80e557b816cec98 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期一, 03 七月 2023 16:53:03 +0800 Subject: [PATCH] 增加与第三方对接的接口 --- src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 475 ++++++++++++++++++++++++++++++++++++++--------------------- 1 files changed, 307 insertions(+), 168 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 2dd4e61..03e00c2 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 @@ -11,6 +11,7 @@ import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; +import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; @@ -33,19 +34,16 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; -import com.genersoft.iot.vmp.utils.redis.RedisUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; -import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Service; import org.springframework.util.ObjectUtils; -import org.springframework.web.context.request.async.DeferredResult; +import javax.sdp.*; import javax.sip.InvalidArgumentException; import javax.sip.ResponseEvent; import javax.sip.SipException; @@ -53,8 +51,8 @@ import java.math.RoundingMode; import java.text.ParseException; import java.util.List; -import java.util.Objects; import java.util.UUID; +import java.util.Vector; @SuppressWarnings(value = {"rawtypes", "unchecked"}) @Service @@ -105,52 +103,27 @@ @Autowired private ZlmHttpHookSubscribe subscribe; - - @Qualifier("taskExecutor") @Autowired - private ThreadPoolTaskExecutor taskExecutor; + private SSRCFactory ssrcFactory; + + @Autowired + private RedisTemplate<Object, Object> redisTemplate; + @Override - public PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, - ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, - Runnable timeoutCallback) { + public void play(MediaServerItem mediaServerItem, String deviceId, String channelId, + ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, + Runnable timeoutCallback) { if (mediaServerItem == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "鏈壘鍒板彲鐢ㄧ殑zlm"); } - PlayResult playResult = new PlayResult(); - RequestMessage msg = new RequestMessage(); String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId; + + RequestMessage msg = new RequestMessage(); msg.setKey(key); - String uuid = UUID.randomUUID().toString(); - msg.setId(uuid); - playResult.setUuid(uuid); - DeferredResult<WVPResult<StreamInfo>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); - playResult.setResult(result); - // 褰曞儚鏌ヨ浠hannelId浣滀负deviceId鏌ヨ - resultHolder.put(key, uuid, result); Device device = redisCatchStorage.getDevice(deviceId); StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); - playResult.setDevice(device); - - result.onCompletion(() -> { - // 鐐规挱缁撴潫鏃惰皟鐢ㄦ埅鍥炬帴鍙� - taskExecutor.execute(() -> { - // TODO 搴旇鍦ㄤ笂娴佹椂璋冪敤鏇村ソ锛岀粨鏉熶篃鍙兘鏄敊璇粨鏉� - String path = "snap"; - String fileName = deviceId + "_" + channelId + ".jpg"; - WVPResult wvpResult = (WVPResult) result.getResult(); - if (Objects.requireNonNull(wvpResult).getCode() == 0) { - StreamInfo streamInfoForSuccess = (StreamInfo) wvpResult.getData(); - MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId()); - String streamUrl = streamInfoForSuccess.getFmp4(); - - // 璇锋眰鎴浘 - logger.info("[璇锋眰鎴浘]: " + fileName); - zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName); - } - }); - }); if (streamInfo != null) { String streamId = streamInfo.getStream(); @@ -160,7 +133,7 @@ wvpResult.setMsg("鐐规挱澶辫触锛� redis缂撳瓨streamId绛変簬null"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); - return playResult; + return; } String mediaServerId = streamInfo.getMediaServerId(); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); @@ -170,7 +143,7 @@ if (rtpInfo.getBoolean("exist")) { int localPort = rtpInfo.getInteger("local_port"); if (localPort == 0) { - logger.warn("[鐐规挱]锛岀偣鎾椂鍙戠幇rtpServerC瀛樺湪锛屼絾鏄皻鏈紑濮嬫帹娴�"); + logger.warn("[鐐规挱]锛岀偣鎾椂鍙戠幇rtpServer瀛樺湪锛屼絾鏄皻鏈紑濮嬫帹娴�"); // 姝ゆ椂璇存槑rtpServer宸茬粡鍒涘缓浣嗘槸娴佽繕娌℃湁鎺ㄤ笂鏉� WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); @@ -178,14 +151,13 @@ msg.setData(wvpResult); resultHolder.invokeAllResult(msg); - return playResult; + return; } else { WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.SUCCESS.getCode()); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); wvpResult.setData(streamInfo); msg.setData(wvpResult); - resultHolder.invokeAllResult(msg); if (hookEvent != null) { hookEvent.response(mediaServerItem, JSON.parseObject(JSON.toJSONString(streamInfo))); @@ -210,8 +182,7 @@ if (mediaServerItem.isRtpEnable()) { streamId = String.format("%s_%s", device.getDeviceId(), channelId); } - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); - logger.info(JSONObject.toJSONString(ssrcInfo)); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam()); if (ssrcInfo == null) { WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); @@ -219,7 +190,7 @@ msg.setData(wvpResult); resultHolder.invokeAllResult(msg); - return playResult; + return; } play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response) -> { if (hookEvent != null) { @@ -247,18 +218,24 @@ msg.setData(wvpResult); // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰 resultHolder.invokeAllResult(msg); - }, uuid); + }); } - return playResult; } @Override public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, - InviteTimeOutCallback timeoutCallback, String uuid) { + InviteTimeOutCallback timeoutCallback) { - logger.info("[鐐规挱寮�濮媇 deviceId: {}, channelId: {},鏀舵祦绔彛锛� {}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); + logger.info("\r\n" + + "[鐐规挱寮�濮媇 \r\n" + + "deviceId : {}, \r\n" + + "channelId : {},\r\n" + + "鏀舵祦绔彛 锛歿}, \r\n" + + "鏀舵祦妯″紡 锛歿}, \r\n" + + "SSRC : {}, \r\n" + + "SSRC鏍¢獙 锛歿} ", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); // 瓒呮椂澶勭悊 String timeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(timeOutTaskKey, () -> { @@ -276,19 +253,46 @@ mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); + // 鍙栨秷璁㈤槄娑堟伅鐩戝惉 + HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId()); + subscribe.removeSubscribe(hookSubscribe); } } }, userSetting.getPlayTimeout()); + //绔彛鑾峰彇澶辫触鐨剆srcInfo 娌℃湁蹇呰鍙戦�佺偣鎾寚浠� + if (ssrcInfo.getPort() <= 0) { + logger.info("[鐐规挱绔彛鍒嗛厤寮傚父]锛宒eviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, JSON.toJSONString(ssrcInfo)); + dynamicTask.stop(timeOutTaskKey); + // 閲婃斁ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); + RequestMessage msg = new RequestMessage(); + msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + device.getDeviceId() + channelId); + msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "鐐规挱绔彛鍒嗛厤寮傚父")); + resultHolder.invokeAllResult(msg); + return; + } try { cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString()); - System.out.println("鍋滄瓒呮椂浠诲姟锛� " + timeOutTaskKey); + dynamicTask.stop(timeOutTaskKey); // hook鍝嶅簲 - onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId, uuid); + onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId); hookEvent.response(mediaServerItemInuse, response); logger.info("[鐐规挱鎴愬姛] deviceId: {}, channelId: {}", device.getDeviceId(), channelId); + String streamUrl; + if (mediaServerItemInuse.getRtspPort() != 0) { + streamUrl = String.format("rtsp://127.0.0.1:%s/%s/%s", mediaServerItemInuse.getRtspPort(), "rtp", ssrcInfo.getStream()); + }else { + streamUrl = String.format("http://127.0.0.1:%s/%s/%s.live.mp4", mediaServerItemInuse.getHttpPort(), "rtp", ssrcInfo.getStream()); + } + String path = "snap"; + String fileName = device.getDeviceId() + "_" + channelId + ".jpg"; + // 璇锋眰鎴浘 + logger.info("[璇锋眰鎴浘]: " + fileName); + zlmresTfulUtils.getSnap(mediaServerItemInuse, streamUrl, 15, 1, path, fileName); }, (event) -> { ResponseEvent responseEvent = (ResponseEvent) event.event; @@ -298,26 +302,40 @@ // 妫�鏌ユ槸鍚︽湁y瀛楁 if (ssrcIndex >= 0) { //ssrc瑙勫畾闀垮害涓�10瀛楄妭锛屼笉鍙栦綑涓嬮暱搴︿互閬垮厤鍚庣画杩樻湁鈥渇=鈥濆瓧娈� TODO 鍚庣画瀵逛笉瑙勮寖鐨勯潪10浣峴src鍏煎 - String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); + String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12).trim(); // 鏌ヨ鍒皊src涓嶄竴鑷翠笖寮�鍚簡ssrc鏍¢獙鍒欓渶瑕侀拡瀵瑰鐞� if (ssrcInfo.getSsrc().equals(ssrcInResponse)) { + if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) { + String substring = contentString.substring(0, contentString.indexOf("y=")); + try { + SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring); + int port = -1; + Vector mediaDescriptions = sdp.getMediaDescriptions(true); + for (Object description : mediaDescriptions) { + MediaDescription mediaDescription = (MediaDescription) description; + Media media = mediaDescription.getMedia(); + + Vector mediaFormats = media.getMediaFormats(false); + if (mediaFormats.contains("96")) { + port = media.getMediaPort(); + break; + } + } + logger.info("[鐐规挱-TCP涓诲姩杩炴帴瀵规柟] deviceId: {}, channelId: {}, 杩炴帴瀵规柟鐨勫湴鍧�锛歿}:{}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, sdp.getConnection().getAddress(), port, device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); + JSONObject jsonObject = zlmresTfulUtils.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getStream()); + logger.info("[鐐规挱-TCP涓诲姩杩炴帴瀵规柟] 缁撴灉锛� {}", jsonObject); + } catch (SdpException e) { + logger.error("[鐐规挱-TCP涓诲姩杩炴帴瀵规柟] deviceId: {}, channelId: {}, 瑙f瀽200OK鐨凷DP淇℃伅澶辫触", device.getDeviceId(), channelId, e); + } + } return; } logger.info("[鐐规挱娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse); if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { logger.info("[鐐规挱娑堟伅] SSRC淇 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); - if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { - // ssrc 涓嶅彲鐢� - // 閲婃斁ssrc - mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); - streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - event.msg = "涓嬬骇鑷畾涔変簡ssrc,浣嗘槸姝src涓嶅彲鐢�"; - event.statusCode = 400; - errorEvent.response(event); - return; - } - + // 閲婃斁涓嶈浣跨敤鐨剆src + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚� if (!mediaServerItem.isRtpEnable()) { // 娣诲姞璁㈤槄 @@ -328,14 +346,34 @@ logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); dynamicTask.stop(timeOutTaskKey); // hook鍝嶅簲 - onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid); + onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId); hookEvent.response(mediaServerItemInUse, response); }); } + // 鍏抽棴rtp server - mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); - // 閲嶆柊寮�鍚痵src server - mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false, ssrcInfo.getPort()); + mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream(), result->{ + if (result) { + // 閲嶆柊寮�鍚痵src server + mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false, ssrcInfo.getPort(), true, device.getStreamModeForParam()); + }else { + try { + logger.warn("[鍋滄鐐规挱] {}/{}", device.getDeviceId(), channelId); + cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null); + } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍋滄鐐规挱锛� 鍙戦�丅YE: {}", e.getMessage()); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "鍛戒护鍙戦�佸け璐�: " + e.getMessage()); + } + + dynamicTask.stop(timeOutTaskKey); + + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); + event.msg = "涓嬬骇鑷畾涔変簡ssrc,閲嶆柊璁剧疆鏀舵祦淇℃伅澶辫触"; + event.statusCode = 500; + errorEvent.response(event); + } + }); + } } @@ -357,20 +395,19 @@ mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); + SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(); + eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent; + eventResult.statusCode = -1; eventResult.msg = "鍛戒护鍙戦�佸け璐�"; errorEvent.response(eventResult); } } @Override - public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) { - RequestMessage msg = new RequestMessage(); - if (uuid != null) { - msg.setId(uuid); - } - msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId); + public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId) { StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId); + RequestMessage msg = new RequestMessage(); + msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId); if (streamInfo != null) { DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); if (deviceChannel != null) { @@ -383,13 +420,39 @@ wvpResult.setCode(ErrorCode.SUCCESS.getCode()); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); wvpResult.setData(streamInfo); - msg.setData(wvpResult); + msg.setData(wvpResult); resultHolder.invokeAllResult(msg); + } else { logger.warn("璁惧棰勮API璋冪敤澶辫触锛�"); msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "璁惧棰勮API璋冪敤澶辫触锛�")); resultHolder.invokeAllResult(msg); + } + } + + private void onPublishHandlerForPlayback(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, PlayBackCallback playBackCallback) { + + StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId); + PlayBackResult<StreamInfo> playBackResult = new PlayBackResult<>(); + if (streamInfo != null) { + DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); + if (deviceChannel != null) { + deviceChannel.setStreamId(streamInfo.getStream()); + storager.startPlay(deviceId, channelId, streamInfo.getStream()); + } + redisCatchStorage.startPlay(streamInfo); + + + playBackResult.setCode(ErrorCode.SUCCESS.getCode()); + playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); + playBackResult.setData(streamInfo); + playBackCallback.call(playBackResult); + } else { + logger.warn("褰曞儚鍥炴斁璋冪敤澶辫触锛�"); + playBackResult.setCode(ErrorCode.ERROR100.getCode()); + playBackResult.setMsg("褰曞儚鍥炴斁璋冪敤澶辫触锛�"); + playBackCallback.call(playBackResult); } } @@ -400,7 +463,7 @@ } MediaServerItem mediaServerItem; if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) { - mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(); + mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(null); } else { mediaServerItem = mediaServerService.getOne(device.getMediaServerId()); } @@ -411,45 +474,55 @@ } @Override - public DeferredResult<WVPResult<StreamInfo>> playBack(String deviceId, String channelId, String startTime, + public MediaServerItem getNewMediaServerItemHasAssist(Device device) { + if (device == null) { + return null; + } + MediaServerItem mediaServerItem; + if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) { + mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(true); + } else { + mediaServerItem = mediaServerService.getOne(device.getMediaServerId()); + } + if (mediaServerItem == null) { + logger.warn("[鑾峰彇鍙敤鐨刏LM鑺傜偣]鏈壘鍒板彲浣跨敤鐨刏LM..."); + } + return mediaServerItem; + } + + @Override + public void playBack(String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback inviteStreamCallback, PlayBackCallback callback) { Device device = storager.queryVideoDevice(deviceId); if (device == null) { - return null; + return; } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, device.isSsrcCheck(), true); - - return playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam()); + playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback); } @Override - public DeferredResult<WVPResult<StreamInfo>> playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, + public void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback) { if (mediaServerItem == null || ssrcInfo == null) { - return null; + return; } - String uuid = UUID.randomUUID().toString(); - String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId; + Device device = storager.queryVideoDevice(deviceId); if (device == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "璁惧锛� " + deviceId + "涓嶅瓨鍦�"); } - DeferredResult<WVPResult<StreamInfo>> result = new DeferredResult<>(30000L); - resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid, result); - RequestMessage requestMessage = new RequestMessage(); - requestMessage.setId(uuid); - requestMessage.setKey(key); - PlayBackResult<RequestMessage> playBackResult = new PlayBackResult<>(); + logger.info("[鍥炴斁娑堟伅] deviceId: {}, channelId: {},鏀舵祦绔彛锛� {}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); + PlayBackResult<StreamInfo> playBackResult = new PlayBackResult<>(); String playBackTimeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(playBackTimeOutTaskKey, () -> { logger.warn(String.format("璁惧鍥炴斁瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); playBackResult.setCode(ErrorCode.ERROR100.getCode()); playBackResult.setMsg("鍥炴斁瓒呮椂"); - playBackResult.setData(requestMessage); try { cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null); @@ -461,19 +534,14 @@ mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); } - // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰 playBackCallback.call(playBackResult); - result.setResult(WVPResult.fail(ErrorCode.ERROR100.getCode(), "鍥炴斁瓒呮椂")); - resultHolder.exist(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid); }, userSetting.getPlayTimeout()); SipSubscribe.Event errorEvent = event -> { dynamicTask.stop(playBackTimeOutTaskKey); - requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("鍥炴斁澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg))); playBackResult.setCode(ErrorCode.ERROR100.getCode()); playBackResult.setMsg(String.format("鍥炴斁澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg)); - playBackResult.setData(requestMessage); playBackResult.setEvent(event); playBackCallback.call(playBackResult); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); @@ -491,11 +559,9 @@ return; } redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId()); - WVPResult<StreamInfo> success = WVPResult.success(streamInfo); - requestMessage.setData(success); playBackResult.setCode(ErrorCode.SUCCESS.getCode()); playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); - playBackResult.setData(requestMessage); + playBackResult.setData(streamInfo); playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem()); playBackResult.setResponse(inviteStreamInfo.getResponse()); playBackCallback.call(playBackResult); @@ -521,16 +587,8 @@ if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { logger.info("[鍥炴斁娑堟伅] SSRC淇 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); - if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { - // ssrc 涓嶅彲鐢� - // 閲婃斁ssrc - mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); - streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - eventResult.msg = "涓嬬骇鑷畾涔変簡ssrc,浣嗘槸姝src涓嶅彲鐢�"; - eventResult.statusCode = 400; - errorEvent.response(eventResult); - return; - } + // 閲婃斁涓嶈浣跨敤鐨剆src + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚� if (!mediaServerItem.isRtpEnable()) { @@ -542,14 +600,35 @@ logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); dynamicTask.stop(playBackTimeOutTaskKey); // hook鍝嶅簲 - onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid); + onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, playBackCallback); hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream())); }); } + // 鍏抽棴rtp server - mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); - // 閲嶆柊寮�鍚痵src server - mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort()); + mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream(), result->{ + if (result) { + // 閲嶆柊寮�鍚痵src server + mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort(), true, device.getStreamModeForParam()); + }else { + try { + logger.warn("[鍥炴斁娑堟伅]鍋滄 {}/{}", device.getDeviceId(), channelId); + cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null); + } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍋滄鐐规挱 鍋滄锛� 鍙戦�丅YE: {}", e.getMessage()); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "鍛戒护鍙戦�佸け璐�: " + e.getMessage()); + } + + dynamicTask.stop(playBackTimeOutTaskKey); + // 閲婃斁ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); + errorEvent.response(eventResult); + eventResult.msg = "涓嬬骇鑷畾涔変簡ssrc,閲嶆柊璁剧疆鏀舵祦淇℃伅澶辫触"; + eventResult.statusCode = 500; + errorEvent.response(eventResult); + } + }); } } } @@ -558,52 +637,50 @@ } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 鍥炴斁: {}", e.getMessage()); - SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); + SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(); + eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent; + eventResult.statusCode = -1; eventResult.msg = "鍛戒护鍙戦�佸け璐�"; errorEvent.response(eventResult); } - return result; } + + @Override - public DeferredResult<WVPResult<StreamInfo>> download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) { + public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback) { Device device = storager.queryVideoDevice(deviceId); if (device == null) { - return null; + return; } - MediaServerItem newMediaServerItem = getNewMediaServerItem(device); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, device.isSsrcCheck(), true); - - return download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, infoCallBack, hookCallBack); + MediaServerItem newMediaServerItem = getNewMediaServerItemHasAssist(device); + if (newMediaServerItem == null) { + PlayBackResult<StreamInfo> downloadResult = new PlayBackResult<>(); + downloadResult.setCode(ErrorCode.ERROR100.getCode()); + downloadResult.setMsg("鏈壘鍒癮ssist鏈嶅姟"); + playBackCallback.call(downloadResult); + return; + } + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam()); + download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, infoCallBack, playBackCallback); } + @Override - public DeferredResult<WVPResult<StreamInfo>> download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) { + public void 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; + return; } - String uuid = UUID.randomUUID().toString(); - String key = DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId; - DeferredResult<WVPResult<StreamInfo>> result = new DeferredResult<>(30000L); + Device device = storager.queryVideoDevice(deviceId); if (device == null) { throw new ControllerException(ErrorCode.ERROR400.getCode(), "璁惧锛�" + deviceId + "涓嶅瓨鍦�"); } - - resultHolder.put(key, uuid, result); - RequestMessage requestMessage = new RequestMessage(); - requestMessage.setId(uuid); - requestMessage.setKey(key); - WVPResult<StreamInfo> wvpResult = new WVPResult<>(); - requestMessage.setData(wvpResult); - PlayBackResult<RequestMessage> downloadResult = new PlayBackResult<>(); - downloadResult.setData(requestMessage); - + PlayBackResult<StreamInfo> downloadResult = new PlayBackResult<>(); + logger.info("[褰曞儚涓嬭浇] deviceId: {}, channelId: {},鏀舵祦绔彛锛� {}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); String downLoadTimeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(downLoadTimeOutTaskKey, () -> { logger.warn(String.format("褰曞儚涓嬭浇璇锋眰瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); - wvpResult.setCode(ErrorCode.ERROR100.getCode()); - wvpResult.setMsg("褰曞儚涓嬭浇璇锋眰瓒呮椂"); downloadResult.setCode(ErrorCode.ERROR100.getCode()); downloadResult.setMsg("褰曞儚涓嬭浇璇锋眰瓒呮椂"); hookCallBack.call(downloadResult); @@ -618,47 +695,106 @@ mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); } - // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰 - hookCallBack.call(downloadResult); }, userSetting.getPlayTimeout()); SipSubscribe.Event errorEvent = event -> { dynamicTask.stop(downLoadTimeOutTaskKey); downloadResult.setCode(ErrorCode.ERROR100.getCode()); downloadResult.setMsg(String.format("褰曞儚涓嬭浇澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg)); - wvpResult.setCode(ErrorCode.ERROR100.getCode()); - wvpResult.setMsg(String.format("褰曞儚涓嬭浇澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg)); downloadResult.setEvent(event); hookCallBack.call(downloadResult); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); }; - + InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> { + logger.info("[褰曞儚涓嬭浇]鏀跺埌璁㈤槄娑堟伅锛� " + inviteStreamInfo.getCallId()); + dynamicTask.stop(downLoadTimeOutTaskKey); + StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); + streamInfo.setStartTime(startTime); + streamInfo.setEndTime(endTime); + redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId()); + downloadResult.setCode(ErrorCode.SUCCESS.getCode()); + downloadResult.setMsg(ErrorCode.SUCCESS.getMsg()); + downloadResult.setData(streamInfo); + downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem()); + downloadResult.setResponse(inviteStreamInfo.getResponse()); + hookCallBack.call(downloadResult); + }; try { cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack, - inviteStreamInfo -> { - logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + inviteStreamInfo.getResponse().toJSONString()); - dynamicTask.stop(downLoadTimeOutTaskKey); - StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); - streamInfo.setStartTime(startTime); - streamInfo.setEndTime(endTime); - redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId()); - wvpResult.setCode(ErrorCode.SUCCESS.getCode()); - wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); - wvpResult.setData(streamInfo); - downloadResult.setCode(ErrorCode.SUCCESS.getCode()); - downloadResult.setMsg(ErrorCode.SUCCESS.getMsg()); - downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem()); - downloadResult.setResponse(inviteStreamInfo.getResponse()); - hookCallBack.call(downloadResult); - }, errorEvent); + hookEvent, errorEvent, eventResult -> + { + if (eventResult.type == SipSubscribe.EventResultType.response) { + ResponseEvent responseEvent = (ResponseEvent) eventResult.event; + String contentString = new String(responseEvent.getResponse().getRawContent()); + // 鑾峰彇ssrc + int ssrcIndex = contentString.indexOf("y="); + // 妫�鏌ユ槸鍚︽湁y瀛楁 + if (ssrcIndex >= 0) { + //ssrc瑙勫畾闀垮害涓�10瀛楄妭锛屼笉鍙栦綑涓嬮暱搴︿互閬垮厤鍚庣画杩樻湁鈥渇=鈥濆瓧娈� TODO 鍚庣画瀵逛笉瑙勮寖鐨勯潪10浣峴src鍏煎 + String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); + // 鏌ヨ鍒皊src涓嶄竴鑷翠笖寮�鍚簡ssrc鏍¢獙鍒欓渶瑕侀拡瀵瑰鐞� + if (ssrcInfo.getSsrc().equals(ssrcInResponse)) { + return; + } + logger.info("[褰曞儚涓嬭浇] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse); + if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { + logger.info("[褰曞儚涓嬭浇] SSRC淇 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); + + // 閲婃斁涓嶈浣跨敤鐨剆src + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + + // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚� + if (!mediaServerItem.isRtpEnable()) { + // 娣诲姞璁㈤槄 + HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId()); + subscribe.removeSubscribe(hookSubscribe); + hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase()); + subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> { + logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); + dynamicTask.stop(downLoadTimeOutTaskKey); + // hook鍝嶅簲 + onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, hookCallBack); + hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream())); + }); + } + + // 鍏抽棴rtp server + mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream(), result->{ + if (result) { + // 閲嶆柊寮�鍚痵src server + mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort(), true, device.getStreamModeForParam()); + }else { + try { + logger.warn("[褰曞儚涓嬭浇] 鍋滄{}/{}", device.getDeviceId(), channelId); + cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null); + } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { + logger.error("[鍛戒护鍙戦�佸け璐 褰曞儚涓嬭浇鍋滄锛� 鍙戦�丅YE: {}", e.getMessage()); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "鍛戒护鍙戦�佸け璐�: " + e.getMessage()); + } + + dynamicTask.stop(downLoadTimeOutTaskKey); + // 閲婃斁ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); + eventResult.msg = "涓嬬骇鑷畾涔変簡ssrc,閲嶆柊璁剧疆鏀舵祦淇℃伅澶辫触"; + eventResult.statusCode = 500; + errorEvent.response(eventResult); + } + }); + } + } + } + }); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 褰曞儚涓嬭浇: {}", e.getMessage()); - SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); + SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(); + eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent; + eventResult.statusCode = -1; eventResult.msg = "鍛戒护鍙戦�佸け璐�"; errorEvent.response(eventResult); } - return result; } @Override @@ -678,7 +814,10 @@ } if (mediaServerItem.getRecordAssistPort() > 0) { JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, streamInfo.getApp(), streamInfo.getStream(), null); - if (jsonObject != null && jsonObject.getInteger("code") == 0) { + if (jsonObject == null) { + throw new ControllerException(ErrorCode.ERROR100.getCode(), "杩炴帴Assist鏈嶅姟澶辫触"); + } + if (jsonObject.getInteger("code") == 0) { long duration = jsonObject.getLong("data"); if (duration == 0) { @@ -826,7 +965,7 @@ throw new ServiceException("streamId涓嶅瓨鍦�"); } streamInfo.setPause(true); - RedisUtil.set(key, streamInfo); + redisTemplate.opsForValue().set(key, streamInfo); MediaServerItem mediaServerItem = mediaServerService.getOne(streamInfo.getMediaServerId()); if (null == mediaServerItem) { logger.warn("mediaServer 涓嶅瓨鍦�!"); @@ -850,7 +989,7 @@ throw new ServiceException("streamId涓嶅瓨鍦�"); } streamInfo.setPause(false); - RedisUtil.set(key, streamInfo); + redisTemplate.opsForValue().set(key, streamInfo); MediaServerItem mediaServerItem = mediaServerService.getOne(streamInfo.getMediaServerId()); if (null == mediaServerItem) { logger.warn("mediaServer 涓嶅瓨鍦�!"); -- Gitblit v1.8.0