From 6fa5b37b962b05b3aa4b7bf019eb47c4cdbb5738 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期四, 05 一月 2023 11:07:06 +0800 Subject: [PATCH] Merge branch 'main' into main2 --- src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java | 30 +++ src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 178 ++++++++++++++++--- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java | 2 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/BroadcastResponseMessageHandler.java | 31 ++ src/main/java/com/genersoft/iot/vmp/gb28181/session/AudioBroadcastManager.java | 15 + src/main/resources/all-application.yml | 2 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java | 109 ++++------- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java | 1 src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java | 1 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java | 64 +++++-- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java | 19 +- src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java | 7 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 8 src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java | 10 + src/main/java/com/genersoft/iot/vmp/service/IPlayService.java | 12 + 15 files changed, 343 insertions(+), 146 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index 47e51ba..3db903b 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java @@ -69,6 +69,7 @@ public static final String SYSTEM_INFO_NET_PREFIX = "VMP_SYSTEM_INFO_NET_"; public static final String SYSTEM_INFO_DISK_PREFIX = "VMP_SYSTEM_INFO_DISK_"; + public static final String BROADCAST_WAITE_INVITE = "task_broadcast_waite_invite_"; diff --git a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java index 581ea6f..564fb64 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java @@ -43,6 +43,8 @@ private Boolean syncChannelOnDeviceOnline = Boolean.FALSE; + private Boolean pushStreamAfterAck = Boolean.FALSE; + private String serverId = "000000"; private String thirdPartyGBIdReg = "[\\s\\S]*"; @@ -206,4 +208,12 @@ public void setBroadcastForPlatform(String broadcastForPlatform) { this.broadcastForPlatform = broadcastForPlatform; } + + public Boolean getPushStreamAfterAck() { + return pushStreamAfterAck; + } + + public void setPushStreamAfterAck(Boolean pushStreamAfterAck) { + this.pushStreamAfterAck = pushStreamAfterAck; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/session/AudioBroadcastManager.java b/src/main/java/com/genersoft/iot/vmp/gb28181/session/AudioBroadcastManager.java index 072d0cb..5acbf8e 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/session/AudioBroadcastManager.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/session/AudioBroadcastManager.java @@ -83,4 +83,19 @@ return audioBroadcastCatch; } + + public List<AudioBroadcastCatch> get(String deviceId) { + List<AudioBroadcastCatch> audioBroadcastCatchList= new ArrayList<>(); + if (SipUtils.isFrontEnd(deviceId)) { + audioBroadcastCatchList.add(data.get(deviceId)); + }else { + for (String key : data.keySet()) { + if (key.startsWith(deviceId)) { + audioBroadcastCatchList.add(data.get(key)); + } + } + } + + return audioBroadcastCatchList; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java index ddb5169..f3dfa96 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java @@ -1,14 +1,10 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request; -import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; -import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.SIPSender; import com.genersoft.iot.vmp.gb28181.utils.SipUtils; -import gov.nist.javax.sip.SipProviderImpl; import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPResponse; -import gov.nist.javax.sip.stack.SIPServerTransactionImpl; import org.apache.commons.lang3.ArrayUtils; import org.dom4j.Document; import org.dom4j.DocumentException; @@ -17,14 +13,14 @@ 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.security.core.parameters.P; import javax.sip.*; import javax.sip.address.Address; import javax.sip.address.AddressFactory; import javax.sip.address.SipURI; -import javax.sip.header.*; +import javax.sip.header.ContentTypeHeader; +import javax.sip.header.ExpiresHeader; +import javax.sip.header.HeaderFactory; import javax.sip.message.MessageFactory; import javax.sip.message.Request; import javax.sip.message.Response; @@ -157,7 +153,10 @@ responseAckExtraParam.content = sdp; responseAckExtraParam.sipURI = sipURI; - return responseAck(request, Response.OK, null, responseAckExtraParam); + SIPResponse sipResponse = responseAck(request, Response.OK, null, responseAckExtraParam); + + + return sipResponse; } /** @@ -190,7 +189,8 @@ reader.setEncoding(charset); // 瀵规捣搴峰嚭鐜扮殑鏈浆涔夊瓧绗﹀仛澶勭悊銆� String[] destStrArray = new String[]{"<",">","&","'","""}; - char despChar = '&'; // 鎴栬鍙墿灞曞吋瀹瑰叾浠栧瓧绗� + // 鎴栬鍙墿灞曞吋瀹瑰叾浠栧瓧绗� + char despChar = '&'; byte destBye = (byte) despChar; List<Byte> result = new ArrayList<>(); byte[] rawContent = request.getRawContent(); @@ -220,4 +220,5 @@ return xml.getRootElement(); } + } 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 10b0daa..189976d 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,24 +1,18 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; -import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.conf.DynamicTask; -import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; -import com.genersoft.iot.vmp.gb28181.bean.AudioBroadcastCatch; -import com.genersoft.iot.vmp.gb28181.bean.Device; +import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; -import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; -import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; -import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; 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.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IMediaServerService; +import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg; import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; @@ -29,15 +23,15 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import javax.sip.InvalidArgumentException; import javax.sip.RequestEvent; -import javax.sip.SipException; 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.text.ParseException; +import java.util.HashMap; +import java.util.Map; /** * SIP鍛戒护绫诲瀷锛� ACK璇锋眰 @@ -46,7 +40,7 @@ @Component public class AckRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor { - private Logger logger = LoggerFactory.getLogger(AckRequestProcessor.class); + private final Logger logger = LoggerFactory.getLogger(AckRequestProcessor.class); private final String method = "ACK"; @Autowired @@ -74,31 +68,20 @@ private IMediaServerService mediaServerService; @Autowired - private ZlmHttpHookSubscribe subscribe; - - @Autowired private DynamicTask dynamicTask; - - @Autowired - private ISIPCommander cmder; - - @Autowired - private IDeviceService deviceService; - - @Autowired - private ISIPCommanderForPlatform commanderForPlatform; - - @Autowired - private AudioBroadcastManager audioBroadcastManager; @Autowired private RedisGbPlayMsgListener redisGbPlayMsgListener; + @Autowired + private UserSetting userSetting; + + @Autowired + private IPlayService playService; + /** * 澶勭悊 ACK璇锋眰 - * - * @param evt */ @Override public void process(RequestEvent evt) { @@ -106,44 +89,45 @@ String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser(); logger.info("[鏀跺埌ACK]锛� platformGbId->{}", platformGbId); - ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformGbId); - // 鍙栨秷璁剧疆鐨勮秴鏃朵换鍔� - dynamicTask.stop(callIdHeader.getCallId()); - String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser(); - SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId()); - if (sendRtpItem == null) { - logger.warn("[鏀跺埌ACK]锛氭湭鎵惧埌閫氶亾({})鐨勬帹娴佷俊鎭�", channelId); - return; - } - String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; - MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - logger.info("鏀跺埌ACK锛宺tp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}, RTCP={}", sendRtpItem.getStreamId(), - sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isRtcp()); - if (mediaInfo == null) { - RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance( - sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(), - sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(), - sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio()); - redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, json -> { - startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, json, callIdHeader); - }); - }else { - JSONObject startSendRtpStreamResult = zlmrtpServerFactory.startSendRtp(mediaInfo, sendRtpItem); - if (startSendRtpStreamResult != null) { - startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, callIdHeader); + if (userSetting.getPushStreamAfterAck()) { + ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformGbId); + // 鍙栨秷璁剧疆鐨勮秴鏃朵换鍔� + dynamicTask.stop(callIdHeader.getCallId()); + String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser(); + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId()); + if (sendRtpItem == null) { + logger.warn("[鏀跺埌ACK]锛氭湭鎵惧埌閫氶亾({})鐨勬帹娴佷俊鎭�", channelId); + return; + } + String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; + MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + logger.info("鏀跺埌ACK锛宺tp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}, RTCP={}", sendRtpItem.getStreamId(), + sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isRtcp()); + if (mediaInfo == null) { + RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance( + sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(), + sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(), + sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio()); + redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, json -> { + startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, json, callIdHeader); + }); + }else { + JSONObject startSendRtpStreamResult = zlmrtpServerFactory.startSendRtp(mediaInfo, sendRtpItem); + if (startSendRtpStreamResult != null) { + startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, callIdHeader); + } } } } - private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform, - JSONObject jsonObject, CallIdHeader callIdHeader) { + JSONObject jsonObject, Map<String, Object> param, CallIdHeader callIdHeader) { if (jsonObject == null) { logger.error("RTP鎺ㄦ祦澶辫触: 璇锋鏌LM鏈嶅姟"); } else if (jsonObject.getInteger("code") == 0) { logger.info("璋冪敤ZLM鎺ㄦ祦鎺ュ彛, 缁撴灉锛� {}", jsonObject); - logger.info("RTP鎺ㄦ祦鎴愬姛[ {}/{} ]锛寋}->{}:{}, " ,sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getIp(), sendRtpItem.getPort()); + logger.info("RTP鎺ㄦ祦鎴愬姛[ {}/{} ]锛寋}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port")); } else { - logger.error("RTP鎺ㄦ祦澶辫触: {}, 鍙傛暟锛歿}",jsonObject.getString("msg"), JSON.toJSONString(sendRtpItem)); + logger.error("RTP鎺ㄦ祦澶辫触: {}, 鍙傛暟锛歿}",jsonObject.getString("msg"), JSON.toJSONString(param)); if (sendRtpItem.isOnlyAudio()) { Device device = deviceService.getDevice(sendRtpItem.getDeviceId()); AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId()); @@ -152,17 +136,12 @@ cmder.streamByeCmd(device, sendRtpItem.getChannelId(), audioBroadcastCatch.getSipTransactionInfo(), null); } catch (SipException | ParseException | InvalidArgumentException | SsrcTransactionNotFoundException e) { - logger.error("[鍛戒护鍙戦�佸け璐 鍋滄璇煶鍠婅瘽: {}", e.getMessage()); + logger.error("[鍛戒护鍙戦�佸け璐 鍋滄璇煶瀵硅: {}", e.getMessage()); } - } - }else { - // 鍚戜笂绾у钩鍙� - try { - commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId()); - } catch (SipException | InvalidArgumentException | ParseException e) { - logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage()); } } } + } + } 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 6e188a5..e314962 100644 --- 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 @@ -1,6 +1,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.UserSetting; @@ -439,18 +440,23 @@ try { // 瓒呮椂鏈敹鍒癆ck搴旇鍥炲bye,褰撳墠绛夊緟鏃堕棿涓�10绉� - dynamicTask.startDelay(callIdHeader.getCallId(), () -> { - logger.info("Ack 绛夊緟瓒呮椂"); - mediaServerService.releaseSsrc(mediaServerItemInUSe.getId(), sendRtpItem.getSsrc()); - // 鍥炲bye - try { - cmderFroPlatform.streamByeCmd(platform, callIdHeader.getCallId()); - } catch (SipException | InvalidArgumentException | ParseException e) { - logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage()); - } - }, 60 * 1000); - responseSdpAck(request, content.toString(), platform); + if (userSetting.getPushStreamAfterAck()) { + dynamicTask.startDelay(callIdHeader.getCallId(), () -> { + logger.info("Ack 绛夊緟瓒呮椂"); + mediaServerService.releaseSsrc(mediaServerItemInUSe.getId(), sendRtpItem.getSsrc()); + // 鍥炲bye + try { + cmderFroPlatform.streamByeCmd(platform, callIdHeader.getCallId()); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage()); + } + }, 60 * 1000); + } + SIPResponse sipResponse = responseSdpAck(request, content.toString(), platform); + if (!userSetting.getPushStreamAfterAck()) { + playService.startPushStream(sendRtpItem, sipResponse, platform, request.getCallIdHeader()); + } } catch (SipException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { @@ -878,7 +884,11 @@ content.append("f=\r\n"); try { - return responseSdpAck(request, content.toString(), platform); + SIPResponse sipResponse = responseSdpAck(request, content.toString(), platform); + if (!userSetting.getPushStreamAfterAck()) { + playService.startPushStream(sendRtpItem, sipResponse, platform, request.getCallIdHeader()); + } + return sipResponse; } catch (SipException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { @@ -905,11 +915,14 @@ } if (device != null) { logger.info("鏀跺埌璁惧" + requesterId + "鐨勮闊冲箍鎾璉nvite璇锋眰"); - + String key = VideoManagerConstants.BROADCAST_WAITE_INVITE + device.getDeviceId() + audioBroadcastCatch.getChannelId(); + dynamicTask.stop(key); try { responseAck(request, Response.TRYING); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 invite BAD_REQUEST: {}", e.getMessage()); + playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId()); + return; } String contentString = new String(request.getRawContent()); // jainSip涓嶆敮鎸亂=瀛楁锛� 绉婚櫎绉婚櫎浠ヨВ鏋愩�� @@ -964,11 +977,14 @@ responseAck(request, Response.UNSUPPORTED_MEDIA_TYPE); // 涓嶆敮鎸佺殑鏍煎紡锛屽彂415 } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 invite 涓嶆敮鎸佺殑濯掍綋鏍煎紡: {}", e.getMessage()); + playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId()); + return; } return; } String addressStr = sdp.getOrigin().getAddress(); - logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}", requesterId, addressStr, port, ssrc); + logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}, {}", requesterId, addressStr, port, ssrc, + mediaTransmissionTCP ? (tcpActive? "TCP涓诲姩":"TCP琚姩") : "UDP"); MediaServerItem mediaServerItem = playService.getNewMediaServerItem(device); if (mediaServerItem == null) { @@ -977,6 +993,7 @@ responseAck(request, Response.BUSY_HERE); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 invite 鏈壘鍒板彲鐢ㄧ殑zlm: {}", e.getMessage()); + playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId()); } return; } @@ -990,13 +1007,12 @@ responseAck(request, Response.BUSY_HERE); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 invite 鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�: {}", e.getMessage()); + playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId()); + return; } return; } - sendRtpItem.setTcp(mediaTransmissionTCP); - if (tcpActive != null) { - sendRtpItem.setTcpActive(tcpActive); - } + String app = "broadcast"; String stream = device.getDeviceId() + "_" + audioBroadcastCatch.getChannelId(); @@ -1011,6 +1027,11 @@ sendRtpItem.setUsePs(false); sendRtpItem.setRtcp(false); sendRtpItem.setOnlyAudio(true); + sendRtpItem.setTcp(mediaTransmissionTCP); + if (tcpActive != null) { + sendRtpItem.setTcpActive(tcpActive); + } + redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -1023,11 +1044,13 @@ responseAck(request, Response.GONE); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 璇煶閫氳瘽 鍥炲410澶辫触锛� {}", e.getMessage()); + return; } playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId()); } } catch (SdpException e) { logger.error("[SDP瑙f瀽寮傚父]", e); + playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId()); } } else { logger.warn("鏉ヨ嚜鏃犳晥璁惧/骞冲彴鐨勮姹�"); @@ -1084,6 +1107,11 @@ audioBroadcastCatch.setSipTransactionInfoByRequset(sipResponse); audioBroadcastManager.update(audioBroadcastCatch); + // 寮�鍚彂娴侊紝澶у崕鍦ㄦ敹鍒�200OK鍚庡氨浼氬紑濮嬪缓绔嬭繛鎺� + if (!userSetting.getPushStreamAfterAck()) { + playService.startPushStream(sendRtpItem, sipResponse, parentPlatform, request.getCallIdHeader()); + } + } catch (SipException | InvalidArgumentException | ParseException | SdpParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 璇煶鍠婅瘽 鍥炲200OK锛圫DP锛�: {}", e.getMessage()); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/BroadcastResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/BroadcastResponseMessageHandler.java index 2cf2072..1c0dd58 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/BroadcastResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/BroadcastResponseMessageHandler.java @@ -1,5 +1,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd; +import com.genersoft.iot.vmp.common.VideoManagerConstants; +import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.gb28181.bean.AudioBroadcastCatch; import com.genersoft.iot.vmp.gb28181.bean.AudioBroadcastCatchStatus; import com.genersoft.iot.vmp.gb28181.bean.Device; @@ -9,6 +11,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler; +import com.genersoft.iot.vmp.service.IPlayService; import gov.nist.javax.sip.message.SIPRequest; import org.dom4j.Element; import org.slf4j.Logger; @@ -35,10 +38,13 @@ private ResponseMessageHandler responseMessageHandler; @Autowired - private DeferredResultHolder deferredResultHolder; + private DynamicTask dynamicTask; @Autowired private AudioBroadcastManager audioBroadcastManager; + + @Autowired + private IPlayService playService; @Override public void afterPropertiesSet() throws Exception { @@ -47,6 +53,8 @@ @Override public void handForDevice(RequestEvent evt, Device device, Element rootElement) { + + SIPRequest request = (SIPRequest) evt.getRequest(); try { String channelId = getText(rootElement, "DeviceID"); if (!audioBroadcastManager.exit(device.getDeviceId(), channelId)) { @@ -55,12 +63,23 @@ return; } String result = getText(rootElement, "Result"); - logger.info("鏀跺埌璇煶骞挎挱鐨勫洖澶� {}锛歿}/{}", result, device.getDeviceId(), channelId ); - AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(device.getDeviceId(), channelId); - audioBroadcastCatch.setStatus(AudioBroadcastCatchStatus.WaiteInvite); - audioBroadcastManager.update(audioBroadcastCatch); + logger.info("[璇煶骞挎挱]鍥炲锛歿}, {}/{}", result, device.getDeviceId(), channelId ); + // 鍥炲200 OK - responseAck((SIPRequest) evt.getRequest(), Response.OK); + responseAck(request, Response.OK); + if (result.equalsIgnoreCase("OK")) { + AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(device.getDeviceId(), channelId); + audioBroadcastCatch.setStatus(AudioBroadcastCatchStatus.WaiteInvite); + audioBroadcastManager.update(audioBroadcastCatch); + // 绛夊緟invite娑堟伅锛� 瓒呮椂鍒欑粨鏉� + String key = VideoManagerConstants.BROADCAST_WAITE_INVITE + device.getDeviceId() + channelId; + dynamicTask.startDelay(key, ()->{ + logger.info("[璇煶骞挎挱]绛夊緟invite娑堟伅瓒呮椂锛歿}/{}", device.getDeviceId(), channelId); + playService.stopAudioBroadcast(device.getDeviceId(), channelId); + }, 2000); + }else { + playService.stopAudioBroadcast(device.getDeviceId(), channelId); + } } catch (ParseException | SipException | InvalidArgumentException e) { logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 璇煶鍠婅瘽: {}", e.getMessage()); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java index 761481b..2129ee1 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java @@ -35,6 +35,7 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler { private Logger logger = LoggerFactory.getLogger(CatalogResponseMessageHandler.class); + private final String cmdType = "Catalog"; @Autowired diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java index 51ff7ad..fd4f391 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -323,7 +323,7 @@ }); if ("rtsp".equals(param.getSchema())){ - logger.info("on_stream_changed锛氭敞鍐�->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream()); + logger.info("娴佸彉鍖栵細娉ㄥ唽->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream()); if (param.isRegist()) { mediaServerService.addCount(param.getMediaServerId()); }else { @@ -383,10 +383,10 @@ } }else { - logger.info("[璇煶鍠婅瘽] 鎺ㄦ祦鎸囧悜鐨劼烽�氶亾{}鏈壘鍒�", channelId); + logger.info("[璇煶瀵硅] 鏈壘鍒伴�氶亾锛歿}", channelId); } - }else { - logger.info("[璇煶鍠婅瘽] 鎺ㄦ祦鎸囧悜鐨劼疯澶噞}鏈壘鍒�", deviceId); + }else{ + logger.info("[璇煶瀵硅] 鏈壘鍒拌澶囷細{}", deviceId); } }else { logger.info("[璇煶鍠婅瘽] 鎺ㄦ祦鏍煎紡鏈夎, 鏍煎紡涓猴細 broadcast/璁惧缂栧彿_閫氶亾缂栧彿 "); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java index 3eb61f5..fc7d90a 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java @@ -36,7 +36,7 @@ // 璁剧疆杩炴帴瓒呮椂鏃堕棿 httpClientBuilder.connectTimeout(5,TimeUnit.SECONDS); // 璁剧疆璇诲彇瓒呮椂鏃堕棿 - httpClientBuilder.readTimeout(5,TimeUnit.SECONDS); + httpClientBuilder.readTimeout(15,TimeUnit.SECONDS); // 璁剧疆杩炴帴姹� httpClientBuilder.connectionPool(new ConnectionPool(16, 5, TimeUnit.MINUTES)); if (logger.isDebugEnabled()) { diff --git a/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java b/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java index 3f7e13d..a24f4f4 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java +++ b/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java @@ -3,9 +3,7 @@ import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.exception.ServiceException; -import com.genersoft.iot.vmp.gb28181.bean.Device; -import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback; -import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo; +import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; @@ -15,11 +13,14 @@ import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult; import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.AudioBroadcastEvent; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; +import gov.nist.javax.sip.message.SIPResponse; import org.springframework.web.context.request.async.DeferredResult; import javax.sip.InvalidArgumentException; import javax.sip.SipException; +import javax.sip.header.CallIdHeader; import java.text.ParseException; +import java.util.Map; /** * 鐐规挱澶勭悊 @@ -64,4 +65,9 @@ void pauseRtp(String streamId) throws ServiceException, InvalidArgumentException, ParseException, SipException; void resumeRtp(String streamId) throws ServiceException, InvalidArgumentException, ParseException, SipException; + + void startPushStream(SendRtpItem sendRtpItem, SIPResponse sipResponse, ParentPlatform platform, CallIdHeader callIdHeader); + + void startSendRtpStreamHand(SendRtpItem sendRtpItem, ParentPlatform parentPlatform, + JSONObject jsonObject, Map<String, Object> param, CallIdHeader callIdHeader); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java index d25e537..eb208d3 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java @@ -3,12 +3,15 @@ import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.*; +import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask; import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask; import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler; +import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; +import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IDeviceChannelService; import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IMediaServerService; @@ -32,9 +35,7 @@ import javax.sip.SipException; import java.text.ParseException; import java.time.Instant; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; +import java.util.*; import java.util.concurrent.TimeUnit; /** @@ -88,6 +89,12 @@ @Autowired private IMediaServerService mediaServerService; + + @Autowired + private AudioBroadcastManager audioBroadcastManager; + + @Autowired + private ZLMRESTfulUtils zlmresTfulUtils; @Override public void online(Device device) { @@ -183,6 +190,23 @@ // 绉婚櫎璁㈤槄 removeCatalogSubscribe(device); removeMobilePositionSubscribe(device); + List<AudioBroadcastCatch> audioBroadcastCatches = audioBroadcastManager.get(deviceId); + if (audioBroadcastCatches.size() > 0) { + for (AudioBroadcastCatch audioBroadcastCatch : audioBroadcastCatches) { + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null); + if (sendRtpItem != null) { + redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null); + MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + Map<String, Object> param = new HashMap<>(); + param.put("vhost", "__defaultVhost__"); + param.put("app", sendRtpItem.getApp()); + param.put("stream", sendRtpItem.getStreamId()); + zlmresTfulUtils.stopSendRtp(mediaInfo, param); + } + + audioBroadcastManager.del(deviceId, audioBroadcastCatch.getChannelId()); + } + } } @Override 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 5e62981..8ce0ae8 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 @@ -24,16 +24,15 @@ import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; +import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaService; import com.genersoft.iot.vmp.service.IPlayService; -import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback; -import com.genersoft.iot.vmp.service.bean.PlayBackCallback; -import com.genersoft.iot.vmp.service.bean.PlayBackResult; -import com.genersoft.iot.vmp.service.bean.SSRCInfo; +import com.genersoft.iot.vmp.service.bean.*; +import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; @@ -42,6 +41,7 @@ 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.AudioBroadcastEvent; +import gov.nist.javax.sip.message.SIPResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -54,13 +54,11 @@ import javax.sip.InvalidArgumentException; import javax.sip.ResponseEvent; import javax.sip.SipException; +import javax.sip.header.CallIdHeader; import java.math.BigDecimal; import java.math.RoundingMode; import java.text.ParseException; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.UUID; +import java.util.*; @SuppressWarnings(value = {"rawtypes", "unchecked"}) @Service @@ -119,10 +117,19 @@ @Autowired private ZlmHttpHookSubscribe subscribe; + @Autowired + private ISIPCommanderForPlatform commanderForPlatform; + @Qualifier("taskExecutor") @Autowired private ThreadPoolTaskExecutor taskExecutor; + + @Autowired + private RedisGbPlayMsgListener redisGbPlayMsgListener; + + @Autowired + private ZlmHttpHookSubscribe hookSubscribe; @Override @@ -1024,8 +1031,20 @@ return false; } // 鏌ヨ閫氶亾浣跨敤鐘舵�� - if (audioBroadcastInUse(device, channelId)) { - return false; + if (audioBroadcastManager.exit(device.getDeviceId(), channelId)) { + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null); + if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) { + // 鏌ヨ娴佹槸鍚﹀瓨鍦紝涓嶅瓨鍦ㄥ垯璁や负鏄紓甯哥姸鎬� + MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStreamId()); + if (streamReady) { + logger.warn("璇煶骞挎挱宸茬粡寮�鍚細 {}", channelId); + event.call("璇煶骞挎挱宸茬粡寮�鍚�"); + return; + } else { + stopAudioBroadcast(device.getDeviceId(), channelId); + } + } } // 鍙戦�侀�氱煡 @@ -1063,28 +1082,31 @@ @Override public void stopAudioBroadcast(String deviceId, String channelId) { - AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(deviceId, channelId); - if (audioBroadcastCatch != null) { + List<AudioBroadcastCatch> audioBroadcastCatchList = new ArrayList<>(); + if (channelId == null) { + audioBroadcastCatchList.addAll(audioBroadcastManager.get(deviceId)); + }else { + audioBroadcastCatchList.add(audioBroadcastManager.get(deviceId, channelId)); + } + if (audioBroadcastCatchList.size() > 0) { + for (AudioBroadcastCatch audioBroadcastCatch : audioBroadcastCatchList) { + Device device = deviceService.getDevice(deviceId); + if (device == null || audioBroadcastCatch == null ) { + return; + } + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null); + if (sendRtpItem != null) { + redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null); + MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + Map<String, Object> param = new HashMap<>(); + param.put("vhost", "__defaultVhost__"); + param.put("app", sendRtpItem.getApp()); + param.put("stream", sendRtpItem.getStreamId()); + zlmresTfulUtils.stopSendRtp(mediaInfo, param); + } - Device device = deviceService.getDevice(deviceId); - if (device == null) { - return; + audioBroadcastManager.del(deviceId, channelId); } - SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null); - if (sendRtpItem != null) { - redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null); - MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - Map<String, Object> param = new HashMap<>(); - param.put("vhost", "__defaultVhost__"); - param.put("app", sendRtpItem.getApp()); - param.put("stream", sendRtpItem.getStreamId()); - zlmresTfulUtils.stopSendRtp(mediaInfo, param); - } - if (audioBroadcastCatch.isFromPlatform()) { - // TODO 鍚戜笂绾у彂閫丅YE缁撴潫璇煶鍠婅瘽 - } - - audioBroadcastManager.del(deviceId, channelId); } } @@ -1187,4 +1209,100 @@ Device device = storager.queryVideoDevice(streamInfo.getDeviceID()); cmder.playResumeCmd(device, streamInfo); } + + @Override + public void startPushStream(SendRtpItem sendRtpItem, SIPResponse sipResponse, ParentPlatform platform, CallIdHeader callIdHeader) { + + // 寮�濮嬪彂娴� + // 鍙栨秷璁剧疆鐨勮秴鏃朵换鍔� +// String channelId = request.getCallIdHeader().getCallId(); + + String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; + MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + logger.info("鏀跺埌ACK锛宺tp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}, RTCP={}", sendRtpItem.getStreamId(), + sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isRtcp()); + Map<String, Object> param = new HashMap<>(12); + param.put("vhost","__defaultVhost__"); + param.put("app",sendRtpItem.getApp()); + param.put("stream",sendRtpItem.getStreamId()); + param.put("ssrc", sendRtpItem.getSsrc()); + param.put("src_port", sendRtpItem.getLocalPort()); + param.put("pt", sendRtpItem.getPt()); + param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0"); + param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0"); + param.put("is_udp", is_Udp); + if (!sendRtpItem.isTcp()) { + // udp妯″紡涓嬪紑鍚痳tcp淇濇椿 + param.put("udp_rtcp_timeout", sendRtpItem.isRtcp()? "1":"0"); + } + + if (mediaInfo == null) { + RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance( + sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(), + sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(), + sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio()); + redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, json -> { + startSendRtpStreamHand(sendRtpItem, platform, json, param, callIdHeader); + }); + } else { + // 濡傛灉鏄潪涓ユ牸妯″紡锛岄渶瑕佸叧闂鍙e崰鐢� + JSONObject startSendRtpStreamResult = null; + if (sendRtpItem.getLocalPort() != 0) { + HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(sendRtpItem.getSsrc(), null, mediaInfo.getId()); + hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout); + if (zlmrtpServerFactory.releasePort(mediaInfo, sendRtpItem.getSsrc())) { + if (sendRtpItem.isTcpActive()) { + startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param); + }else { + param.put("dst_url", sendRtpItem.getIp()); + param.put("dst_port", sendRtpItem.getPort()); + startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); + } + } + }else { + if (sendRtpItem.isTcpActive()) { + startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param); + }else { + param.put("dst_url", sendRtpItem.getIp()); + param.put("dst_port", sendRtpItem.getPort()); + startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); + } + } + if (startSendRtpStreamResult != null) { + startSendRtpStreamHand(sendRtpItem, platform, startSendRtpStreamResult, param, callIdHeader); + } + } + } + + @Override + public void startSendRtpStreamHand(SendRtpItem sendRtpItem, ParentPlatform parentPlatform, + JSONObject jsonObject, Map<String, Object> param, CallIdHeader callIdHeader) { + if (jsonObject == null) { + logger.error("RTP鎺ㄦ祦澶辫触: 璇锋鏌LM鏈嶅姟"); + } else if (jsonObject.getInteger("code") == 0) { + logger.info("璋冪敤ZLM鎺ㄦ祦鎺ュ彛, 缁撴灉锛� {}", jsonObject); + logger.info("RTP鎺ㄦ祦鎴愬姛[ {}/{} ]锛寋}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port")); + } else { + logger.error("RTP鎺ㄦ祦澶辫触: {}, 鍙傛暟锛歿}",jsonObject.getString("msg"), JSON.toJSONString(param)); + if (sendRtpItem.isOnlyAudio()) { + Device device = deviceService.getDevice(sendRtpItem.getDeviceId()); + AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId()); + if (audioBroadcastCatch != null) { + try { + cmder.streamByeCmd(device, sendRtpItem.getChannelId(), audioBroadcastCatch.getSipTransactionInfo(), null); + } catch (SipException | ParseException | InvalidArgumentException | + SsrcTransactionNotFoundException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍋滄璇煶瀵硅: {}", e.getMessage()); + } + } + }else { + // 鍚戜笂绾у钩鍙� + try { + commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId()); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage()); + } + } + } + } } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java index 377b20f..e8997cb 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java @@ -6,7 +6,6 @@ import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem; -import com.genersoft.iot.vmp.service.IGbStreamService; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; @@ -89,12 +88,6 @@ @Autowired private PlatformGbStreamMapper platformGbStreamMapper; - - @Autowired - private IGbStreamService gbStreamService; - - @Autowired - private ParentPlatformMapper parentPlatformMapper; /** * 鏍规嵁璁惧ID鍒ゆ柇璁惧鏄惁瀛樺湪 diff --git a/src/main/resources/all-application.yml b/src/main/resources/all-application.yml index 92c2acb..1b4d0af 100644 --- a/src/main/resources/all-application.yml +++ b/src/main/resources/all-application.yml @@ -197,6 +197,8 @@ sync-channel-on-device-online: false # 鍥芥爣绾ц仈璇煶鍠婅瘽鍙戞祦妯″紡 * UDP:udp浼犺緭 TCP-ACTIVE锛歵cp涓诲姩妯″紡 TCP-PASSIVE锛歵cp琚姩妯″紡 broadcast-for-platform: UDP + # 鏀跺埌ack娑堟伅鍚庡紑濮嬪彂娴侊紝榛樿false锛� 鍥炲200ok鍚庣洿鎺ュ紑濮嬪彂娴� + push-stream-after-ack: false # 鍏抽棴鍦ㄧ嚎鏂囨。锛堢敓浜х幆澧冨缓璁叧闂級 springdoc: -- Gitblit v1.8.0