From cec266d3cd27e79e61b7e614d0c85e0b43492115 Mon Sep 17 00:00:00 2001 From: dsk <2514169858@qq.com> Date: 星期日, 25 六月 2023 16:00:37 +0800 Subject: [PATCH] 修复请求报错404等情况下导致ResponseBody未关闭的问题 --- src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 426 +++++++++++++++++++++++++++++++++++----------------- 1 files changed, 284 insertions(+), 142 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 a12cf23..e52fac2 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 @@ -1,6 +1,5 @@ package com.genersoft.iot.vmp.service.impl; -import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.common.InviteInfo; import com.genersoft.iot.vmp.common.InviteSessionStatus; @@ -25,6 +24,8 @@ import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; +import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam; +import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam; import com.genersoft.iot.vmp.service.*; import com.genersoft.iot.vmp.service.bean.ErrorCallback; import com.genersoft.iot.vmp.service.bean.InviteErrorCode; @@ -115,28 +116,43 @@ @Override - public SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId, ErrorCallback<Object> callback) { + public SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId,boolean isSubStream, ErrorCallback<Object> callback) { if (mediaServerItem == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "鏈壘鍒板彲鐢ㄧ殑zlm"); } Device device = redisCatchStorage.getDevice(deviceId); - InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); - + InviteInfo inviteInfo; + if(device.isSwitchPrimarySubStream()){ + inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId,isSubStream); + }else { + inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + } if (inviteInfo != null ) { if (inviteInfo.getStreamInfo() == null) { // 鐐规挱鍙戣捣浜嗕絾鏄皻鏈垚鍔�, 浠呮敞鍐屽洖璋冪瓑寰呯粨鏋滃嵆鍙� - inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback); + if(device.isSwitchPrimarySubStream()){ + inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId,isSubStream, null, callback); + }else { + inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback); + } return inviteInfo.getSsrcInfo(); }else { StreamInfo streamInfo = inviteInfo.getStreamInfo(); String streamId = streamInfo.getStream(); if (streamId == null) { callback.run(InviteErrorCode.ERROR_FOR_CATCH_DATA.getCode(), "鐐规挱澶辫触锛� redis缂撳瓨streamId绛変簬null", null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_CATCH_DATA.getCode(), - "鐐规挱澶辫触锛� redis缂撳瓨streamId绛変簬null", - null); + if(device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_CATCH_DATA.getCode(), + "鐐规挱澶辫触锛� redis缂撳瓨streamId绛変簬null", + null); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_CATCH_DATA.getCode(), + "鐐规挱澶辫触锛� redis缂撳瓨streamId绛変簬null", + null); + } return inviteInfo.getSsrcInfo(); } String mediaServerId = streamInfo.getMediaServerId(); @@ -145,41 +161,64 @@ Boolean ready = zlmrtpServerFactory.isStreamReady(mediaInfo, "rtp", streamId); if (ready != null && ready) { callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.SUCCESS.getCode(), - InviteErrorCode.SUCCESS.getMsg(), - streamInfo); + if(device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); + } return inviteInfo.getSsrcInfo(); }else { // 鐐规挱鍙戣捣浜嗕絾鏄皻鏈垚鍔�, 浠呮敞鍐屽洖璋冪瓑寰呯粨鏋滃嵆鍙� - inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback); - storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); - inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + if(device.isSwitchPrimarySubStream()) { + inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback); + storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + }else { + inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId,isSubStream, null, callback); + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId,isSubStream); + } } } } String streamId = null; if (mediaServerItem.isRtpEnable()) { - streamId = String.format("%s_%s", device.getDeviceId(), channelId); + if(device.isSwitchPrimarySubStream()){ + streamId = StreamInfo.getPlayStream(deviceId, channelId, isSubStream); + }else { + streamId = String.format("%s_%s", device.getDeviceId(), channelId); + } } SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam()); if (ssrcInfo == null) { callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), - InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), - null); + if(device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), + null); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), + null); + } return null; } // TODO 璁板綍鐐规挱鐨勭姸鎬� - play(mediaServerItem, ssrcInfo, device, channelId, callback); + play(mediaServerItem, ssrcInfo, device, channelId,isSubStream, callback); return ssrcInfo; } @Override - public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, + public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,boolean isSubStream, ErrorCallback<Object> callback) { if (mediaServerItem == null || ssrcInfo == null) { @@ -188,8 +227,11 @@ null); return; } - logger.info("[鐐规挱寮�濮媇 deviceId: {}, channelId: {},鏀舵祦绔彛锛歿}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); - + if( device.isSwitchPrimarySubStream() ){ + logger.info("[鐐规挱寮�濮媇 deviceId: {}, channelId: {},鐮佹祦绫诲瀷锛歿},鏀舵祦绔彛锛� {}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId,isSubStream ? "杈呯爜娴�" : "涓荤爜娴�", ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); + }else { + logger.info("[鐐规挱寮�濮媇 deviceId: {}, channelId: {},鏀舵祦绔彛锛� {}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); + } //绔彛鑾峰彇澶辫触鐨剆srcInfo 娌℃湁蹇呰鍙戦�佺偣鎾寚浠� if (ssrcInfo.getPort() <= 0) { logger.info("[鐐规挱绔彛鍒嗛厤寮傚父]锛宒eviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo); @@ -198,23 +240,50 @@ streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "鐐规挱绔彛鍒嗛厤寮傚父", null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "鐐规挱绔彛鍒嗛厤寮傚父", null); + if(device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "鐐规挱绔彛鍒嗛厤寮傚父", null); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "鐐规挱绔彛鍒嗛厤寮傚父", null); + } return; } // 鍒濆鍖杛edis涓殑invite娑堟伅鐘舵�� - InviteInfo inviteInfo = InviteInfo.getinviteInfo(device.getDeviceId(), channelId, ssrcInfo.getStream(), ssrcInfo, - mediaServerItem.getSdpIp(), ssrcInfo.getPort(), device.getStreamMode(), InviteSessionType.PLAY, - InviteSessionStatus.ready); - inviteStreamService.updateInviteInfo(inviteInfo); + InviteInfo inviteInfo; + + if(device.isSwitchPrimarySubStream()){ + // 鍒濆鍖杛edis涓殑invite娑堟伅鐘舵�� + inviteInfo = InviteInfo.getInviteInfo(device.getDeviceId(), channelId,isSubStream, ssrcInfo.getStream(), ssrcInfo, + mediaServerItem.getSdpIp(), ssrcInfo.getPort(), device.getStreamMode(), InviteSessionType.PLAY, + InviteSessionStatus.ready); + inviteStreamService.updateInviteInfoSub(inviteInfo); + }else { + // 鍒濆鍖杛edis涓殑invite娑堟伅鐘舵�� + inviteInfo = InviteInfo.getinviteInfo(device.getDeviceId(), channelId, ssrcInfo.getStream(), ssrcInfo, + mediaServerItem.getSdpIp(), ssrcInfo.getPort(), device.getStreamMode(), InviteSessionType.PLAY, + InviteSessionStatus.ready); + inviteStreamService.updateInviteInfo(inviteInfo); + } // 瓒呮椂澶勭悊 String timeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(timeOutTaskKey, () -> { // 鎵ц瓒呮椂浠诲姟鏃舵煡璇㈡槸鍚﹀凡缁忔垚鍔燂紝鎴愬姛浜嗗垯涓嶆墽琛岃秴鏃朵换鍔★紝闃叉瓒呮椂浠诲姟鍙栨秷澶辫触鐨勬儏鍐� - InviteInfo inviteInfoForTimeOut = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + InviteInfo inviteInfoForTimeOut; + if(device.isSwitchPrimarySubStream()){ + // 鍒濆鍖杛edis涓殑invite娑堟伅鐘舵�� + inviteInfoForTimeOut = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream); + }else { + // 鍒濆鍖杛edis涓殑invite娑堟伅鐘舵�� + inviteInfoForTimeOut = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + } if (inviteInfoForTimeOut == null || inviteInfoForTimeOut.getStreamInfo() == null) { - logger.info("[鐐规挱瓒呮椂] 鏀舵祦瓒呮椂 deviceId: {}, channelId: {}锛岀鍙o細{}, SSRC: {}", device.getDeviceId(), channelId, ssrcInfo.getPort(), ssrcInfo.getSsrc()); + if( device.isSwitchPrimarySubStream()){ + logger.info("[鐐规挱瓒呮椂] 鏀舵祦瓒呮椂 deviceId: {}, channelId: {},鐮佹祦绫诲瀷锛歿}锛岀鍙o細{}, SSRC: {}", device.getDeviceId(), channelId,isSubStream ? "杈呯爜娴�" : "涓荤爜娴�", ssrcInfo.getPort(), ssrcInfo.getSsrc()); + }else { + logger.info("[鐐规挱瓒呮椂] 鏀舵祦瓒呮椂 deviceId: {}, channelId: {}锛岀鍙o細{}, SSRC: {}", device.getDeviceId(), channelId, ssrcInfo.getPort(), ssrcInfo.getSsrc()); + } // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� // InviteInfo inviteInfoForTimeout = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.play, device.getDeviceId(), channelId); // if (inviteInfoForTimeout == null) { @@ -226,10 +295,16 @@ // // TODO 鍙戦�乧ancel // } callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null); + if( device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null); + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream); - inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null); + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + } try { cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { @@ -247,25 +322,42 @@ }, userSetting.getPlayTimeout()); try { - cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { - logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString()); + cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId,isSubStream, (mediaServerItemInuse, hookParam ) -> { + logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + hookParam); dynamicTask.stop(timeOutTaskKey); // hook鍝嶅簲 - StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId); + StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInuse, hookParam, device.getDeviceId(), channelId,isSubStream); if (streamInfo == null){ callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), - InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + if( device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + } return; } callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.SUCCESS.getCode(), - InviteErrorCode.SUCCESS.getMsg(), - streamInfo); - logger.info("[鐐规挱鎴愬姛] deviceId: {}, channelId: {}", device.getDeviceId(), channelId); + if( device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); + } + if( device.isSwitchPrimarySubStream() ){ + logger.info("[鐐规挱鎴愬姛] deviceId: {}, channelId: {},鐮佹祦绫诲瀷锛歿}", device.getDeviceId(), channelId,isSubStream ? "杈呯爜娴�" : "涓荤爜娴�"); + }else { + 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()); @@ -321,9 +413,15 @@ callback.run(InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(), InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(), - InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null); + if(device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null); + } } } return; @@ -331,21 +429,8 @@ logger.info("[鐐规挱娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse); if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { logger.info("[鐐规挱娑堟伅] SSRC淇 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); - if (!ssrcFactory.checkSsrc(mediaServerItem.getId(),ssrcInResponse)) { - // ssrc 涓嶅彲鐢� - logger.info("[鐐规挱娑堟伅] SSRC淇鏃跺彂鐜皊src涓嶅彲浣跨敤 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); - // 閲婃斁ssrc - ssrcFactory.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); - streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - - callback.run(InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getCode(), - InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getMsg(), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getCode(), - InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getMsg(), null); - - return; - } + // 閲婃斁ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄噸鏂拌缃洃鍚嵆鍙� if (!mediaServerItem.isRtpEnable()) { // 娣诲姞璁㈤槄 @@ -354,25 +439,38 @@ String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase(); hookSubscribe.getContent().put("stream", stream); inviteInfo.setStream(stream); - subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> { - logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); + subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> { + logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + hookParam); dynamicTask.stop(timeOutTaskKey); // hook鍝嶅簲 - StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId); + StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInUse, hookParam, device.getDeviceId(), channelId,isSubStream); if (streamInfo == null){ callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), - InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + if( device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + } return; } callback.run(InviteErrorCode.SUCCESS.getCode(), - InviteErrorCode.SUCCESS.getMsg(), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.SUCCESS.getCode(), - InviteErrorCode.SUCCESS.getMsg(), - streamInfo); + InviteErrorCode.SUCCESS.getMsg(), streamInfo); + if( device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); + } }); return; } @@ -395,9 +493,15 @@ callback.run(InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), "涓嬬骇鑷畾涔変簡ssrc,閲嶆柊璁剧疆鏀舵祦淇℃伅澶辫触", null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), - "涓嬬骇鑷畾涔変簡ssrc,閲嶆柊璁剧疆鏀舵祦淇℃伅澶辫触", null); + if( device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), + "涓嬬骇鑷畾涔変簡ssrc,閲嶆柊璁剧疆鏀舵祦淇℃伅澶辫触", null); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), + "涓嬬骇鑷畾涔変簡ssrc,閲嶆柊璁剧疆鏀舵祦淇℃伅澶辫触", null); + } }else { ssrcInfo.setSsrc(ssrcInResponse); @@ -408,7 +512,11 @@ logger.info("[鐐规挱娑堟伅] 鏀跺埌invite 200, 涓嬬骇鑷畾涔変簡ssrc, 浣嗘槸褰撳墠妯″紡鏃犻渶淇"); } } - inviteStreamService.updateInviteInfo(inviteInfo); + if(device.isSwitchPrimarySubStream()){ + inviteStreamService.updateInviteInfoSub(inviteInfo); + }else { + inviteStreamService.updateInviteInfo(inviteInfo); + } }, (event) -> { dynamicTask.stop(timeOutTaskKey); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); @@ -419,11 +527,19 @@ callback.run(InviteErrorCode.ERROR_FOR_SIGNALLING_ERROR.getCode(), String.format("鐐规挱澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), - String.format("鐐规挱澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg), null); + if( device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), + String.format("鐐规挱澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg), null); - inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), + String.format("鐐规挱澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg), null); + + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + } }); } catch (InvalidArgumentException | SipException | ParseException e) { @@ -437,36 +553,61 @@ callback.run(InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(), InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getMsg(), null); - inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, - InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(), - InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getMsg(), null); + if( device.isSwitchPrimarySubStream()){ + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream, null, + InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(), + InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getMsg(), null); - inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId,isSubStream); + }else { + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(), + InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getMsg(), null); + + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + } } } - private StreamInfo onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId) { - StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId); + private StreamInfo onPublishHandlerForPlay(MediaServerItem mediaServerItem, HookParam hookParam, String deviceId, String channelId, boolean isSubStream) { + StreamInfo streamInfo = null; + Device device = redisCatchStorage.getDevice(deviceId); + OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam; + if( device.isSwitchPrimarySubStream() ){ + streamInfo = onPublishHandler(mediaServerItem, streamChangedHookParam, deviceId, channelId,isSubStream); + }else { + streamInfo = onPublishHandler(mediaServerItem, streamChangedHookParam, deviceId, channelId); + } if (streamInfo != null) { - DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); - if (deviceChannel != null) { - deviceChannel.setStreamId(streamInfo.getStream()); - storager.startPlay(deviceId, channelId, streamInfo.getStream()); + InviteInfo inviteInfo; + if(device.isSwitchPrimarySubStream()){ + inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId,isSubStream); + }else { + DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); + if (deviceChannel != null) { + deviceChannel.setStreamId(streamInfo.getStream()); + storager.startPlay(deviceId, channelId, streamInfo.getStream()); + } + inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); } - InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); if (inviteInfo != null) { inviteInfo.setStatus(InviteSessionStatus.ok); inviteInfo.setStreamInfo(streamInfo); - inviteStreamService.updateInviteInfo(inviteInfo); + if(device.isSwitchPrimarySubStream()){ + inviteStreamService.updateInviteInfoSub(inviteInfo); + }else { + inviteStreamService.updateInviteInfo(inviteInfo); + } + } } return streamInfo; } - private StreamInfo onPublishHandlerForPlayback(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String startTime, String endTime) { - - StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId); + private StreamInfo onPublishHandlerForPlayback(MediaServerItem mediaServerItem, HookParam param, String deviceId, String channelId, String startTime, String endTime) { + OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam) param; + StreamInfo streamInfo = onPublishHandler(mediaServerItem, streamChangedHookParam, deviceId, channelId); if (streamInfo != null) { streamInfo.setStartTime(startTime); streamInfo.setEndTime(endTime); @@ -585,10 +726,10 @@ inviteStreamService.removeInviteInfo(inviteInfo); }; - ZlmHttpHookSubscribe.Event hookEvent = (mediaServerItemInuse, jsonObject) -> { - logger.info("鏀跺埌鍥炴斁璁㈤槄娑堟伅锛� " + jsonObject); + ZlmHttpHookSubscribe.Event hookEvent = (mediaServerItemInuse, hookParam) -> { + logger.info("鏀跺埌鍥炴斁璁㈤槄娑堟伅锛� " + hookParam); dynamicTask.stop(playBackTimeOutTaskKey); - StreamInfo streamInfo = onPublishHandlerForPlayback(mediaServerItemInuse, jsonObject, deviceId, channelId, startTime, endTime); + StreamInfo streamInfo = onPublishHandlerForPlayback(mediaServerItemInuse, hookParam, deviceId, channelId, startTime, endTime); if (streamInfo == null) { logger.warn("璁惧鍥炴斁API璋冪敤澶辫触锛�"); callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), @@ -654,17 +795,8 @@ if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { logger.info("[褰曞儚鍥炴斁] SSRC淇 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); - if (!ssrcFactory.checkSsrc(mediaServerItem.getId(),ssrcInResponse)) { - // ssrc 涓嶅彲鐢� - logger.info("[褰曞儚鍥炴斁] SSRC淇鏃跺彂鐜皊src涓嶅彲浣跨敤 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); - // 閲婃斁ssrc - dynamicTask.stop(playBackTimeOutTaskKey); - mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); - streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - callback.run(InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getCode(), - InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getMsg(), null); - return; - } + // 閲婃斁ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚� if (!mediaServerItem.isRtpEnable()) { @@ -674,11 +806,11 @@ String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase(); hookSubscribe.getContent().put("stream", stream); inviteInfo.setStream(stream); - subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> { - logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); + subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> { + logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + hookParam); dynamicTask.stop(playBackTimeOutTaskKey); // hook鍝嶅簲 - hookEvent.response(mediaServerItemInUse, response); + hookEvent.response(mediaServerItemInUse, hookParam); }); } // 鏇存柊ssrc @@ -790,10 +922,10 @@ streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); inviteStreamService.removeInviteInfo(inviteInfo); }; - ZlmHttpHookSubscribe.Event hookEvent = (mediaServerItemInuse, jsonObject) -> { - logger.info("[褰曞儚涓嬭浇]鏀跺埌璁㈤槄娑堟伅锛� " + jsonObject); + ZlmHttpHookSubscribe.Event hookEvent = (mediaServerItemInuse, hookParam) -> { + logger.info("[褰曞儚涓嬭浇]鏀跺埌璁㈤槄娑堟伅锛� " + hookParam); dynamicTask.stop(downLoadTimeOutTaskKey); - StreamInfo streamInfo = onPublishHandlerForDownload(mediaServerItemInuse, jsonObject, deviceId, channelId, startTime, endTime); + StreamInfo streamInfo = onPublishHandlerForDownload(mediaServerItemInuse, hookParam, deviceId, channelId, startTime, endTime); if (streamInfo == null) { logger.warn("[褰曞儚涓嬭浇] 鑾峰彇娴佸湴鍧�淇℃伅澶辫触"); callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), @@ -858,15 +990,8 @@ if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { logger.info("[褰曞儚涓嬭浇] SSRC淇 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); - if (!ssrcFactory.checkSsrc(mediaServerItem.getId(),ssrcInResponse)) { - // ssrc 涓嶅彲鐢� - // 閲婃斁ssrc - mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); - streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - callback.run(InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getCode(), - InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getMsg(), null); - return; - } + // 閲婃斁ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚� if (!mediaServerItem.isRtpEnable()) { @@ -874,10 +999,10 @@ 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()); + subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam) -> { + logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + hookParam); dynamicTask.stop(downLoadTimeOutTaskKey); - hookEvent.response(mediaServerItemInUse, response); + hookEvent.response(mediaServerItemInUse, hookParam); }); } @@ -967,8 +1092,9 @@ return null; } - private StreamInfo onPublishHandlerForDownload(MediaServerItem mediaServerItemInuse, JSONObject response, String deviceId, String channelId, String startTime, String endTime) { - StreamInfo streamInfo = onPublishHandler(mediaServerItemInuse, response, deviceId, channelId); + private StreamInfo onPublishHandlerForDownload(MediaServerItem mediaServerItemInuse, HookParam hookParam, String deviceId, String channelId, String startTime, String endTime) { + OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam) hookParam; + StreamInfo streamInfo = onPublishHandler(mediaServerItemInuse, streamChangedHookParam, deviceId, channelId); if (streamInfo != null) { streamInfo.setProgress(0); streamInfo.setStartTime(startTime); @@ -985,14 +1111,13 @@ } - public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId) { - String streamId = resonse.getString("stream"); - JSONArray tracks = resonse.getJSONArray("tracks"); - StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, "rtp", streamId, tracks, null); + public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, OnStreamChangedHookParam hookParam, String deviceId, String channelId) { + StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, "rtp", hookParam.getStream(), hookParam.getTracks(), null); streamInfo.setDeviceID(deviceId); streamInfo.setChannelId(channelId); return streamInfo; } + @Override public void zlmServerOffline(String mediaServerId) { @@ -1130,14 +1255,18 @@ } @Override - public void getSnap(String deviceId, String channelId, String fileName, ErrorCallback errorCallback) { + public void getSnap(String deviceId, String channelId, String fileName,boolean isSubStream, ErrorCallback errorCallback) { Device device = deviceService.getDevice(deviceId); if (device == null) { errorCallback.run(InviteErrorCode.ERROR_FOR_PARAMETER_ERROR.getCode(), InviteErrorCode.ERROR_FOR_PARAMETER_ERROR.getMsg(), null); return; } - - InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + InviteInfo inviteInfo; + if(device.isSwitchPrimarySubStream()){ + inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId,isSubStream); + }else { + inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + } if (inviteInfo != null) { if (inviteInfo.getStreamInfo() != null) { // 宸插瓨鍦ㄧ嚎鐩存帴鎴浘 @@ -1163,11 +1292,11 @@ } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); - play(newMediaServerItem, deviceId, channelId, (code, msg, data)->{ + play(newMediaServerItem, deviceId, channelId,isSubStream, (code, msg, data)->{ if (code == InviteErrorCode.SUCCESS.getCode()) { InviteInfo inviteInfoForPlay = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); if (inviteInfoForPlay != null && inviteInfoForPlay.getStreamInfo() != null) { - getSnap(deviceId, channelId, fileName, errorCallback); + getSnap(deviceId, channelId, fileName,isSubStream, errorCallback); }else { errorCallback.run(InviteErrorCode.FAIL.getCode(), InviteErrorCode.FAIL.getMsg(), null); } @@ -1177,4 +1306,17 @@ }); } + + /*======================璁惧涓诲瓙鐮佹祦閫昏緫START=========================*/ + public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, OnStreamChangedHookParam hookParam, String deviceId, String channelId,boolean isSubStream) { + String streamId = hookParam.getStream(); + List<OnStreamChangedHookParam.MediaTrack> tracks = hookParam.getTracks(); + StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, "rtp", streamId, tracks, null); + streamInfo.setDeviceID(deviceId); + streamInfo.setChannelId(channelId); + streamInfo.setSubStream(isSubStream); + return streamInfo; + } + /*======================璁惧涓诲瓙鐮佹祦閫昏緫END=========================*/ + } -- Gitblit v1.8.0