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/cmd/impl/SIPCommanderFroPlatform.java | 202 +++++++++++++++++++++++++++++++++++++++++--------- 1 files changed, 164 insertions(+), 38 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java old mode 100644 new mode 100755 index d223b8f..6a6bfcf --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java @@ -1,24 +1,32 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl; import com.alibaba.fastjson2.JSON; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.conf.DynamicTask; +import com.genersoft.iot.vmp.conf.UserSetting; +import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.gb28181.SipLayer; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; +import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.SIPSender; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider; import com.genersoft.iot.vmp.gb28181.utils.SipUtils; -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.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.service.bean.GPSMsgInfo; +import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo; import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.GitUtil; import gov.nist.javax.sip.message.MessageFactoryImpl; import gov.nist.javax.sip.message.SIPRequest; +import gov.nist.javax.sip.message.SIPResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -28,6 +36,7 @@ import org.springframework.util.ObjectUtils; import javax.sip.InvalidArgumentException; +import javax.sip.ResponseEvent; import javax.sip.SipException; import javax.sip.SipFactory; import javax.sip.header.CallIdHeader; @@ -56,13 +65,20 @@ private SipSubscribe sipSubscribe; @Autowired - private ZLMRTPServerFactory zlmrtpServerFactory; - - @Autowired private SipLayer sipLayer; @Autowired private SIPSender sipSender; + + @Autowired + private HookSubscribe subscribe; + + @Autowired + private UserSetting userSetting; + + + @Autowired + private VideoStreamSessionManager streamSession; @Autowired private DynamicTask dynamicTask; @@ -148,13 +164,13 @@ CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); - Request request = headerProviderPlatformProvider.createMessageRequest( - parentPlatform, - keepaliveXml.toString(), - SipUtils.getNewFromTag(), - SipUtils.getNewViaTag(), - callIdHeader); - sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, errorEvent, okEvent); + Request request = headerProviderPlatformProvider.createMessageRequest( + parentPlatform, + keepaliveXml.toString(), + SipUtils.getNewFromTag(), + SipUtils.getNewViaTag(), + callIdHeader); + sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, errorEvent, okEvent); return callIdHeader.getCallId(); } @@ -214,6 +230,7 @@ continue; }else { if (channel.getChannelId().length() != 20) { + catalogXml.append("</Item>\r\n"); logger.warn("[缂栧彿闀垮害寮傚父] {} 闀垮害閿欒锛岃浣跨敤20浣嶉暱搴︾殑鍥芥爣缂栧彿,褰撳墠闀垮害锛歿}", channel.getChannelId(), channel.getChannelId().length()); catalogXml.append("</Item>\r\n"); continue; @@ -336,12 +353,12 @@ }else { catalogXml.append("<Password></Password>\r\n"); } - if (!ObjectUtils.isEmpty(channel.getPTZType())) { - catalogXml.append("<PTZType>" + channel.getPTZType() + "</PTZType>\r\n"); + if (!ObjectUtils.isEmpty(channel.getPtzType())) { + catalogXml.append("<PTZType>" + channel.getPtzType() + "</PTZType>\r\n"); }else { catalogXml.append("<PTZType></PTZType>\r\n"); } - catalogXml.append("<Status>" + (channel.getStatus() == 1?"ON":"OFF") + "</Status>\r\n"); + catalogXml.append("<Status>" + (channel.isStatus() ?"ON":"OFF") + "</Status>\r\n"); catalogXml.append("<Longitude>" + (channel.getLongitudeWgs84() != 0? channel.getLongitudeWgs84():channel.getLongitude()) @@ -461,6 +478,7 @@ sipSender.transmitRequest(parentPlatform.getDeviceIp(), request); } + /** * 鍚戜笂绾у洖澶岲eviceStatus鏌ヨ淇℃伅 * @param parentPlatform 骞冲彴淇℃伅 @@ -469,11 +487,11 @@ * @return */ @Override - public void deviceStatusResponse(ParentPlatform parentPlatform,String channelId, String sn, String fromTag,int status) throws SipException, InvalidArgumentException, ParseException { + public void deviceStatusResponse(ParentPlatform parentPlatform,String channelId, String sn, String fromTag,boolean status) throws SipException, InvalidArgumentException, ParseException { if (parentPlatform == null) { return ; } - String statusStr = (status==1)?"ONLINE":"OFFLINE"; + String statusStr = (status)?"ONLINE":"OFFLINE"; String characterSet = parentPlatform.getCharacterSet(); StringBuffer deviceStatusXml = new StringBuffer(600); deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n") @@ -497,9 +515,7 @@ if (parentPlatform == null) { return; } - if (logger.isDebugEnabled()) { - logger.debug("[鍙戦�� 绉诲姩浣嶇疆璁㈤槄] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat()); - } + logger.info("[鍙戦�� 绉诲姩浣嶇疆璁㈤槄] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat()); String characterSet = parentPlatform.getCharacterSet(); StringBuffer deviceStatusXml = new StringBuffer(600); @@ -556,7 +572,7 @@ @Override public void sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo, Integer index) throws InvalidArgumentException, ParseException, NoSuchFieldException, SipException, IllegalAccessException { - if (parentPlatform == null || deviceChannels == null || deviceChannels.size() == 0 || subscribeInfo == null) { + if (parentPlatform == null || deviceChannels == null || deviceChannels.isEmpty() || subscribeInfo == null) { return; } if (index == null) { @@ -574,6 +590,8 @@ Integer finalIndex = index; String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channels, deviceChannels.size(), type, subscribeInfo); + System.out.println(catalogXmlContent); + logger.info("[鍙戦�丯OTIFY閫氱煡]绫诲瀷锛� {}锛屽彂閫佹暟閲忥細 {}", type, channels.size()); sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> { logger.error("鍙戦�丯OTIFY閫氱煡娑堟伅澶辫触銆傞敊璇細{} {}", eventResult.statusCode, eventResult.msg); }, (eventResult -> { @@ -597,21 +615,20 @@ SIPRequest notifyRequest = headerProviderPlatformProvider.createNotifyRequest(parentPlatform, catalogXmlContent, subscribeInfo); - sipSender.transmitRequest(parentPlatform.getDeviceIp(), notifyRequest); + sipSender.transmitRequest(parentPlatform.getDeviceIp(), notifyRequest, errorEvent, okEvent); } private String getCatalogXmlContentForCatalogAddOrUpdate(ParentPlatform parentPlatform, List<DeviceChannel> channels, int sumNum, String type, SubscribeInfo subscribeInfo) { StringBuffer catalogXml = new StringBuffer(600); - String characterSet = parentPlatform.getCharacterSet(); catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n") .append("<Notify>\r\n") .append("<CmdType>Catalog</CmdType>\r\n") .append("<SN>" + (int) ((Math.random() * 9 + 1) * 100000) + "</SN>\r\n") .append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n") - .append("<SumNum>1</SumNum>\r\n") + .append("<SumNum>"+ sumNum +"</SumNum>\r\n") .append("<DeviceList Num=\"" + channels.size() + "\">\r\n"); - if (channels.size() > 0) { + if (!channels.isEmpty()) { for (DeviceChannel channel : channels) { if (parentPlatform.getServerGBId().equals(channel.getParentId())) { channel.setParentId(parentPlatform.getDeviceGBId()); @@ -634,13 +651,15 @@ catalogXml.append("<Manufacturer>" + channel.getManufacture() + "</Manufacturer>\r\n") .append("<Secrecy>" + channel.getSecrecy() + "</Secrecy>\r\n") .append("<RegisterWay>" + channel.getRegisterWay() + "</RegisterWay>\r\n") - .append("<Status>" + (channel.getStatus() == 0 ? "OFF" : "ON") + "</Status>\r\n"); + .append("<Status>" + (channel.isStatus() ? "ON" : "OFF") + "</Status>\r\n"); if (channel.getChannelType() != 2) { // 涓氬姟鍒嗙粍/铏氭嫙缁勭粐/琛屾斂鍖哄垝 涓嶈缃互涓嬪睘鎬� catalogXml.append("<Model>" + channel.getModel() + "</Model>\r\n") .append("<Owner> " + channel.getOwner()+ "</Owner>\r\n") .append("<CivilCode>" + channel.getCivilCode() + "</CivilCode>\r\n") .append("<Address>" + channel.getAddress() + "</Address>\r\n"); + catalogXml.append("<Longitude>" + channel.getLongitude() + "</Longitude>\r\n"); + catalogXml.append("<Latitude>" + channel.getLatitude() + "</Latitude>\r\n"); } if (!"presence".equals(subscribeInfo.getEventType())) { catalogXml.append("<Event>" + type + "</Event>\r\n"); @@ -678,6 +697,7 @@ }else { channels = deviceChannels.subList(index, deviceChannels.size()); } + logger.info("[鍙戦�丯OTIFY閫氱煡]绫诲瀷锛� {}锛屽彂閫佹暟閲忥細 {}", type, channels.size()); Integer finalIndex = index; String catalogXmlContent = getCatalogXmlContentForCatalogOther(parentPlatform, channels, type); sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> { @@ -724,13 +744,14 @@ if ( parentPlatform ==null) { return ; } + logger.info("[鍥芥爣绾ц仈] 鍙戦�佸綍鍍忔暟鎹�氶亾锛� {}", recordInfo.getChannelId()); String characterSet = parentPlatform.getCharacterSet(); StringBuffer recordXml = new StringBuffer(600); recordXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n") .append("<Response>\r\n") .append("<CmdType>RecordInfo</CmdType>\r\n") .append("<SN>" +recordInfo.getSn() + "</SN>\r\n") - .append("<DeviceID>" + recordInfo.getDeviceId() + "</DeviceID>\r\n") + .append("<DeviceID>" + deviceChannel.getChannelId() + "</DeviceID>\r\n") .append("<SumNum>" + recordInfo.getSumNum() + "</SumNum>\r\n"); if (recordInfo.getRecordList() == null ) { recordXml.append("<RecordList Num=\"0\">\r\n"); @@ -740,7 +761,7 @@ for (RecordItem recordItem : recordInfo.getRecordList()) { recordXml.append("<Item>\r\n"); if (deviceChannel != null) { - recordXml.append("<DeviceID>" + recordItem.getDeviceId() + "</DeviceID>\r\n") + recordXml.append("<DeviceID>" + deviceChannel.getChannelId() + "</DeviceID>\r\n") .append("<Name>" + recordItem.getName() + "</Name>\r\n") .append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "</StartTime>\r\n") .append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getEndTime()) + "</EndTime>\r\n") @@ -760,12 +781,14 @@ recordXml.append("</RecordList>\r\n") .append("</Response>\r\n"); - + logger.info("[鍥芥爣绾ц仈] 鍙戦�佸綍鍍忔暟鎹�氶亾锛歿}, 鍐呭锛� {}", recordInfo.getChannelId(), recordXml); // callid CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport()); Request request = headerProviderPlatformProvider.createMessageRequest(parentPlatform, recordXml.toString(), fromTag, SipUtils.getNewViaTag(), callIdHeader); - sipSender.transmitRequest(parentPlatform.getDeviceIp(), request); + sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, null, eventResult -> { + logger.info("[鍥芥爣绾ц仈] 鍙戦�佸綍鍍忔暟鎹�氶亾锛歿}, 鍙戦�佹垚鍔�", recordInfo.getChannelId()); + }); } @@ -805,26 +828,129 @@ } @Override - public void streamByeCmd(ParentPlatform parentPlatform, SendRtpItem sendRtpItem) throws SipException, InvalidArgumentException, ParseException { + public synchronized void streamByeCmd(ParentPlatform platform, SendRtpItem sendRtpItem) throws SipException, InvalidArgumentException, ParseException { if (sendRtpItem == null ) { logger.info("[鍚戜笂绾у彂閫丅YE]锛� sendRtpItem 涓篘ULL"); return; } - if (parentPlatform == null) { + if (platform == null) { logger.info("[鍚戜笂绾у彂閫丅YE]锛� platform 涓篘ULL"); return; } - logger.info("[鍚戜笂绾у彂閫丅YE]锛� {}/{}", parentPlatform.getServerGBId(), sendRtpItem.getChannelId()); + logger.info("[鍚戜笂绾у彂閫丅YE]锛� {}/{}", platform.getServerGBId(), sendRtpItem.getChannelId()); String mediaServerId = sendRtpItem.getMediaServerId(); - MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); + MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId); if (mediaServerItem != null) { mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc()); - zlmrtpServerFactory.closeRtpServer(mediaServerItem, sendRtpItem.getStreamId()); + mediaServerService.closeRTPServer(mediaServerItem, sendRtpItem.getStream()); } - SIPRequest byeRequest = headerProviderPlatformProvider.createByeRequest(parentPlatform, sendRtpItem); + SIPRequest byeRequest = headerProviderPlatformProvider.createByeRequest(platform, sendRtpItem); if (byeRequest == null) { logger.warn("[鍚戜笂绾у彂閫乥ye]锛氭棤娉曞垱寤� byeRequest"); } - sipSender.transmitRequest(parentPlatform.getDeviceIp(),byeRequest); + sipSender.transmitRequest(platform.getDeviceIp(),byeRequest); + } + + @Override + public void streamByeCmd(ParentPlatform platform, String channelId, String stream, String callId, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException, SsrcTransactionNotFoundException { + SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(platform.getServerGBId(), channelId, callId, stream); + if (ssrcTransaction == null) { + throw new SsrcTransactionNotFoundException(platform.getServerGBId(), channelId, callId, stream); + } + + mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc()); + mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream()); + streamSession.remove(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream()); + + Request byteRequest = headerProviderPlatformProvider.createByteRequest(platform, channelId, ssrcTransaction.getSipTransactionInfo()); + sipSender.transmitRequest(sipLayer.getLocalIp(platform.getDeviceIp()), byteRequest, null, okEvent); + } + + @Override + public void broadcastResultCmd(ParentPlatform platform, DeviceChannel deviceChannel, String sn, boolean result, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException { + if (platform == null || deviceChannel == null) { + return; + } + String characterSet = platform.getCharacterSet(); + StringBuffer mediaStatusXml = new StringBuffer(200); + mediaStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n"); + mediaStatusXml.append("<Response>\r\n"); + mediaStatusXml.append("<CmdType>Broadcast</CmdType>\r\n"); + mediaStatusXml.append("<SN>" + sn + "</SN>\r\n"); + mediaStatusXml.append("<DeviceID>" + deviceChannel.getChannelId() + "</DeviceID>\r\n"); + mediaStatusXml.append("<Result>" + (result?"OK":"ERROR") + "</Result>\r\n"); + mediaStatusXml.append("</Response>\r\n"); + + CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(platform.getDeviceIp(), platform.getTransport()); + + SIPRequest messageRequest = (SIPRequest)headerProviderPlatformProvider.createMessageRequest(platform, mediaStatusXml.toString(), + SipUtils.getNewFromTag(), SipUtils.getNewViaTag(), callIdHeader); + + sipSender.transmitRequest(platform.getDeviceIp(),messageRequest, errorEvent, okEvent); + } + + @Override + public void broadcastInviteCmd(ParentPlatform platform, String channelId, MediaServer mediaServerItem, + SSRCInfo ssrcInfo, HookSubscribe.Event event, SipSubscribe.Event okEvent, + SipSubscribe.Event errorEvent) throws ParseException, SipException, InvalidArgumentException { + String stream = ssrcInfo.getStream(); + + if (platform == null) { + return; + } + + logger.info("{} 鍒嗛厤鐨刏LM涓�: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort()); + Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", stream, mediaServerItem.getId()); + subscribe.addSubscribe(hook, (hookData) -> { + if (event != null) { + event.response(hookData); + subscribe.removeSubscribe(hook); + } + }); + String sdpIp = mediaServerItem.getSdpIp(); + + StringBuffer content = new StringBuffer(200); + content.append("v=0\r\n"); + content.append("o=" + channelId + " 0 0 IN IP4 " + sdpIp + "\r\n"); + content.append("s=Play\r\n"); + content.append("c=IN IP4 " + sdpIp + "\r\n"); + content.append("t=0 0\r\n"); + + if ("TCP-PASSIVE".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { + content.append("m=audio " + ssrcInfo.getPort() + " TCP/RTP/AVP 8 96\r\n"); + } else if ("TCP-ACTIVE".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { + content.append("m=audio " + ssrcInfo.getPort() + " TCP/RTP/AVP 8 96\r\n"); + } else if ("UDP".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { + content.append("m=audio " + ssrcInfo.getPort() + " RTP/AVP 8 96\r\n"); + } + + content.append("a=recvonly\r\n"); + content.append("a=rtpmap:8 PCMA/8000\r\n"); + content.append("a=rtpmap:96 PS/90000\r\n"); + if ("TCP-PASSIVE".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { + content.append("a=setup:passive\r\n"); + content.append("a=connection:new\r\n"); + }else if ("TCP-ACTIVE".equalsIgnoreCase(userSetting.getBroadcastForPlatform())) { + content.append("a=setup:active\r\n"); + content.append("a=connection:new\r\n"); + } + + content.append("y=" + ssrcInfo.getSsrc() + "\r\n");//ssrc + CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(platform.getDeviceIp()), platform.getTransport()); + + Request request = headerProviderPlatformProvider.createInviteRequest(platform, channelId, + content.toString(), SipUtils.getNewViaTag(), SipUtils.getNewFromTag(), ssrcInfo.getSsrc(), + callIdHeader); + sipSender.transmitRequest(sipLayer.getLocalIp(platform.getDeviceIp()), request, (e -> { + streamSession.remove(platform.getServerGBId(), channelId, ssrcInfo.getStream()); + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + subscribe.removeSubscribe(hook); + errorEvent.response(e); + }), e -> { + ResponseEvent responseEvent = (ResponseEvent) e.event; + SIPResponse response = (SIPResponse) responseEvent.getResponse(); + streamSession.put(platform.getServerGBId(), channelId, callIdHeader.getCallId(), stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response, InviteSessionType.BROADCAST); + okEvent.response(e); + }); } } -- Gitblit v1.8.0