From 4c53b10737ac97861adaacd7cdc7571970860e9f Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Tue, 1 Mar 2022 17:42:22 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E5=AF=B9=E6=8E=A5=E4=B8=AD?= =?UTF-8?q?=E9=81=87=E5=88=B0=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/conf/SipPlatformRunner.java | 6 +- .../iot/vmp/gb28181/bean/SubscribeInfo.java | 23 ++- .../iot/vmp/gb28181/event/EventPublisher.java | 15 +- .../event/subscribe/catalog/CatalogEvent.java | 6 +- .../subscribe/catalog/CatalogEventLister.java | 12 +- .../transmit/SIPProcessorObserver.java | 6 +- .../cmd/ISIPCommanderForPlatform.java | 2 +- .../cmd/SIPRequestHeaderPlarformProvider.java | 26 +++- .../cmd/SIPRequestHeaderProvider.java | 4 +- .../cmd/impl/SIPCommanderFroPlatform.java | 132 +++++++++--------- .../request/SIPRequestProcessorParent.java | 17 ++- .../request/impl/InviteRequestProcessor.java | 4 +- .../impl/SubscribeRequestProcessor.java | 10 +- .../cmd/CatalogNotifyMessageHandler.java | 19 ++- .../query/cmd/CatalogQueryMessageHandler.java | 20 ++- .../vmp/media/zlm/ZLMHttpHookListener.java | 4 +- .../iot/vmp/media/zlm/dto/MediaItem.java | 12 ++ .../vmp/service/impl/GbStreamServiceImpl.java | 2 +- .../vmp/service/impl/RedisGPSMsgListener.java | 5 + .../service/impl/StreamPushServiceImpl.java | 15 +- .../iot/vmp/storager/dao/DeviceMapper.java | 2 +- .../storager/dao/PlatformChannelMapper.java | 8 +- .../gb28181/platform/PlatformController.java | 32 ++++- 23 files changed, 249 insertions(+), 133 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java b/src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java index 41b68cc0..b69bf68e 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java @@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.conf; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; +import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorager; @@ -30,7 +31,7 @@ public class SipPlatformRunner implements CommandLineRunner { private EventPublisher publisher; @Autowired - private ZLMRTPServerFactory zlmrtpServerFactory; + private ISIPCommanderForPlatform sipCommanderForPlatform; @Override @@ -57,6 +58,9 @@ public class SipPlatformRunner implements CommandLineRunner { parentPlatformCatch.setId(parentPlatform.getServerGBId()); redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch); + // 取消订阅 + sipCommanderForPlatform.unregister(parentPlatform, null, null); + Thread.sleep(500); // 发送平台未注册消息 publisher.platformNotRegisterEventPublish(parentPlatform.getServerGBId()); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java index 60418f2c..355edcc7 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java @@ -18,13 +18,16 @@ public class SubscribeInfo { this.fromTag = fromHeader.getTag(); ExpiresHeader expiresHeader = (ExpiresHeader)request.getHeader(ExpiresHeader.NAME); this.expires = expiresHeader.getExpires(); - this.event = ((EventHeader)request.getHeader(EventHeader.NAME)).getName(); + EventHeader eventHeader = (EventHeader)request.getHeader(EventHeader.NAME); + this.eventId = eventHeader.getEventId(); + this.eventType = eventHeader.getEventType(); } private String id; private int expires; private String callId; - private String event; + private String eventId; + private String eventType; private String fromTag; private String toTag; @@ -68,11 +71,19 @@ public class SubscribeInfo { this.fromTag = fromTag; } - public String getEvent() { - return event; + public String getEventId() { + return eventId; } - public void setEvent(String event) { - this.event = event; + public void setEventId(String eventId) { + this.eventId = eventId; + } + + public String getEventType() { + return eventType; + } + + public void setEventType(String eventType) { + this.eventType = eventType; } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java index 8c8565be..746467d6 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java @@ -101,14 +101,14 @@ public class EventPublisher { applicationEventPublisher.publishEvent(outEvent); } - @Async + public void catalogEventPublish(String platformId, DeviceChannel deviceChannel, String type) { List deviceChannelList = new ArrayList<>(); deviceChannelList.add(deviceChannel); catalogEventPublish(platformId, deviceChannelList, type); } - @Async + public void catalogEventPublish(String platformId, List deviceChannels, String type) { CatalogEvent outEvent = new CatalogEvent(this); List channels = new ArrayList<>(); @@ -130,8 +130,8 @@ public class EventPublisher { applicationEventPublisher.publishEvent(outEvent); } - @Async - public void catalogEventPublishForStream(String platformId, GbStream[] gbStreams, String type) { + + public void catalogEventPublishForStream(String platformId, List gbStreams, String type) { CatalogEvent outEvent = new CatalogEvent(this); outEvent.setGbStreams(gbStreams); outEvent.setType(type); @@ -139,9 +139,10 @@ public class EventPublisher { applicationEventPublisher.publishEvent(outEvent); } - @Async + public void catalogEventPublishForStream(String platformId, GbStream gbStream, String type) { - GbStream[] gbStreams = {gbStream}; - catalogEventPublishForStream(platformId, gbStreams, type); + List gbStreamList = new ArrayList<>(); + gbStreamList.add(gbStream); + catalogEventPublishForStream(platformId, gbStreamList, type); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEvent.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEvent.java index e343fec7..c035b808 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEvent.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEvent.java @@ -20,7 +20,7 @@ public class CatalogEvent extends ApplicationEvent { public static final String UPDATE = "UPDATE"; // 更新 private List deviceChannels; - private GbStream[] gbStreams; + private List gbStreams; private String type; private String platformId; @@ -48,11 +48,11 @@ public class CatalogEvent extends ApplicationEvent { this.platformId = platformId; } - public GbStream[] getGbStreams() { + public List getGbStreams() { return gbStreams; } - public void setGbStreams(GbStream[] gbStreams) { + public void setGbStreams(List gbStreams) { this.gbStreams = gbStreams; } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java index 3bb28887..fac55cb2 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java @@ -63,7 +63,11 @@ public class CatalogEventLister implements ApplicationListener { if (parentPlatform != null && !parentPlatform.isStatus())return; String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() + "_Catalog_" + event.getPlatformId(); subscribe = redisCatchStorage.getSubscribe(key); - if (subscribe == null) return; + + if (subscribe == null) { + logger.debug("发送订阅消息时发现订阅信息已经不存在"); + return; + } }else { // 获取所用订阅 List platforms = redisCatchStorage.getAllSubscribePlatform(); @@ -94,7 +98,7 @@ public class CatalogEventLister implements ApplicationListener { if (event.getDeviceChannels() != null) { deviceChannelList.addAll(event.getDeviceChannels()); } - if (event.getGbStreams() != null && event.getGbStreams().length > 0){ + if (event.getGbStreams() != null && event.getGbStreams().size() > 0){ for (GbStream gbStream : event.getGbStreams()) { DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), parentPlatform.getDeviceGBId()); deviceChannelList.add(deviceChannelByStream); @@ -134,7 +138,7 @@ public class CatalogEventLister implements ApplicationListener { if (event.getDeviceChannels() != null) { deviceChannelList.addAll(event.getDeviceChannels()); } - if (event.getGbStreams() != null && event.getGbStreams().length > 0){ + if (event.getGbStreams() != null && event.getGbStreams().size() > 0){ for (GbStream gbStream : event.getGbStreams()) { DeviceChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStream(gbStream, gbStream.getCatalogId(), parentPlatform.getDeviceGBId()); deviceChannelList.add(deviceChannelByStream); @@ -142,7 +146,7 @@ public class CatalogEventLister implements ApplicationListener { } if (deviceChannelList.size() > 0) { logger.info("[Catalog事件: {}]平台：{}，影响通道{}个", event.getType(), event.getPlatformId(), deviceChannelList.size()); - sipCommanderFroPlatform.sendNotifyForCatalogAddOrUpdate(event.getType(), parentPlatform, deviceChannelList, subscribe); + sipCommanderFroPlatform.sendNotifyForCatalogAddOrUpdate(event.getType(), parentPlatform, deviceChannelList, subscribe, null); } }else if (parentPlatformMap.keySet().size() > 0) { for (String gbId : parentPlatformMap.keySet()) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java index 71025c00..4732cc1c 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java @@ -73,6 +73,7 @@ public class SIPProcessorObserver implements ISIPProcessorObserver { @Override @Async public void processRequest(RequestEvent requestEvent) { + logger.debug("\n收到请求：\n{}", requestEvent.getRequest()); String method = requestEvent.getRequest().getMethod(); ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method); if (sipRequestProcessor == null) { @@ -90,9 +91,8 @@ public class SIPProcessorObserver implements ISIPProcessorObserver { @Override @Async public void processResponse(ResponseEvent responseEvent) { - logger.debug(responseEvent.getResponse().toString()); Response response = responseEvent.getResponse(); - logger.debug(responseEvent.getResponse().toString()); + logger.debug("\n收到响应：\n{}", responseEvent.getResponse()); int status = response.getStatusCode(); if (((status >= 200) && (status < 300)) || status == 401) { // Success! CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME); @@ -107,8 +107,8 @@ public class SIPProcessorObserver implements ISIPProcessorObserver { SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId()); if (subscribe != null) { SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent); - subscribe.response(eventResult); sipSubscribe.removeOkSubscribe(callIdHeader.getCallId()); + subscribe.response(eventResult); } } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java index 7325889f..feb32fcb 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java @@ -78,7 +78,7 @@ public interface ISIPCommanderForPlatform { * @param parentPlatform * @param deviceChannels */ - boolean sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List deviceChannels, SubscribeInfo subscribeInfo); + boolean sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List deviceChannels, SubscribeInfo subscribeInfo, Integer index); /** * 回复catalog事件-删除 diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java index 27125e1e..0addc576 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java @@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd; import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; +import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import gov.nist.javax.sip.message.MessageFactoryImpl; import org.springframework.beans.factory.annotation.Autowired; @@ -223,30 +224,31 @@ public class SIPRequestHeaderPlarformProvider { UserAgentHeader userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam); request.addHeader(userAgentHeader); - ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "MANSCDP+xml"); + ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml"); request.setContent(content, contentTypeHeader); return request; } - public Request createNotifyRequest(ParentPlatform parentPlatform, String content, String fromTag, String toTag, CallIdHeader callIdHeader) throws PeerUnavailableException, ParseException, InvalidArgumentException { + public Request createNotifyRequest(ParentPlatform parentPlatform, String content, CallIdHeader callIdHeader, String viaTag, SubscribeInfo subscribeInfo) throws PeerUnavailableException, ParseException, InvalidArgumentException { Request request = null; // sipuri SipURI requestURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerIP()+ ":" + parentPlatform.getServerPort()); // via ArrayList viaHeaders = new ArrayList(); ViaHeader viaHeader = sipFactory.createHeaderFactory().createViaHeader(parentPlatform.getDeviceIp(), Integer.parseInt(parentPlatform.getDevicePort()), - parentPlatform.getTransport(), null); + parentPlatform.getTransport(), viaTag); viaHeader.setRPort(); viaHeaders.add(viaHeader); // from SipURI fromSipURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getDeviceGBId(), parentPlatform.getDeviceIp() + ":" + parentPlatform.getDevicePort()); Address fromAddress = sipFactory.createAddressFactory().createAddress(fromSipURI); - FromHeader fromHeader = sipFactory.createHeaderFactory().createFromHeader(fromAddress, fromTag); + String tm = Long.toString(System.currentTimeMillis()); + FromHeader fromHeader = sipFactory.createHeaderFactory().createFromHeader(fromAddress, "fromtag" + tm); // to SipURI toSipURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerGBDomain()); Address toAddress = sipFactory.createAddressFactory().createAddress(toSipURI); - ToHeader toHeader = sipFactory.createHeaderFactory().createToHeader(toAddress, toTag); + ToHeader toHeader = sipFactory.createHeaderFactory().createToHeader(toAddress, subscribeInfo.getToTag()); // Forwards MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70); @@ -262,7 +264,19 @@ public class SIPRequestHeaderPlarformProvider { UserAgentHeader userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam); request.addHeader(userAgentHeader); - ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "MANSCDP+xml"); + EventHeader event = sipFactory.createHeaderFactory().createEventHeader(subscribeInfo.getEventType()); + event.setEventId(subscribeInfo.getEventId()); + request.addHeader(event); + + SubscriptionStateHeader active = sipFactory.createHeaderFactory().createSubscriptionStateHeader("active"); + request.setHeader(active); + + String sipAddress = sipConfig.getIp() + ":" + sipConfig.getPort(); + Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory() + .createSipURI(parentPlatform.getDeviceGBId(), sipAddress)); + request.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress)); + + ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml"); request.setContent(content, contentTypeHeader); return request; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java index ee237c0c..2da4c466 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java @@ -67,7 +67,7 @@ public class SIPRequestHeaderProvider { request = sipFactory.createMessageFactory().createRequest(requestURI, Request.MESSAGE, callIdHeader, cSeqHeader, fromHeader, toHeader, viaHeaders, maxForwards); - ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "MANSCDP+xml"); + ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml"); request.setContent(content, contentTypeHeader); return request; } @@ -217,7 +217,7 @@ public class SIPRequestHeaderProvider { EventHeader eventHeader = sipFactory.createHeaderFactory().createEventHeader(event); request.addHeader(eventHeader); - ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "MANSCDP+xml"); + ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml"); request.setContent(content, contentTypeHeader); return request; } 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 index a00da62f..a855c1db 100644 --- 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 @@ -32,21 +32,9 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { private final Logger logger = LoggerFactory.getLogger(SIPCommanderFroPlatform.class); - // @Autowired - // private SipConfig sipConfig; - - // @Autowired - // private SIPRequestHeaderProvider headerProvider; - @Autowired private SIPRequestHeaderPlarformProvider headerProviderPlarformProvider; - // @Autowired - // private VideoStreamSessionManager streamSession; - - // @Autowired - // private IVideoManagerStorager storager; - @Autowired private IRedisCatchStorage redisCatchStorage; @@ -173,6 +161,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { } private void transmitRequest(ParentPlatform parentPlatform, Request request, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws SipException { + logger.debug("\n发送消息：\n{}", request); if("TCP".equals(parentPlatform.getTransport())) { tcpSipProvider.sendRequest(request); @@ -223,7 +212,9 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { catalogXml.append("" + channel.getCivilCode() + "\r\n"); catalogXml.append("
" + channel.getAddress() + "
\r\n"); catalogXml.append("" + channel.getParental() + "\r\n"); - catalogXml.append("" + channel.getParentId() + "\r\n"); + if (channel.getParentId() != null) { + catalogXml.append("" + channel.getParentId() + "\r\n"); + } catalogXml.append("" + channel.getSecrecy() + "\r\n"); catalogXml.append("" + channel.getRegisterWay() + "\r\n"); catalogXml.append("" + (channel.getStatus() == 0?"OFF":"ON") + "\r\n"); @@ -358,7 +349,9 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { String tm = Long.toString(System.currentTimeMillis()); - Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, deviceStatusXml.toString(), subscribeInfo.getToTag(), subscribeInfo.getFromTag(), callIdHeader); + Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, + deviceStatusXml.toString(),callIdHeader, + "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), subscribeInfo); transmitRequest(parentPlatform, request); } catch (SipException | ParseException | InvalidArgumentException e) { @@ -369,63 +362,73 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { } @Override - public boolean sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List deviceChannels, SubscribeInfo subscribeInfo) { + public boolean sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List deviceChannels, SubscribeInfo subscribeInfo, Integer index) { if (parentPlatform == null || deviceChannels == null || deviceChannels.size() == 0 || subscribeInfo == null) { return false; } - for (DeviceChannel channel : deviceChannels) { - try { - StringBuffer catalogXml = new StringBuffer(600); - catalogXml.append("\r\n"); - catalogXml.append("\r\n"); - catalogXml.append("Catalog\r\n"); - catalogXml.append("" + (int) ((Math.random() * 9 + 1) * 100000) + "\r\n"); - catalogXml.append("" + deviceChannels.size() + "\r\n"); - catalogXml.append("\r\n"); - catalogXml.append("\r\n"); - catalogXml.append("" + channel.getChannelId() + "\r\n"); - catalogXml.append("" + type + "\r\n"); - catalogXml.append("" + channel.getName() + "\r\n"); - catalogXml.append("" + channel.getManufacture() + "\r\n"); - catalogXml.append("" + channel.getModel() + "\r\n"); - catalogXml.append("" + channel.getOwner() + "\r\n"); - catalogXml.append("" + channel.getCivilCode() + "\r\n"); - catalogXml.append("
" + channel.getAddress() + "
\r\n"); - catalogXml.append("" + channel.getParental() + "\r\n"); - catalogXml.append("" + channel.getParentId() + "\r\n"); - catalogXml.append("" + channel.getSecrecy() + "\r\n"); - catalogXml.append("" + channel.getRegisterWay() + "\r\n"); - catalogXml.append("" + (channel.getStatus() == 0 ? "OFF" : "ON") + "\r\n"); - catalogXml.append("" + channel.getLongitude() + "\r\n"); - catalogXml.append("" + channel.getLatitude() + "\r\n"); - catalogXml.append("" + channel.getIpAddress() + "\r\n"); - catalogXml.append("" + channel.getPort() + "\r\n"); - catalogXml.append("\r\n"); - catalogXml.append("" + channel.getPTZType() + "\r\n"); - catalogXml.append("\r\n"); - catalogXml.append("\r\n"); - catalogXml.append("\r\n"); - catalogXml.append("\r\n"); - - CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId() - : udpSipProvider.getNewCallId(); - callIdHeader.setCallId(subscribeInfo.getCallId()); - - String tm = Long.toString(System.currentTimeMillis()); + if (index == null) { + index = 0; + } - Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXml.toString(), subscribeInfo.getToTag(), subscribeInfo.getFromTag(), callIdHeader); - transmitRequest(parentPlatform, request); - Thread.sleep(10); - } catch (SipException | ParseException | InvalidArgumentException e) { - e.printStackTrace(); - return false; - } catch (InterruptedException e) { - e.printStackTrace(); + try { + if (index == deviceChannels.size() - 1) { + return true; } + Request request = getCatalogNotifyRequest(parentPlatform, deviceChannels.get(index), deviceChannels.size(), type, subscribeInfo); + index += 1; + Integer finalIndex = index; + transmitRequest(parentPlatform, request, null, (eventResult -> { + sendNotifyForCatalogAddOrUpdate(type, parentPlatform, deviceChannels, subscribeInfo, finalIndex); + })); + } catch (SipException | ParseException | InvalidArgumentException e) { + e.printStackTrace(); + return false; } return true; } + private Request getCatalogNotifyRequest(ParentPlatform parentPlatform, DeviceChannel channel, int size, String type, + SubscribeInfo subscribeInfo) throws ParseException, InvalidArgumentException, + PeerUnavailableException { + String catalogXmlContent = getCatalogXmlContent(parentPlatform, channel, size, type); + + CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId() + : udpSipProvider.getNewCallId(); + callIdHeader.setCallId(subscribeInfo.getCallId()); + Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXmlContent, + callIdHeader, "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), subscribeInfo); + return request; + } + + private String getCatalogXmlContent(ParentPlatform parentPlatform, DeviceChannel channel, int sumNum, String type) { + StringBuffer catalogXml = new StringBuffer(600); + catalogXml.append("\r\n"); + catalogXml.append("\r\n"); + catalogXml.append("Catalog\r\n"); + catalogXml.append("" + (int) ((Math.random() * 9 + 1) * 100000) + "\r\n"); + catalogXml.append("" + parentPlatform.getDeviceGBId() + "\r\n"); + catalogXml.append("" + sumNum + "\r\n"); + catalogXml.append("\r\n"); + catalogXml.append("\r\n"); + catalogXml.append("" + channel.getChannelId() + "\r\n"); + catalogXml.append("" + channel.getName() + "\r\n"); + catalogXml.append("" + channel.getManufacture() + "\r\n"); + catalogXml.append("" + channel.getModel() + "\r\n"); + catalogXml.append("0\r\n"); + catalogXml.append("CivilCode\r\n"); + catalogXml.append("
" + channel.getAddress() + "
\r\n"); + catalogXml.append("" + channel.getParental() + "\r\n"); + catalogXml.append("" + channel.getParentId() + "\r\n"); + catalogXml.append("" + channel.getSecrecy() + "\r\n"); + catalogXml.append("" + channel.getRegisterWay() + "\r\n"); + catalogXml.append("" + (channel.getStatus() == 0 ? "OFF" : "ON") + "\r\n"); + catalogXml.append("" + type + "\r\n"); + catalogXml.append("\r\n"); + catalogXml.append("\r\n"); + catalogXml.append("\r\n"); + return catalogXml.toString(); + } + @Override public boolean sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List deviceChannels, SubscribeInfo subscribeInfo) { if (parentPlatform == null @@ -442,6 +445,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { catalogXml.append("\r\n"); catalogXml.append("Catalog\r\n"); catalogXml.append("" + (int) ((Math.random() * 9 + 1) * 100000) + "\r\n"); + catalogXml.append("" + parentPlatform.getDeviceGBId() + "\r\n"); catalogXml.append("" + deviceChannels.size() + "\r\n"); catalogXml.append("\r\n"); catalogXml.append("\r\n"); @@ -457,7 +461,9 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { String tm = Long.toString(System.currentTimeMillis()); - Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXml.toString(), subscribeInfo.getToTag(), subscribeInfo.getFromTag(), callIdHeader); + Request request = headerProviderPlarformProvider.createNotifyRequest(parentPlatform, catalogXml.toString(), + callIdHeader, + "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), subscribeInfo); transmitRequest(parentPlatform, request); Thread.sleep(200); } catch (SipException | ParseException | InvalidArgumentException e) { 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 390565c7..34d45368 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,5 +1,6 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request; +import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import gov.nist.javax.sip.SipProviderImpl; import gov.nist.javax.sip.SipStackImpl; import gov.nist.javax.sip.message.SIPRequest; @@ -160,13 +161,18 @@ public abstract class SIPRequestProcessorParent { * @throws InvalidArgumentException * @throws ParseException */ - public void responseSdpAck(RequestEvent evt, String sdp) throws SipException, InvalidArgumentException, ParseException { + public void responseSdpAck(RequestEvent evt, String sdp, ParentPlatform platform) throws SipException, InvalidArgumentException, ParseException { Response response = getMessageFactory().createResponse(Response.OK, evt.getRequest()); SipFactory sipFactory = SipFactory.getInstance(); ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "SDP"); response.setContent(sdp, contentTypeHeader); + // 兼容国标中的使用编码@域名作为RequestURI的情况 SipURI sipURI = (SipURI)evt.getRequest().getRequestURI(); + if (sipURI.getPort() == -1) { + sipURI = sipFactory.createAddressFactory().createSipURI(platform.getServerGBId(), platform.getServerIP()+":"+platform.getServerPort()); + } + logger.debug("responseSdpAck SipURI: {}:{}", sipURI.getHost(), sipURI.getPort()); Address concatAddress = sipFactory.createAddressFactory().createAddress( sipFactory.createAddressFactory().createSipURI(sipURI.getUser(), sipURI.getHost()+":"+sipURI.getPort() @@ -183,13 +189,18 @@ public abstract class SIPRequestProcessorParent { * @throws InvalidArgumentException * @throws ParseException */ - public Response responseXmlAck(RequestEvent evt, String xml) throws SipException, InvalidArgumentException, ParseException { + public Response responseXmlAck(RequestEvent evt, String xml, ParentPlatform platform) throws SipException, InvalidArgumentException, ParseException { Response response = getMessageFactory().createResponse(Response.OK, evt.getRequest()); SipFactory sipFactory = SipFactory.getInstance(); - ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "MANSCDP+xml"); + ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml"); response.setContent(xml, contentTypeHeader); + // 兼容国标中的使用编码@域名作为RequestURI的情况 SipURI sipURI = (SipURI)evt.getRequest().getRequestURI(); + if (sipURI.getPort() == -1) { + sipURI = sipFactory.createAddressFactory().createSipURI(platform.getServerGBId(), platform.getServerIP()+":"+platform.getServerPort()); + } + logger.debug("responseXmlAck SipURI: {}:{}", sipURI.getHost(), sipURI.getPort()); Address concatAddress = sipFactory.createAddressFactory().createAddress( sipFactory.createAddressFactory().createSipURI(sipURI.getUser(), sipURI.getHost()+":"+sipURI.getPort() 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 ae2819ca..0c228c7f 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 @@ -252,7 +252,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements content.append("f=\r\n"); try { - responseSdpAck(evt, content.toString()); + responseSdpAck(evt, content.toString(), platform); } catch (SipException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { @@ -309,7 +309,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements content.append("f=\r\n"); try { - responseSdpAck(evt, content.toString()); + responseSdpAck(evt, content.toString(), platform); } catch (SipException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { 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 8d0bdd2f..a2c6cbfb 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 @@ -4,6 +4,7 @@ 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.gb28181.bean.CmdType; +import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo; import com.genersoft.iot.vmp.gb28181.task.GPSSubscribeTask; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; @@ -105,9 +106,6 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme logger.info("processRequest serverTransactionId is null."); } } - - - } catch (ParseException e) { e.printStackTrace(); } catch (SipException e) { @@ -155,7 +153,8 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme try { - Response response = responseXmlAck(evt, resultXml.toString()); + 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); @@ -196,7 +195,8 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme } try { - Response response = responseXmlAck(evt, resultXml.toString()); + 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); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/CatalogNotifyMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/CatalogNotifyMessageHandler.java index 4b99f902..21f36b3a 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/CatalogNotifyMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/CatalogNotifyMessageHandler.java @@ -72,6 +72,9 @@ public class CatalogNotifyMessageHandler extends SIPRequestProcessorParent imple List catalogs = storager.queryCatalogInPlatform(parentPlatform.getServerGBId()); if (catalogs.size() > 0) { for (PlatformCatalog catalog : catalogs) { + if (catalog.getParentId().equals(catalog.getPlatformId())) { + catalog.setParentId(parentPlatform.getDeviceGBId()); + } DeviceChannel deviceChannel = new DeviceChannel(); deviceChannel.setChannelId(catalog.getId()); deviceChannel.setName(catalog.getName()); @@ -83,29 +86,35 @@ public class CatalogNotifyMessageHandler extends SIPRequestProcessorParent imple deviceChannel.setParental(1); deviceChannel.setParentId(catalog.getParentId()); deviceChannel.setRegisterWay(1); - deviceChannel.setCivilCode(config.getDomain()); + deviceChannel.setCivilCode(config.getDomain().substring(0, config.getDomain().length() - 2)); deviceChannel.setModel("live"); deviceChannel.setOwner("wvp-pro"); deviceChannel.setSecrecy("0"); cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size); // 防止发送过快 - Thread.sleep(50); + Thread.sleep(100); } } // 回复级联的通道 if (channelReduces.size() > 0) { for (ChannelReduce channelReduce : channelReduces) { + if (channelReduce.getCatalogId().equals(parentPlatform.getServerGBId())) { + channelReduce.setCatalogId(parentPlatform.getDeviceGBId()); + } DeviceChannel deviceChannel = storager.queryChannel(channelReduce.getDeviceId(), channelReduce.getChannelId()); deviceChannel.setParental(0); deviceChannel.setParentId(channelReduce.getCatalogId()); cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size); // 防止发送过快 - Thread.sleep(50); + Thread.sleep(100); } } // 回复直播的通道 if (gbStreams.size() > 0) { for (GbStream gbStream : gbStreams) { + if (gbStream.getCatalogId().equals(parentPlatform.getServerGBId())) { + gbStream.setCatalogId(null); + } DeviceChannel deviceChannel = new DeviceChannel(); deviceChannel.setChannelId(gbStream.getGbId()); deviceChannel.setName(gbStream.getName()); @@ -116,14 +125,14 @@ public class CatalogNotifyMessageHandler extends SIPRequestProcessorParent imple deviceChannel.setStatus(gbStream.isStatus()?1:0); deviceChannel.setParentId(gbStream.getCatalogId()); deviceChannel.setRegisterWay(1); - deviceChannel.setCivilCode(config.getDomain()); + deviceChannel.setCivilCode(config.getDomain().substring(0, config.getDomain().length() - 2)); deviceChannel.setModel("live"); deviceChannel.setOwner("wvp-pro"); deviceChannel.setParental(0); deviceChannel.setSecrecy("0"); cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size); // 防止发送过快 - Thread.sleep(50); + Thread.sleep(100); } } if (size == 0) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java index c5c2961b..925a21e5 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java @@ -75,6 +75,9 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem int size = catalogs.size() + channelReduces.size() + gbStreams.size(); if (catalogs.size() > 0) { for (PlatformCatalog catalog : catalogs) { + if (catalog.getParentId().equals(parentPlatform.getServerGBId())) { + catalog.setParentId(parentPlatform.getDeviceGBId()); + } DeviceChannel deviceChannel = new DeviceChannel(); deviceChannel.setChannelId(catalog.getId()); deviceChannel.setName(catalog.getName()); @@ -86,30 +89,35 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem deviceChannel.setParental(1); deviceChannel.setParentId(catalog.getParentId()); deviceChannel.setRegisterWay(1); - deviceChannel.setCivilCode(config.getDomain()); + deviceChannel.setCivilCode(config.getDomain().substring(0, config.getDomain().length() - 2)); deviceChannel.setModel("live"); deviceChannel.setOwner("wvp-pro"); deviceChannel.setSecrecy("0"); cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size); // 防止发送过快 - Thread.sleep(50); + Thread.sleep(100); } } // 回复级联的通道 if (channelReduces.size() > 0) { for (ChannelReduce channelReduce : channelReduces) { + if (channelReduce.getCatalogId().equals(parentPlatform.getServerGBId())) { + channelReduce.setCatalogId(parentPlatform.getDeviceGBId()); + } DeviceChannel deviceChannel = storager.queryChannel(channelReduce.getDeviceId(), channelReduce.getChannelId()); - // TODO 目前暂时认为这里只用通道没有目录 deviceChannel.setParental(0); deviceChannel.setParentId(channelReduce.getCatalogId()); cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size); // 防止发送过快 - Thread.sleep(50); + Thread.sleep(100); } } // 回复直播的通道 if (gbStreams.size() > 0) { for (GbStream gbStream : gbStreams) { + if (gbStream.getCatalogId().equals(parentPlatform.getServerGBId())) { + gbStream.setCatalogId(parentPlatform.getDeviceGBId()); + } DeviceChannel deviceChannel = new DeviceChannel(); deviceChannel.setChannelId(gbStream.getGbId()); deviceChannel.setName(gbStream.getName()); @@ -120,7 +128,7 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem deviceChannel.setStatus(gbStream.isStatus()?1:0); deviceChannel.setParentId(gbStream.getCatalogId()); deviceChannel.setRegisterWay(1); - deviceChannel.setCivilCode(config.getDomain()); + deviceChannel.setCivilCode(config.getDomain().substring(0, config.getDomain().length() - 2)); deviceChannel.setModel("live"); deviceChannel.setOwner("wvp-pro"); deviceChannel.setParental(0); @@ -128,7 +136,7 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size); // 防止发送过快 - Thread.sleep(50); + Thread.sleep(100); } } if (size == 0) { 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 4f458135..424188d3 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 @@ -358,6 +358,8 @@ public class ZLMHttpHookListener { if (mediaServerItem != null){ if (regist) { StreamPushItem streamPushItem = null; + StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks); + item.setStreamInfo(streamInfoByAppAndStream); redisCatchStorage.addStream(mediaServerItem, type, app, streamId, item); if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal() || item.getOriginType() == OriginType.RTMP_PUSH.ordinal() @@ -375,7 +377,7 @@ public class ZLMHttpHookListener { } } if (gbStreams.size() > 0) { - eventPublisher.catalogEventPublishForStream(null, gbStreams.toArray(new GbStream[0]), CatalogEvent.ON); + eventPublisher.catalogEventPublishForStream(null, gbStreams, CatalogEvent.ON); } }else { diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaItem.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaItem.java index 4685d1fd..ad158ec7 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaItem.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaItem.java @@ -1,5 +1,7 @@ package com.genersoft.iot.vmp.media.zlm.dto; +import com.genersoft.iot.vmp.common.StreamInfo; + import java.util.List; public class MediaItem { @@ -281,6 +283,8 @@ public class MediaItem { } } + private StreamInfo streamInfo; + public String getApp() { return app; } @@ -402,4 +406,12 @@ public class MediaItem { public void setMediaServerId(String mediaServerId) { this.mediaServerId = mediaServerId; } + + public StreamInfo getStreamInfo() { + return streamInfo; + } + + public void setStreamInfo(StreamInfo streamInfo) { + this.streamInfo = streamInfo; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java index 2b4515f7..32a1305d 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java @@ -103,7 +103,7 @@ public class GbStreamServiceImpl implements IGbStreamService { deviceChannel.setStatus(gbStream.isStatus()?1:0); deviceChannel.setParentId(catalogId ==null?gbStream.getCatalogId():catalogId); deviceChannel.setRegisterWay(1); - deviceChannel.setCivilCode(sipConfig.getDomain()); + deviceChannel.setCivilCode(sipConfig.getDomain().substring(0, sipConfig.getDomain().length() - 2)); deviceChannel.setModel("live"); deviceChannel.setOwner("wvp-pro"); deviceChannel.setParental(0); diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisGPSMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisGPSMsgListener.java index e3bfcde7..92bcd5e8 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisGPSMsgListener.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisGPSMsgListener.java @@ -3,6 +3,8 @@ package com.genersoft.iot.vmp.service.impl; import com.alibaba.fastjson.JSON; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.connection.MessageListener; @@ -11,11 +13,14 @@ import org.springframework.stereotype.Component; @Component public class RedisGPSMsgListener implements MessageListener { + private final static Logger logger = LoggerFactory.getLogger(RedisGPSMsgListener.class); + @Autowired private IRedisCatchStorage redisCatchStorage; @Override public void onMessage(Message message, byte[] bytes) { + logger.debug("收到来自REDIS的GPS通知： {}", new String(message.getBody())); GPSMsgInfo gpsMsgInfo = JSON.parseObject(message.getBody(), GPSMsgInfo.class); redisCatchStorage.updateGpsMsgInfo(gpsMsgInfo); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java index 8a015d9d..a13dc29d 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java @@ -377,7 +377,7 @@ public class StreamPushServiceImpl implements IStreamPushService { if (streamPushItemsForPlatform.size() > 0) { List streamPushItemListFroPlatform = new ArrayList<>(); - Map> platformForEvent = new HashMap<>(); + Map> platformForEvent = new HashMap<>(); // 遍历存储结果，查找app+Stream->platformId+catalogId的对应关系，然后执行批量写入 for (StreamPushItem streamPushItem : streamPushItemsForPlatform) { List platFormInfoList = streamPushItemsForAll.get(streamPushItem.getApp() + streamPushItem.getStream()); @@ -390,16 +390,17 @@ public class StreamPushServiceImpl implements IStreamPushService { // 数组 platFormInfoArray 0 为平台ID。 1为目录ID streamPushItemForPlatform.setPlatformId(platFormInfoArray[0]); - List streamPushItemsInPlatform = platformForEvent.get(streamPushItem.getPlatformId()); - if (streamPushItemsInPlatform == null) { - streamPushItemsInPlatform = new ArrayList<>(); - platformForEvent.put(platFormInfoArray[0], streamPushItemsInPlatform); + List gbStreamList = platformForEvent.get(streamPushItem.getPlatformId()); + if (gbStreamList == null) { + gbStreamList = new ArrayList<>(); + platformForEvent.put(platFormInfoArray[0], gbStreamList); } // 为发送通知整理数据 + streamPushItemForPlatform.setName(streamPushItem.getName()); streamPushItemForPlatform.setApp(streamPushItem.getApp()); streamPushItemForPlatform.setStream(streamPushItem.getStream()); streamPushItemForPlatform.setGbId(streamPushItem.getGbId()); - streamPushItemsInPlatform.add(streamPushItemForPlatform); + gbStreamList.add(streamPushItemForPlatform); } if (platFormInfoArray.length > 1) { streamPushItemForPlatform.setCatalogId(platFormInfoArray[1]); @@ -416,7 +417,7 @@ public class StreamPushServiceImpl implements IStreamPushService { // 发送通知 for (String platformId : platformForEvent.keySet()) { eventPublisher.catalogEventPublishForStream( - platformId, platformForEvent.get(platformId).toArray(new GbStream[0]), CatalogEvent.ADD); + platformId, platformForEvent.get(platformId), CatalogEvent.ADD); } } } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java index ed2d8c89..28f79f18 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java @@ -74,7 +74,7 @@ public interface DeviceMapper { ", keepaliveTime='${keepaliveTime}'" + ", expires=${expires}" + ", charset='${charset}'" + - ", subscribeCycleForCatalog=#{subscribeCycleForCatalog}" + + ", subscribeCycleForCatalog=${subscribeCycleForCatalog}" + "WHERE deviceId='${deviceId}'"+ " "}) int update(Device device); diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java index c3c11a88..190a7b18 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java @@ -58,11 +58,15 @@ public interface PlatformChannelMapper { DeviceChannel queryChannelInParentPlatform(String platformId, String channelId); @Select("select dc.channelId as id, dc.name as name, pgc.platformId as platformId, pgc.catalogId as parentId, 0 as childrenCount, 1 as type " + - "from device_channel dc left join platform_gb_channel pgc on dc.deviceId = pgc.deviceId and dc.channelId = pgc.channelId " + + "from device_channel dc left join platform_gb_channel pgc on dc.id = pgc.deviceChannelId" + "where pgc.platformId=#{platformId} and pgc.catalogId=#{catalogId}") List queryChannelInParentPlatformAndCatalog(String platformId, String catalogId); - @Select("SELECT * FROM device WHERE deviceId = (SELECT deviceId FROM platform_gb_channel WHERE platformId='${platformId}' AND channelId='${channelId}')") + @Select("select d.*\n" + + "from platform_gb_channel pgc\n" + + " left join device_channel dc on dc.id = pgc.deviceChannelId\n" + + " left join device d on dc.deviceId = d.deviceId\n" + + "where dc.channelId = #{channelId} and pgc.platformId=#{platformId}") Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId); @Delete("