From 7e755f405dd1112a201b9dbde7e2d6cff8abc8f4 Mon Sep 17 00:00:00 2001 From: 648540858 <456panlinlin> Date: 星期五, 01 四月 2022 16:45:29 +0800 Subject: [PATCH] 支持国标移动位置订阅,收到新位置发送redis消息;支持通道redis消息拉起第三方推流;云台控制添加ControlPriority字段;处理sql的bug --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java | 83 ++++++++++++++++++++++++++--------------- 1 files changed, 53 insertions(+), 30 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java index a765b3a..2cd5b00 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java @@ -2,9 +2,10 @@ import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.DynamicTask; -import com.genersoft.iot.vmp.conf.UserSetup; +import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.CmdType; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; +import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder; import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo; import com.genersoft.iot.vmp.gb28181.task.GPSSubscribeTask; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; @@ -14,21 +15,20 @@ import com.genersoft.iot.vmp.gb28181.utils.SipUtils; import com.genersoft.iot.vmp.gb28181.utils.XmlUtil; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; -import com.genersoft.iot.vmp.storager.IVideoManagerStorager; +import com.genersoft.iot.vmp.storager.IVideoManagerStorage; +import gov.nist.javax.sip.SipProviderImpl; import org.dom4j.DocumentException; import org.dom4j.Element; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Component; -import javax.sip.InvalidArgumentException; -import javax.sip.RequestEvent; -import javax.sip.ServerTransaction; -import javax.sip.SipException; +import javax.sip.*; import javax.sip.header.ExpiresHeader; -import javax.sip.header.ToHeader; import javax.sip.message.Request; import javax.sip.message.Response; import java.text.ParseException; @@ -52,13 +52,27 @@ private ISIPCommanderForPlatform sipCommanderForPlatform; @Autowired - private IVideoManagerStorager storager; + private IVideoManagerStorage storager; + + @Lazy + @Autowired + @Qualifier(value="tcpSipProvider") + private SipProviderImpl tcpSipProvider; + + @Lazy + @Autowired + @Qualifier(value="udpSipProvider") + private SipProviderImpl udpSipProvider; @Autowired private DynamicTask dynamicTask; @Autowired - private UserSetup userSetup; + private UserSetting userSetting; + + + @Autowired + private SubscribeHolder subscribeHolder; @Override public void afterPropertiesSet() throws Exception { @@ -119,12 +133,21 @@ /** * 澶勭悊绉诲姩浣嶇疆璁㈤槄娑堟伅 */ - private void processNotifyMobilePosition(RequestEvent evt, Element rootElement) { + private void processNotifyMobilePosition(RequestEvent evt, Element rootElement) throws SipException { String platformId = SipUtils.getUserIdFromFromHeader(evt.getRequest()); String deviceID = XmlUtil.getText(rootElement, "DeviceID"); + ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId); SubscribeInfo subscribeInfo = new SubscribeInfo(evt, platformId); + if (evt.getServerTransaction() == null) { + ServerTransaction serverTransaction = platform.getTransport().equals("TCP") ? tcpSipProvider.getNewServerTransaction(evt.getRequest()) + : udpSipProvider.getNewServerTransaction(evt.getRequest()); + subscribeInfo.setTransaction(serverTransaction); + Dialog dialog = serverTransaction.getDialog(); + dialog.terminateOnBye(false); + subscribeInfo.setDialog(dialog); + } String sn = XmlUtil.getText(rootElement, "SN"); - String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() + "_MobilePosition_" + platformId; + String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_MobilePosition_" + platformId; logger.info("鎺ユ敹鍒皗}鐨凪obilePosition璁㈤槄", platformId); StringBuilder resultXml = new StringBuilder(200); resultXml.append("<?xml version=\"1.0\" ?>\r\n") @@ -136,25 +159,20 @@ .append("</Response>\r\n"); if (subscribeInfo.getExpires() > 0) { - if (redisCatchStorage.getSubscribe(key) != null) { + if (subscribeHolder.getMobilePositionSubscribe(platformId) != null) { dynamicTask.stop(key); } String interval = XmlUtil.getText(rootElement, "Interval"); // GPS涓婃姤鏃堕棿闂撮殧 - dynamicTask.startCron(key, new GPSSubscribeTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, sn, key), Integer.parseInt(interval)); - - redisCatchStorage.updateSubscribe(key, subscribeInfo); + dynamicTask.startCron(key, new GPSSubscribeTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, sn, key, subscribeHolder), Integer.parseInt(interval)); + subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo); }else if (subscribeInfo.getExpires() == 0) { dynamicTask.stop(key); - redisCatchStorage.delSubscribe(key); + subscribeHolder.removeMobilePositionSubscribe(platformId); } try { ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId); - Response response = responseXmlAck(evt, resultXml.toString(), parentPlatform); - ToHeader toHeader = (ToHeader)response.getHeader(ToHeader.NAME); - subscribeInfo.setToTag(toHeader.getTag()); - redisCatchStorage.updateSubscribe(key, subscribeInfo); - + responseXmlAck(evt, resultXml.toString(), parentPlatform); } catch (SipException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { @@ -168,12 +186,21 @@ } - private void processNotifyCatalogList(RequestEvent evt, Element rootElement) { + private void processNotifyCatalogList(RequestEvent evt, Element rootElement) throws SipException { String platformId = SipUtils.getUserIdFromFromHeader(evt.getRequest()); String deviceID = XmlUtil.getText(rootElement, "DeviceID"); + ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId); SubscribeInfo subscribeInfo = new SubscribeInfo(evt, platformId); + if (evt.getServerTransaction() == null) { + ServerTransaction serverTransaction = platform.getTransport().equals("TCP") ? tcpSipProvider.getNewServerTransaction(evt.getRequest()) + : udpSipProvider.getNewServerTransaction(evt.getRequest()); + subscribeInfo.setTransaction(serverTransaction); + Dialog dialog = serverTransaction.getDialog(); + dialog.terminateOnBye(false); + subscribeInfo.setDialog(dialog); + } String sn = XmlUtil.getText(rootElement, "SN"); - String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() + "_Catalog_" + platformId; + String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_Catalog_" + platformId; logger.info("鎺ユ敹鍒皗}鐨凜atalog璁㈤槄", platformId); StringBuilder resultXml = new StringBuilder(200); resultXml.append("<?xml version=\"1.0\" ?>\r\n") @@ -185,18 +212,14 @@ .append("</Response>\r\n"); if (subscribeInfo.getExpires() > 0) { - redisCatchStorage.updateSubscribe(key, subscribeInfo); + subscribeHolder.putCatalogSubscribe(platformId, subscribeInfo); }else if (subscribeInfo.getExpires() == 0) { - redisCatchStorage.delSubscribe(key); + subscribeHolder.removeCatalogSubscribe(platformId); } try { ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId); - Response response = responseXmlAck(evt, resultXml.toString(), parentPlatform); - ToHeader toHeader = (ToHeader)response.getHeader(ToHeader.NAME); - subscribeInfo.setToTag(toHeader.getTag()); - redisCatchStorage.updateSubscribe(key, subscribeInfo); - + responseXmlAck(evt, resultXml.toString(), parentPlatform); } catch (SipException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { -- Gitblit v1.8.0