From 764d04b497356ba6bcbb75fd42b51eca750f7223 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期三, 29 五月 2024 15:02:51 +0800 Subject: [PATCH] 调整上级观看消息的发送 --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java | 194 ++++++++++++++++++++++++++++------------------- 1 files changed, 115 insertions(+), 79 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 59ff50c..de1a929 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -8,6 +8,7 @@ import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.UserSetting; +import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager; import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; @@ -18,17 +19,24 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor; import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; import com.genersoft.iot.vmp.gb28181.utils.SipUtils; +import com.genersoft.iot.vmp.media.bean.MediaInfo; +import com.genersoft.iot.vmp.media.bean.MediaServer; +import com.genersoft.iot.vmp.media.event.hook.Hook; +import com.genersoft.iot.vmp.media.event.hook.HookSubscribe; +import com.genersoft.iot.vmp.media.event.hook.HookType; +import com.genersoft.iot.vmp.media.service.IMediaServerService; +import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem; +import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; +import com.genersoft.iot.vmp.service.IPlayService; +import com.genersoft.iot.vmp.service.IStreamProxyService; +import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService; -import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory; -import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; -import com.genersoft.iot.vmp.media.zlm.dto.*; -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; import com.genersoft.iot.vmp.service.bean.MessageForPushChannel; import com.genersoft.iot.vmp.service.bean.SSRCInfo; +import com.genersoft.iot.vmp.service.redisMsg.RedisPushStreamResponseListener; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; @@ -43,6 +51,7 @@ import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; import javax.sdp.*; @@ -53,10 +62,10 @@ import javax.sip.message.Response; import java.text.ParseException; import java.time.Instant; -import java.util.HashMap; import java.util.Map; import java.util.Random; import java.util.Vector; +import java.util.*; /** * SIP鍛戒护绫诲瀷锛� INVITE璇锋眰 @@ -88,6 +97,9 @@ private IRedisRpcService redisRpcService; @Autowired + private RedisTemplate<Object, Object> redisTemplate; + + @Autowired private SSRCFactory ssrcFactory; @Autowired @@ -103,13 +115,10 @@ private AudioBroadcastManager audioBroadcastManager; @Autowired - private ZLMServerFactory zlmServerFactory; - - @Autowired private IMediaServerService mediaServerService; @Autowired - private ZlmHttpHookSubscribe zlmHttpHookSubscribe; + private HookSubscribe hookSubscribe; @Autowired private SIPProcessorObserver sipProcessorObserver; @@ -125,6 +134,9 @@ @Autowired private SendRtpPortManager sendRtpPortManager; + + @Autowired + private RedisPushStreamResponseListener redisPushStreamResponseListener; @Override @@ -184,7 +196,7 @@ GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId); PlatformCatalog catalog = storager.getCatalog(requesterId, channelId); - MediaServerItem mediaServerItem = null; + MediaServer mediaServerItem = null; StreamPushItem streamPushItem = null; StreamProxyItem proxyByAppAndStream = null; // 涓嶆槸閫氶亾鍙兘鏄洿鎾祦 @@ -323,8 +335,12 @@ return; } String username = sdp.getOrigin().getUsername(); - String addressStr = sdp.getConnection().getAddress(); - + String addressStr; + if(StringUtils.isEmpty(platform.getSendStreamIp())){ + addressStr = sdp.getConnection().getAddress(); + }else { + addressStr = platform.getSendStreamIp(); + } Device device = null; // 閫氳繃 channel 鍜� gbStream 鏄惁涓簄ull 鍊煎垽鏂潵婧愭槸鐩存挱娴佸悎閫傚浗鏍� @@ -367,8 +383,9 @@ } else { streamTypeStr = "UDP"; } - logger.info("[涓婄骇Invite] {}, 骞冲彴锛歿}锛� 閫氶亾锛歿}, 鏀舵祦鍦板潃锛歿}:{}锛屾敹娴佹柟寮忥細{}, ssrc锛歿}", sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc); - SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, + logger.info("[涓婄骇Invite] {}, 骞冲彴锛歿}锛� 閫氶亾锛歿}, 鏀舵祦鍦板潃锛歿}:{}锛屾敹娴佹柟寮忥細{}, ssrc锛歿}", + sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc); + SendRtpItem sendRtpItem = mediaServerService.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp()); if (tcpActive != null) { @@ -390,7 +407,7 @@ Long finalStopTime = stopTime; ErrorCallback<Object> hookEvent = (code, msg, data) -> { StreamInfo streamInfo = (StreamInfo)data; - MediaServerItem mediaServerItemInUSe = mediaServerService.getOne(streamInfo.getMediaServerId()); + MediaServer mediaServerItemInUSe = mediaServerService.getOne(streamInfo.getMediaServerId()); logger.info("[涓婄骇Invite]涓嬬骇宸茬粡寮�濮嬫帹娴併�� 鍥炲200OK(SDP)锛� {}/{}", streamInfo.getApp(), streamInfo.getStream()); // * 0 绛夊緟璁惧鎺ㄦ祦涓婃潵 // * 1 涓嬬骇宸茬粡鎺ㄦ祦锛岀瓑寰呬笂绾у钩鍙板洖澶峚ck @@ -447,42 +464,22 @@ responseSdpAck(request, content.toString(), platform); // tcp涓诲姩妯″紡锛屽洖澶峴dp鍚庡紑鍚洃鍚� if (sendRtpItem.isTcpActive()) { - MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - Map<String, Object> param = new HashMap<>(12); - param.put("vhost","__defaultVhost__"); - param.put("app",sendRtpItem.getApp()); - param.put("stream",sendRtpItem.getStream()); - param.put("ssrc", sendRtpItem.getSsrc()); - if (!sendRtpItem.isTcpActive()) { - param.put("dst_url",sendRtpItem.getIp()); - param.put("dst_port", sendRtpItem.getPort()); - } - String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; - param.put("is_udp", is_Udp); - param.put("src_port", localPort); - param.put("pt", sendRtpItem.getPt()); - param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0"); - param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0"); - if (!sendRtpItem.isTcp()) { - // 寮�鍚痳tcp淇濇椿 - param.put("udp_rtcp_timeout", sendRtpItem.isRtcp()? "1":"0"); - } - JSONObject startSendRtpStreamResult = zlmServerFactory.startSendRtpPassive(mediaInfo, param); - if (startSendRtpStreamResult != null) { - startSendRtpStreamHand(evt, sendRtpItem, null, startSendRtpStreamResult, param, callIdHeader); - } + MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + try { + mediaServerService.startSendRtpPassive(mediaServer, sendRtpItem, 5); + redisCatchStorage.sendPlatformStartPlayMsg(sendRtpItem, platform); + }catch (ControllerException e) {} } } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍥炲SdpAck", e); } }; ErrorCallback<Object> errorEvent = ((statusCode, msg, data) -> { + logger.info("[涓婄骇Invite] {}, 澶辫触, 骞冲彴锛歿}锛� 閫氶亾锛歿}, code锛� {}锛� msg锛泏}", sessionName, username, channelId, statusCode, msg); // 鏈煡閿欒銆傜洿鎺ヨ浆鍙戣澶囩偣鎾殑閿欒 try { - if (statusCode > 0) { - Response response = getMessageFactory().createResponse(statusCode, evt.getRequest()); - sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response); - } + Response response = getMessageFactory().createResponse(statusCode, evt.getRequest()); + sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response); } catch (ParseException | SipException e) { logger.error("鏈鐞嗙殑寮傚父 ", e); } @@ -493,7 +490,7 @@ String startTimeStr = DateUtil.urlFormatter.format(start); String endTimeStr = DateUtil.urlFormatter.format(end); String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr; - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0,false, false, device.getStreamModeForParam()); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0,false,!channel.getHasAudio(), false, device.getStreamModeForParam()); sendRtpItem.setStream(stream); // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -523,7 +520,7 @@ } sendRtpItem.setPlayType(InviteStreamType.DOWNLOAD); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam()); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,!channel.getHasAudio(), false, device.getStreamModeForParam()); sendRtpItem.setStream(ssrcInfo.getStream()); // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -585,18 +582,24 @@ sendRtpItem.setOnlyAudio(false); sendRtpItem.setStatus(0); sendRtpItem.setSessionName(sessionName); - + // 娓呯悊鍙兘瀛樺湪鐨勭紦瀛橀伩鍏嶇敤鍒版棫鐨勬暟鎹� + List<SendRtpItem> sendRtpItemList = redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, gbStream.getStream()); + if (!sendRtpItemList.isEmpty()) { + for (SendRtpItem rtpItem : sendRtpItemList) { + redisCatchStorage.deleteSendRTPServer(rtpItem); + } + } if ("push".equals(gbStream.getStreamType())) { sendRtpItem.setPlayType(InviteStreamType.PUSH); if (streamPushItem != null) { // 浠巖edis鏌ヨ鏄惁姝e湪鎺ユ敹杩欎釜鎺ㄦ祦 - OnStreamChangedHookParam pushListItem = redisCatchStorage.getPushListItem(gbStream.getApp(), gbStream.getStream()); + StreamPushItem pushListItem = redisCatchStorage.getPushListItem(gbStream.getApp(), gbStream.getStream()); if (pushListItem != null) { - sendRtpItem.setServerId(pushListItem.getSeverId()); + sendRtpItem.setServerId(pushListItem.getServerId()); sendRtpItem.setMediaServerId(pushListItem.getMediaServerId()); - StreamPushItem transform = streamPushService.transform(pushListItem); - transform.setSelf(userSetting.getServerId().equals(pushListItem.getSeverId())); + pushListItem.setSelf(userSetting.getServerId().equals(pushListItem.getServerId())); + redisCatchStorage.updateSendRTPSever(sendRtpItem); // 寮�濮嬫帹娴� sendPushStream(sendRtpItem, mediaServerItem, platform, request); }else { @@ -652,9 +655,11 @@ /** * 瀹夋帓鎺ㄦ祦 */ - private void sendProxyStream(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) { - Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); - if (streamReady != null && streamReady) { + private void sendProxyStream(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) { + MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); + + if (mediaInfo != null) { + // 鑷钩鍙板唴瀹� int localPort = sendRtpPortManager.getNextPort(mediaServerItem); if (localPort == 0) { @@ -679,11 +684,11 @@ } } - private void sendPushStream(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) { + private void sendPushStream(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) { // 鎺ㄦ祦 if (sendRtpItem.getServerId().equals(userSetting.getServerId())) { - Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); - if (streamReady != null && streamReady) { + MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); + if (mediaInfo != null ) { // 鑷钩鍙板唴瀹� int localPort = sendRtpPortManager.getNextPort(mediaServerItem); if (localPort == 0) { @@ -720,20 +725,19 @@ /** * 閫氱煡娴佷笂绾� */ - private void notifyProxyStreamOnline(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) { + private void notifyProxyStreamOnline(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) { // TODO 鎺у埗鍚敤浠ヤ娇璁惧涓婄嚎 logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍚敤娴佸悗寮�濮嬫帹娴�", sendRtpItem.getApp(), sendRtpItem.getStream()); // 鐩戝惉娴佷笂绾� - HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", mediaServerItem.getId()); - zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, hookParam) -> { - OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam; - logger.info("[涓婄骇鐐规挱]鎷夋祦浠g悊宸茬粡灏辩华锛� {}/{}", streamChangedHookParam.getApp(), streamChangedHookParam.getStream()); + Hook hook = Hook.getInstance(HookType.on_media_arrival, sendRtpItem.getApp(), sendRtpItem.getStream(), mediaServerItem.getId()); + hookSubscribe.addSubscribe(hook, (hookData)->{ + logger.info("[涓婄骇鐐规挱]鎷夋祦浠g悊宸茬粡灏辩华锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); dynamicTask.stop(sendRtpItem.getCallId()); sendProxyStream(sendRtpItem, mediaServerItem, platform, request); }); dynamicTask.startDelay(sendRtpItem.getCallId(), () -> { logger.info("[ app={}, stream={} ] 绛夊緟鎷夋祦浠g悊娴佽秴鏃�", sendRtpItem.getApp(), sendRtpItem.getStream()); - zlmHttpHookSubscribe.removeSubscribe(hookSubscribe); + hookSubscribe.removeSubscribe(hook); }, userSetting.getPlatformPlayTimeout()); boolean start = streamProxyService.start(sendRtpItem.getApp(), sendRtpItem.getStream()); if (!start) { @@ -742,7 +746,7 @@ } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 invite 閫氶亾鏈帹娴�: {}", e.getMessage()); } - zlmHttpHookSubscribe.removeSubscribe(hookSubscribe); + hookSubscribe.removeSubscribe(hook); dynamicTask.stop(sendRtpItem.getCallId()); } } @@ -750,7 +754,7 @@ /** * 閫氱煡娴佷笂绾� */ - private void notifyPushStreamOnline(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) { + private void notifyPushStreamOnline(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) { // 鍙戦�乺edis娑堟伅浠ヤ娇璁惧涓婄嚎锛屾祦涓婄嚎鍚庤 logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍙戦�乺edis淇℃伅鎺у埗璁惧寮�濮嬫帹娴�", sendRtpItem.getApp(), sendRtpItem.getStream()); MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1, @@ -759,6 +763,7 @@ redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel); // 璁剧疆瓒呮椂 dynamicTask.startDelay(sendRtpItem.getCallId(), () -> { + redisRpcService.stopWaitePushStreamOnline(sendRtpItem); logger.info("[ app={}, stream={} ] 绛夊緟璁惧寮�濮嬫帹娴佽秴鏃�", sendRtpItem.getApp(), sendRtpItem.getStream()); try { responseAck(request, Response.REQUEST_TIMEOUT); // 瓒呮椂 @@ -767,20 +772,35 @@ } }, userSetting.getPlatformPlayTimeout()); // - redisRpcService.waitePushStreamOnline(sendRtpItem, (sendRtpItemFromRedis) -> { + long key = redisRpcService.waitePushStreamOnline(sendRtpItem, (sendRtpItemKey) -> { dynamicTask.stop(sendRtpItem.getCallId()); + if (sendRtpItemKey == null) { + logger.warn("[绾ц仈鐐规挱] 绛夊緟鎺ㄦ祦寰楀埌缁撴灉鏈┖锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); + try { + responseAck(request, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("鏈鐞嗙殑寮傚父 ", e); + } + return; + } + SendRtpItem sendRtpItemFromRedis = (SendRtpItem)redisTemplate.opsForValue().get(sendRtpItemKey); + if (sendRtpItemFromRedis == null) { + logger.warn("[绾ц仈鐐规挱] 绛夊緟鎺ㄦ祦, 鏈壘鍒皉edis涓紦瀛樼殑鍙戞祦淇℃伅锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); + try { + responseAck(request, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("鏈鐞嗙殑寮傚父 ", e); + } + return; + } if (sendRtpItemFromRedis.getServerId().equals(userSetting.getServerId())) { - + logger.info("[绾ц仈鐐规挱] 绛夊緟鐨勬帹娴佸湪鏈钩鍙颁笂绾� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); int localPort = sendRtpPortManager.getNextPort(mediaServerItem); if (localPort == 0) { logger.warn("涓婄骇鐐规椂鍒涘缓sendRTPItem澶辫触锛屽彲鑳芥槸鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); try { responseAck(request, Response.BUSY_HERE); - } catch (SipException e) { - logger.error("鏈鐞嗙殑寮傚父 ", e); - } catch (InvalidArgumentException e) { - logger.error("鏈鐞嗙殑寮傚父 ", e); - } catch (ParseException e) { + } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("鏈鐞嗙殑寮傚父 ", e); } return; @@ -801,7 +821,20 @@ // 鍏朵粬骞冲彴鍐呭 otherWvpPushStream(sendRtpItemFromRedis, request, platform); } - + }); + // 娣诲姞鍥炲鐨勬嫆缁濇垨鑰呴敊璇殑閫氱煡 + // redis娑堟伅渚嬪锛� PUBLISH VM_MSG_STREAM_PUSH_RESPONSE '{"code":1,"msg":"澶辫触","app":"1","stream":"2"}' + redisPushStreamResponseListener.addEvent(sendRtpItem.getApp(), sendRtpItem.getStream(), response -> { + if (response.getCode() != 0) { + dynamicTask.stop(sendRtpItem.getCallId()); + redisRpcService.stopWaitePushStreamOnline(sendRtpItem); + redisRpcService.removeCallback(key); + try { + responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg()); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鐐规挱鍥炲: {}", e.getMessage()); + } + } }); } @@ -811,8 +844,11 @@ * 鏉ヨ嚜鍏朵粬wvp鐨勬帹娴� */ private void otherWvpPushStream(SendRtpItem sendRtpItem, SIPRequest request, ParentPlatform platform) { - logger.info("[绾ц仈鐐规挱]鐩存挱娴佹潵鑷叾浠栧钩鍙帮紝鍙戦�乺edis娑堟伅"); - sendRtpItem = redisRpcService.getSendRtpItem(sendRtpItem); + logger.info("[绾ц仈鐐规挱] 鏉ヨ嚜鍏朵粬wvp鐨勬帹娴� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); + sendRtpItem = redisRpcService.getSendRtpItem(sendRtpItem.getRedisKey()); + if (sendRtpItem == null) { + return; + } // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� sendRtpItem.setStatus(1); SIPResponse response = sendStreamAck(request, sendRtpItem, platform); @@ -968,7 +1004,7 @@ logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}, {}", requesterId, addressStr, port, gb28181Sdp.getSsrc(), mediaTransmissionTCP ? (tcpActive ? "TCP涓诲姩" : "TCP琚姩") : "UDP"); - MediaServerItem mediaServerItem = broadcastCatch.getMediaServerItem(); + MediaServer mediaServerItem = broadcastCatch.getMediaServerItem(); if (mediaServerItem == null) { logger.warn("鏈壘鍒拌闊冲枈璇濅娇鐢ㄧ殑zlm"); try { @@ -983,7 +1019,7 @@ mediaTransmissionTCP ? (tcpActive ? "TCP涓诲姩" : "TCP琚姩") : "UDP", sdp.getSessionName().getValue()); CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME); - SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, gb28181Sdp.getSsrc(), requesterId, + SendRtpItem sendRtpItem = mediaServerService.createSendRtpItem(mediaServerItem, addressStr, port, gb28181Sdp.getSsrc(), requesterId, device.getDeviceId(), broadcastCatch.getChannelId(), mediaTransmissionTCP, false); @@ -1017,7 +1053,7 @@ redisCatchStorage.updateSendRTPSever(sendRtpItem); - Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, broadcastCatch.getApp(), broadcastCatch.getStream()); + Boolean streamReady = mediaServerService.isStreamReady(mediaServerItem, broadcastCatch.getApp(), broadcastCatch.getStream()); if (streamReady) { sendOk(device, sendRtpItem, sdp, request, mediaServerItem, mediaTransmissionTCP, gb28181Sdp.getSsrc()); } else { @@ -1045,7 +1081,7 @@ } } - SIPResponse sendOk(Device device, SendRtpItem sendRtpItem, SessionDescription sdp, SIPRequest request, MediaServerItem mediaServerItem, boolean mediaTransmissionTCP, String ssrc) { + SIPResponse sendOk(Device device, SendRtpItem sendRtpItem, SessionDescription sdp, SIPRequest request, MediaServer mediaServerItem, boolean mediaTransmissionTCP, String ssrc) { SIPResponse sipResponse = null; try { sendRtpItem.setStatus(2); -- Gitblit v1.8.0