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/AckRequestProcessor.java | 165 ++++++++++++++++++++++++++++++++---------------------- 1 files changed, 97 insertions(+), 68 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java index 98ae86f..d4df0ae 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java @@ -1,38 +1,44 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; -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.conf.exception.ControllerException; +import com.genersoft.iot.vmp.gb28181.bean.Device; +import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; 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.media.zlm.ZLMRTPServerFactory; -import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.service.IMediaServerService; +import com.genersoft.iot.vmp.media.bean.MediaServer; +import com.genersoft.iot.vmp.media.service.IMediaServerService; +import com.genersoft.iot.vmp.service.IDeviceService; +import com.genersoft.iot.vmp.service.IPlayService; +import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import com.genersoft.iot.vmp.storager.IVideoManagerStorage; +import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import javax.sip.Dialog; -import javax.sip.DialogState; import javax.sip.RequestEvent; import javax.sip.address.SipURI; +import javax.sip.header.CallIdHeader; import javax.sip.header.FromHeader; import javax.sip.header.HeaderAddress; import javax.sip.header.ToHeader; -import java.util.HashMap; -import java.util.Map; /** * SIP鍛戒护绫诲瀷锛� ACK璇锋眰 + * @author lin */ @Component public class AckRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor { - private Logger logger = LoggerFactory.getLogger(AckRequestProcessor.class); - private String method = "ACK"; + private final Logger logger = LoggerFactory.getLogger(AckRequestProcessor.class); + private final String method = "ACK"; @Autowired private SIPProcessorObserver sipProcessorObserver; @@ -45,79 +51,102 @@ @Autowired private IRedisCatchStorage redisCatchStorage; + @Autowired + private IRedisRpcService redisRpcService; @Autowired - private ZLMRTPServerFactory zlmrtpServerFactory; + private UserSetting userSetting; + + @Autowired + private IVideoManagerStorage storager; + + @Autowired + private IDeviceService deviceService; @Autowired private IMediaServerService mediaServerService; + @Autowired + private DynamicTask dynamicTask; + + @Autowired + private IPlayService playService; + /** * 澶勭悊 ACK璇锋眰 - * - * @param evt */ @Override public void process(RequestEvent evt) { - Dialog dialog = evt.getDialog(); - if (dialog == null) return; - if (dialog.getState()== DialogState.CONFIRMED) { - String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser(); - String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser(); - SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, channelId); - String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; - String deviceId = sendRtpItem.getDeviceId(); - StreamInfo streamInfo = null; - if (deviceId == null) { - streamInfo = new StreamInfo(); - streamInfo.setApp(sendRtpItem.getApp()); - streamInfo.setStream(sendRtpItem.getStreamId()); - }else { - streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); - sendRtpItem.setStreamId(streamInfo.getStream()); - streamInfo.setApp("rtp"); - } + CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME); + dynamicTask.stop(callIdHeader.getCallId()); + String fromUserId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser(); + String toUserId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser(); + logger.info("[鏀跺埌ACK]锛� 鏉ヨ嚜->{}", fromUserId); + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId()); + if (sendRtpItem == null) { + logger.warn("[鏀跺埌ACK]锛氭湭鎵惧埌鏉ヨ嚜{}锛宑allId: {}", fromUserId, callIdHeader.getCallId()); + return; + } + // tcp涓诲姩鏃讹紝姝ゆ椂鏄骇鑱斾笅绾у钩鍙帮紝鍦ㄥ洖澶�200ok鏃讹紝鏈湴宸茬粡璇锋眰zlm寮�鍚洃鍚紝璺宠繃涓嬮潰姝ラ + if (sendRtpItem.isTcpActive()) { + logger.info("鏀跺埌ACK锛宺tp/{} TCP涓诲姩鏂瑰紡鍚庣画澶勭悊", sendRtpItem.getStream()); + return; + } + MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + logger.info("鏀跺埌ACK锛宺tp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}, 鍗忚:{}", + sendRtpItem.getStream(), + sendRtpItem.getIp(), + sendRtpItem.getPort(), + sendRtpItem.getSsrc(), + sendRtpItem.isTcp()?(sendRtpItem.isTcpActive()?"TCP涓诲姩":"TCP琚姩"):"UDP" + ); + ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(fromUserId); - redisCatchStorage.updateSendRTPSever(sendRtpItem); - logger.info(platformGbId); - logger.info(channelId); - Map<String, Object> param = new HashMap<>(); - param.put("vhost","__defaultVhost__"); - param.put("app",streamInfo.getApp()); - param.put("stream",streamInfo.getStream()); - param.put("ssrc", sendRtpItem.getSsrc()); - param.put("dst_url",sendRtpItem.getIp()); - param.put("dst_port", sendRtpItem.getPort()); - param.put("is_udp", is_Udp); - //param.put ("src_port", sendRtpItem.getLocalPort()); - // 璁惧鎺ㄦ祦鏌ヨ锛屾垚鍔熷悗鎵嶈兘杞帹 - boolean rtpPushed = false; - long startTime = System.currentTimeMillis(); - while (!rtpPushed) { - try { - if (System.currentTimeMillis() - startTime < 30 * 1000) { - MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStream())) { - rtpPushed = true; - logger.info("宸茶幏鍙栬澶囨帹娴乕{}/{}]锛屽紑濮嬪悜涓婄骇鎺ㄦ祦[{}:{}]", - streamInfo.getApp() ,streamInfo.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); - zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); - } else { - logger.info("绛夊緟璁惧鎺ㄦ祦[{}/{}].......", - streamInfo.getApp() ,streamInfo.getStream()); - Thread.sleep(1000); - continue; - } - } else { - rtpPushed = true; - logger.info("璁惧鎺ㄦ祦[{}/{}]瓒呮椂锛岀粓姝㈠悜涓婄骇鎺ㄦ祦", - streamInfo.getApp() ,streamInfo.getStream()); - } - } catch (InterruptedException e) { - e.printStackTrace(); + if (parentPlatform != null) { + if (!userSetting.getServerId().equals(sendRtpItem.getServerId())) { + WVPResult wvpResult = redisRpcService.startSendRtp(sendRtpItem.getRedisKey(), sendRtpItem); + if (wvpResult.getCode() == 0) { + redisCatchStorage.sendPlatformStartPlayMsg(sendRtpItem, parentPlatform); } + } else { + try { + if (sendRtpItem.isTcpActive()) { + mediaServerService.startSendRtpPassive(mediaInfo,sendRtpItem, null); + } else { + mediaServerService.startSendRtp(mediaInfo, sendRtpItem); + } + redisCatchStorage.sendPlatformStartPlayMsg(sendRtpItem, parentPlatform); + }catch (ControllerException e) { + logger.error("RTP鎺ㄦ祦澶辫触: {}", e.getMessage()); + playService.startSendRtpStreamFailHand(sendRtpItem, parentPlatform, callIdHeader); + } + } + }else { + Device device = deviceService.getDevice(fromUserId); + if (device == null) { + logger.warn("[鏀跺埌ACK]锛氭潵鑷獅}锛岀洰鏍囦负({})鐨勬帹娴佷俊鎭负鎵惧埌娴佷綋鏈嶅姟[{}]淇℃伅",fromUserId, toUserId, sendRtpItem.getMediaServerId()); + return; + } + // 璁剧疆涓烘敹鍒癆CK鍚庡彂閫佽闊崇殑璁惧宸茬粡鍦ㄥ彂閫�200OK寮�濮嬪彂娴佷簡 + if (!device.isBroadcastPushAfterAck()) { + return; + } + if (mediaInfo == null) { + logger.warn("[鏀跺埌ACK]锛氭潵鑷獅}锛岀洰鏍囦负({})鐨勬帹娴佷俊鎭负鎵惧埌娴佷綋鏈嶅姟[{}]淇℃伅",fromUserId, toUserId, sendRtpItem.getMediaServerId()); + return; + } + try { + if (sendRtpItem.isTcpActive()) { + mediaServerService.startSendRtpPassive(mediaInfo, sendRtpItem, null); + } else { + mediaServerService.startSendRtp(mediaInfo, sendRtpItem); + } + }catch (ControllerException e) { + logger.error("RTP鎺ㄦ祦澶辫触: {}", e.getMessage()); + playService.startSendRtpStreamFailHand(sendRtpItem, null, callIdHeader); } } } + } -- Gitblit v1.8.0