Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'wvp-28181-2.0' into Zafu-Dev-1127

Browse Source
This commit is contained in:
648540858 2022-11-29 15:18:56 +08:00 committed by GitHub
commit ddb2e3ea71
17 changed files with 476 additions and 563 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
sql/mysql.sql
View File

@ -1,4 +1,4 @@
-- MySQL dump 10.13 Distrib 8.0.30, for Linux (x86_64) -- MySQL dump 10.13 Distrib 8.0.31, for Linux (x86_64)
-- --
-- Host: 127.0.0.1 Database: wvp -- Host: 127.0.0.1 Database: wvp
-- ------------------------------------------------------ -- ------------------------------------------------------
@ -34,13 +34,13 @@ CREATE TABLE `device` (
`online` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL, `online` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
`registerTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL, `registerTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
`keepaliveTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL, `keepaliveTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
`ip` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL, `ip` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
`createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`updateTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `updateTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`port` int DEFAULT NULL, `port` int DEFAULT NULL,
`expires` int DEFAULT NULL, `expires` int DEFAULT NULL,
`subscribeCycleForCatalog` int DEFAULT NULL, `subscribeCycleForCatalog` int DEFAULT NULL,
`hostAddress` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL, `hostAddress` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
`charset` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `charset` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`subscribeCycleForMobilePosition` int DEFAULT NULL, `subscribeCycleForMobilePosition` int DEFAULT NULL,
`mobilePositionSubmissionInterval` int DEFAULT '5', `mobilePositionSubmissionInterval` int DEFAULT '5',
@ -48,12 +48,13 @@ CREATE TABLE `device` (
`ssrcCheck` int DEFAULT '0', `ssrcCheck` int DEFAULT '0',
`geoCoordSys` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `geoCoordSys` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`treeType` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `treeType` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`mediaServerId` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT 'auto', `custom_name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
`custom_name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL, `password` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
`password` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL, `sdpIp` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL,
`localIp` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL,
PRIMARY KEY (`id`), PRIMARY KEY (`id`),
UNIQUE KEY `device_deviceId_uindex` (`deviceId`) UNIQUE KEY `device_deviceId_uindex` (`deviceId`)
) ENGINE=InnoDB AUTO_INCREMENT=47 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; ) ENGINE=InnoDB AUTO_INCREMENT=57 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -145,7 +146,7 @@ CREATE TABLE `device_channel` (
PRIMARY KEY (`id`), PRIMARY KEY (`id`),
UNIQUE KEY `device_channel_id_uindex` (`id`), UNIQUE KEY `device_channel_id_uindex` (`id`),
UNIQUE KEY `device_channel_pk` (`channelId`,`deviceId`) UNIQUE KEY `device_channel_pk` (`channelId`,`deviceId`)
) ENGINE=InnoDB AUTO_INCREMENT=60301 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; ) ENGINE=InnoDB AUTO_INCREMENT=74416 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -215,7 +216,7 @@ CREATE TABLE `gb_stream` (
PRIMARY KEY (`gbStreamId`) USING BTREE, PRIMARY KEY (`gbStreamId`) USING BTREE,
UNIQUE KEY `app` (`app`,`stream`) USING BTREE, UNIQUE KEY `app` (`app`,`stream`) USING BTREE,
UNIQUE KEY `gbId` (`gbId`) USING BTREE UNIQUE KEY `gbId` (`gbId`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=301059 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; ) ENGINE=InnoDB AUTO_INCREMENT=331060 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -245,7 +246,7 @@ CREATE TABLE `log` (
`username` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `username` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `createTime` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
PRIMARY KEY (`id`) USING BTREE PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=733627 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; ) ENGINE=InnoDB AUTO_INCREMENT=760908 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -337,7 +338,7 @@ CREATE TABLE `parent_platform` (
PRIMARY KEY (`id`), PRIMARY KEY (`id`),
UNIQUE KEY `parent_platform_id_uindex` (`id`), UNIQUE KEY `parent_platform_id_uindex` (`id`),
UNIQUE KEY `parent_platform_pk` (`serverGBId`) UNIQUE KEY `parent_platform_pk` (`serverGBId`)
) ENGINE=InnoDB AUTO_INCREMENT=44 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; ) ENGINE=InnoDB AUTO_INCREMENT=47 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -389,7 +390,7 @@ CREATE TABLE `platform_gb_channel` (
`catalogId` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `catalogId` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
`deviceChannelId` int NOT NULL, `deviceChannelId` int NOT NULL,
PRIMARY KEY (`id`) PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=102 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; ) ENGINE=InnoDB AUTO_INCREMENT=3146 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -415,7 +416,7 @@ CREATE TABLE `platform_gb_stream` (
`id` int NOT NULL AUTO_INCREMENT, `id` int NOT NULL AUTO_INCREMENT,
PRIMARY KEY (`id`), PRIMARY KEY (`id`),
UNIQUE KEY `platform_gb_stream_pk` (`platformId`,`catalogId`,`gbStreamId`) UNIQUE KEY `platform_gb_stream_pk` (`platformId`,`catalogId`,`gbStreamId`)
) ENGINE=InnoDB AUTO_INCREMENT=301766 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC; ) ENGINE=InnoDB AUTO_INCREMENT=391772 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -457,7 +458,7 @@ CREATE TABLE `stream_proxy` (
`enable_disable_none_reader` bit(1) DEFAULT NULL, `enable_disable_none_reader` bit(1) DEFAULT NULL,
PRIMARY KEY (`id`), PRIMARY KEY (`id`),
UNIQUE KEY `stream_proxy_pk` (`app`,`stream`) UNIQUE KEY `stream_proxy_pk` (`app`,`stream`)
) ENGINE=InnoDB AUTO_INCREMENT=548 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; ) ENGINE=InnoDB AUTO_INCREMENT=568 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -494,7 +495,7 @@ CREATE TABLE `stream_push` (
`self` int DEFAULT NULL, `self` int DEFAULT NULL,
PRIMARY KEY (`id`), PRIMARY KEY (`id`),
UNIQUE KEY `stream_push_pk` (`app`,`stream`) UNIQUE KEY `stream_push_pk` (`app`,`stream`)
) ENGINE=InnoDB AUTO_INCREMENT=310558 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; ) ENGINE=InnoDB AUTO_INCREMENT=361492 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
/*!40101 SET character_set_client = @saved_cs_client */; /*!40101 SET character_set_client = @saved_cs_client */;
-- --
@ -572,4 +573,4 @@ UNLOCK TABLES;
/*!40101 SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION */; /*!40101 SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION */;
/*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */; /*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;
-- Dump completed on 2022-10-18 17:00:02 -- Dump completed on 2022-11-29 11:47:46

2
src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
View File

@ -29,8 +29,6 @@ public class SipLayer implements CommandLineRunner {
@Autowired @Autowired
private ISIPProcessorObserver sipProcessorObserver; private ISIPProcessorObserver sipProcessorObserver;
private final Map<String, SipProviderImpl> tcpSipProviderMap = new ConcurrentHashMap<>(); private final Map<String, SipProviderImpl> tcpSipProviderMap = new ConcurrentHashMap<>();
private final Map<String, SipProviderImpl> udpSipProviderMap = new ConcurrentHashMap<>(); private final Map<String, SipProviderImpl> udpSipProviderMap = new ConcurrentHashMap<>();

72
src/main/java/com/genersoft/iot/vmp/gb28181/auth/DigestServerAuthenticationHelper.java
View File

@ -25,10 +25,9 @@
*/ */
package com.genersoft.iot.vmp.gb28181.auth; package com.genersoft.iot.vmp.gb28181.auth;
import java.security.MessageDigest; import gov.nist.core.InternalErrorHandler;
import java.security.NoSuchAlgorithmException; import org.slf4j.Logger;
import java.time.Instant; import org.slf4j.LoggerFactory;
import java.util.Random;
import javax.sip.address.URI; import javax.sip.address.URI;
import javax.sip.header.AuthorizationHeader; import javax.sip.header.AuthorizationHeader;
@ -36,10 +35,10 @@ import javax.sip.header.HeaderFactory;
import javax.sip.header.WWWAuthenticateHeader; import javax.sip.header.WWWAuthenticateHeader;
import javax.sip.message.Request; import javax.sip.message.Request;
import javax.sip.message.Response; import javax.sip.message.Response;
import java.security.MessageDigest;
import gov.nist.core.InternalErrorHandler; import java.security.NoSuchAlgorithmException;
import org.slf4j.Logger; import java.time.Instant;
import org.slf4j.LoggerFactory; import java.util.Random;
/** /**
* Implements the HTTP digest authentication method server side functionality. * Implements the HTTP digest authentication method server side functionality.
@ -201,12 +200,13 @@ public class DigestServerAuthenticationHelper {
// String ncStr = new DecimalFormat("00000000").format(Integer.parseInt(nc + "", 16)); // String ncStr = new DecimalFormat("00000000").format(Integer.parseInt(nc + "", 16));
String A1 = username + ":" + realm + ":" + pass; String A1 = username + ":" + realm + ":" + pass;
String A2 = request.getMethod().toUpperCase() + ":" + uri.toString(); String A2 = request.getMethod().toUpperCase() + ":" + uri.toString();
byte mdbytes[] = messageDigest.digest(A1.getBytes()); byte mdbytes[] = messageDigest.digest(A1.getBytes());
String HA1 = toHexString(mdbytes); String HA1 = toHexString(mdbytes);
logger.debug("A1: " + A1); logger.debug("A1: " + A1);
logger.debug("A2: " + A2); logger.debug("A2: " + A2);
mdbytes = messageDigest.digest(A2.getBytes()); mdbytes = messageDigest.digest(A2.getBytes());
String HA2 = toHexString(mdbytes); String HA2 = toHexString(mdbytes);
logger.debug("HA1: " + HA1); logger.debug("HA1: " + HA1);
@ -238,58 +238,4 @@ public class DigestServerAuthenticationHelper {
} }
// public static void main(String[] args) throws NoSuchAlgorithmException {
// String realm = "3402000000";
// String username = "44010000001180008012";
// String nonce = "07cab60999fbf643264ace27d3b7de8b";
// String uri = "sip:34020000002000000001@3402000000";
// // qop 保护质量 包含auth（默认的）和auth-int（增加了报文完整性检测）两种策略
// String qop = "auth";
// // 客户端随机数，这是一个不透明的字符串值，由客户端提供，并且客户端和服务器都会使用，以避免用明文文本。
// // 这使得双方都可以查验对方的身份，并对消息的完整性提供一些保护
// //String cNonce = authHeader.getCNonce();
// // nonce计数器，是一个16进制的数值，表示同一nonce下客户端发送出请求的数量
// int nc = 1;
// String ncStr = new DecimalFormat("00000000").format(nc);
// // String ncStr = new DecimalFormat("00000000").format(Integer.parseInt(nc + "", 16));
// MessageDigest messageDigest = MessageDigest.getInstance(DEFAULT_ALGORITHM);
// String A1 = username + ":" + realm + ":" + "12345678";
// String A2 = "REGISTER" + ":" + uri;
// byte mdbytes[] = messageDigest.digest(A1.getBytes());
// String HA1 = toHexString(mdbytes);
// System.out.println("A1: " + A1);
// System.out.println("A2: " + A2);
// mdbytes = messageDigest.digest(A2.getBytes());
// String HA2 = toHexString(mdbytes);
// System.out.println("HA1: " + HA1);
// System.out.println("HA2: " + HA2);
// String cnonce = "0a4f113b";
// System.out.println("nonce: " + nonce);
// System.out.println("nc: " + ncStr);
// System.out.println("cnonce: " + cnonce);
// System.out.println("qop: " + qop);
// String KD = HA1 + ":" + nonce;
// if (qop != null && qop.equals("auth") ) {
// if (nc != -1) {
// KD += ":" + ncStr;
// }
// if (cnonce != null) {
// KD += ":" + cnonce;
// }
// KD += ":" + qop;
// }
// KD += ":" + HA2;
// System.out.println("KD: " + KD);
// mdbytes = messageDigest.digest(KD.getBytes());
// String mdString = toHexString(mdbytes);
// System.out.println("mdString: " + mdString);
// String response = "4f0507d4b87cdecff04bdaf4c96348f0";
// System.out.println("response: " + response);
// }
} }

11
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
View File

@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl; package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.gb28181.SipLayer; import com.genersoft.iot.vmp.gb28181.SipLayer;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
@ -9,13 +8,13 @@ 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.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider; import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils; import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.dto.PlatformRegisterInfo;
import com.genersoft.iot.vmp.utils.DateUtil;
import gov.nist.javax.sip.message.MessageFactoryImpl; import gov.nist.javax.sip.message.MessageFactoryImpl;
import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPRequest;
import org.slf4j.Logger; import org.slf4j.Logger;
@ -26,8 +25,10 @@ import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import javax.sip.*; import javax.sip.InvalidArgumentException;
import javax.sip.header.*; import javax.sip.SipException;
import javax.sip.header.CallIdHeader;
import javax.sip.header.WWWAuthenticateHeader;
import javax.sip.message.Request; import javax.sip.message.Request;
import java.text.ParseException; import java.text.ParseException;
import java.util.ArrayList; import java.util.ArrayList;

9
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
View File

@ -11,7 +11,6 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor; 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.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil; import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils; import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil; import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
@ -31,7 +30,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
@ -77,8 +75,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
@Autowired @Autowired
private IDeviceChannelService deviceChannelService; private IDeviceChannelService deviceChannelService;
private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor") @Qualifier("taskExecutor")
@ -98,9 +94,9 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
}catch (SipException | InvalidArgumentException | ParseException e) { }catch (SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace(); e.printStackTrace();
} }
boolean runed = !taskQueue.isEmpty();
taskQueue.offer(new HandlerCatchData(evt, null, null)); taskQueue.offer(new HandlerCatchData(evt, null, null));
if (!taskQueueHandlerRun) { if (!runed) {
taskQueueHandlerRun = true;
taskExecutor.execute(()-> { taskExecutor.execute(()-> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
try { try {
@ -128,7 +124,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
logger.error("处理NOTIFY消息时错误", e); logger.error("处理NOTIFY消息时错误", e);
} }
} }
taskQueueHandlerRun = false;
}); });
} }
} }

31
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java
View File

@ -9,7 +9,6 @@ import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; 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.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.NotifyMessageHandler; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.NotifyMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil; import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil; import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.service.IDeviceAlarmService; import com.genersoft.iot.vmp.service.IDeviceAlarmService;
@ -27,17 +26,15 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
import javax.sip.SipException; import javax.sip.SipException;
import javax.sip.message.Response; import javax.sip.message.Response;
import java.text.ParseException; import java.text.ParseException;
import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentLinkedQueue;
import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.*; import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText;
/** /**
* 报警事件的处理，参考：9.4 * 报警事件的处理，参考：9.4
@ -72,8 +69,6 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
@Autowired @Autowired
private IDeviceChannelService deviceChannelService; private IDeviceChannelService deviceChannelService;
private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<SipMsgInfo> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<SipMsgInfo> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor") @Qualifier("taskExecutor")
@ -89,20 +84,20 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
@Override @Override
public void handForDevice(RequestEvent evt, Device device, Element rootElement) { public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
logger.info("[收到报警通知]设备：{}", device.getDeviceId()); logger.info("[收到报警通知]设备：{}", device.getDeviceId());
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(new SipMsgInfo(evt, device, rootElement)); taskQueue.offer(new SipMsgInfo(evt, device, rootElement));
if (!taskQueueHandlerRun) { // 回复200 OK
taskQueueHandlerRun = true; try {
responseAck((SIPRequest) evt.getRequest(), Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 报警通知回复: {}", e.getMessage());
}
if (isEmpty) {
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
logger.info("[处理报警通知]待处理数量：{}", taskQueue.size() ); logger.info("[处理报警通知]待处理数量：{}", taskQueue.size() );
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
SipMsgInfo sipMsgInfo = taskQueue.poll();
// 回复200 OK
try { try {
responseAck((SIPRequest) sipMsgInfo.getEvt().getRequest(), Response.OK); SipMsgInfo sipMsgInfo = taskQueue.poll();
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[处理报警通知], 回复200OK失败", e);
}
Element deviceIdElement = sipMsgInfo.getRootElement().element("DeviceID"); Element deviceIdElement = sipMsgInfo.getRootElement().element("DeviceID");
String channelId = deviceIdElement.getText().toString(); String channelId = deviceIdElement.getText().toString();
@ -205,12 +200,12 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
if (redisCatchStorage.deviceIsOnline(sipMsgInfo.getDevice().getDeviceId())) { if (redisCatchStorage.deviceIsOnline(sipMsgInfo.getDevice().getDeviceId())) {
publisher.deviceAlarmEventPublish(deviceAlarm); publisher.deviceAlarmEventPublish(deviceAlarm);
} }
}catch (Exception e) {
logger.warn("[收到报警通知] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
}
} }
taskQueueHandlerRun = false;
}); });
} }
} }
@Override @Override

32
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MobilePositionNotifyMessageHandler.java
View File

@ -22,7 +22,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
@ -57,8 +56,6 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
@Autowired @Autowired
private IDeviceChannelService deviceChannelService; private IDeviceChannelService deviceChannelService;
private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<SipMsgInfo> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<SipMsgInfo> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor") @Qualifier("taskExecutor")
@ -73,21 +70,22 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
@Override @Override
public void handForDevice(RequestEvent evt, Device device, Element rootElement) { public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(new SipMsgInfo(evt, device, rootElement)); taskQueue.offer(new SipMsgInfo(evt, device, rootElement));
if (!taskQueueHandlerRun) { // 回复200 OK
taskQueueHandlerRun = true; try {
responseAck((SIPRequest) evt.getRequest(), Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 移动位置通知回复: {}", e.getMessage());
}
if (isEmpty) {
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
SipMsgInfo sipMsgInfo = taskQueue.poll(); SipMsgInfo sipMsgInfo = taskQueue.poll();
try { try {
Element rootElementAfterCharset = getRootElement(sipMsgInfo.getEvt(), sipMsgInfo.getDevice().getCharset()); Element rootElementAfterCharset = getRootElement(sipMsgInfo.getEvt(), sipMsgInfo.getDevice().getCharset());
if (rootElementAfterCharset == null) { if (rootElementAfterCharset == null) {
try { logger.warn("[移动位置通知] {}处理失败，未识别到信息体", device.getDeviceId());
logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", sipMsgInfo.getEvt().getRequest());
responseAck((SIPRequest) sipMsgInfo.getEvt().getRequest(), Response.BAD_REQUEST);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 移动设备位置数据通知 内容为空: {}", e.getMessage());
}
continue; continue;
} }
MobilePosition mobilePosition = new MobilePosition(); MobilePosition mobilePosition = new MobilePosition();
@ -137,12 +135,6 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
storager.insertMobilePosition(mobilePosition); storager.insertMobilePosition(mobilePosition);
} }
storager.updateChannelPosition(deviceChannel); storager.updateChannelPosition(deviceChannel);
//回复 200 OK
try {
responseAck((SIPRequest) sipMsgInfo.getEvt().getRequest(), Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 移动设备位置数据回复200: {}", e.getMessage());
}
// 发送redis消息。 通知位置信息的变化 // 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject(); JSONObject jsonObject = new JSONObject();
@ -158,14 +150,12 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
} catch (DocumentException e) { } catch (DocumentException e) {
e.printStackTrace(); e.printStackTrace();
} catch (Exception e) {
logger.warn("[移动位置通知] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
} }
} }
taskQueueHandlerRun = false;
}); });
} }
} }
@Override @Override

23
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java
View File

@ -1,18 +1,11 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd; package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.CatalogDataCatch; import com.genersoft.iot.vmp.gb28181.session.CatalogDataCatch;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; 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.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil; import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException; import org.dom4j.DocumentException;
@ -24,7 +17,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
@ -45,12 +37,10 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
private Logger logger = LoggerFactory.getLogger(CatalogResponseMessageHandler.class); private Logger logger = LoggerFactory.getLogger(CatalogResponseMessageHandler.class);
private final String cmdType = "Catalog"; private final String cmdType = "Catalog";
private boolean taskQueueHandlerRun = false;
@Autowired @Autowired
private ResponseMessageHandler responseMessageHandler; private ResponseMessageHandler responseMessageHandler;
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>(); private final ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
@Autowired @Autowired
private IVideoManagerStorage storager; private IVideoManagerStorage storager;
@ -69,6 +59,7 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
@Override @Override
public void handForDevice(RequestEvent evt, Device device, Element element) { public void handForDevice(RequestEvent evt, Device device, Element element) {
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(new HandlerCatchData(evt, device, element)); taskQueue.offer(new HandlerCatchData(evt, device, element));
// 回复200 OK // 回复200 OK
try { try {
@ -76,10 +67,12 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 目录查询回复: {}", e.getMessage()); logger.error("[命令发送失败] 目录查询回复: {}", e.getMessage());
} }
if (!taskQueueHandlerRun) { // 如果不为空则说明已经开启消息处理
taskQueueHandlerRun = true; if (isEmpty) {
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
// 全局异常捕获，保证下一条可以得到处理
try {
HandlerCatchData take = taskQueue.poll(); HandlerCatchData take = taskQueue.poll();
Element rootElement = null; Element rootElement = null;
try { try {
@ -135,8 +128,10 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
} }
} }
}catch (Exception e) {
logger.warn("[收到通道] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
}
} }
taskQueueHandlerRun = false;
}); });
} }

17
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
View File

@ -9,7 +9,6 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorP
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler; 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.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException; import org.dom4j.DocumentException;
import org.dom4j.Element; import org.dom4j.Element;
@ -21,17 +20,17 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
import javax.sip.SipException; import javax.sip.SipException;
import javax.sip.message.Response; import javax.sip.message.Response;
import java.text.ParseException; import java.text.ParseException;
import java.util.*; import java.util.ArrayList;
import java.util.concurrent.BlockingQueue; import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText; import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText;
@ -46,7 +45,6 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
private boolean taskQueueHandlerRun = false;
@Autowired @Autowired
private ResponseMessageHandler responseMessageHandler; private ResponseMessageHandler responseMessageHandler;
@ -70,6 +68,7 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
@Override @Override
public void handForDevice(RequestEvent evt, Device device, Element rootElement) { public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
boolean isEmpty = taskQueue.isEmpty();
try { try {
// 回复200 OK // 回复200 OK
responseAck((SIPRequest) evt.getRequest(), Response.OK); responseAck((SIPRequest) evt.getRequest(), Response.OK);
@ -77,8 +76,7 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage()); logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage());
} }
taskQueue.offer(new HandlerCatchData(evt, device, rootElement)); taskQueue.offer(new HandlerCatchData(evt, device, rootElement));
if (!taskQueueHandlerRun) { if (isEmpty) {
taskQueueHandlerRun = true;
taskExecutor.execute(()->{ taskExecutor.execute(()->{
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
try { try {
@ -151,9 +149,10 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
} }
} catch (DocumentException e) { } catch (DocumentException e) {
logger.error("xml解析异常： ", e); logger.error("xml解析异常： ", e);
} catch (Exception e) {
logger.warn("[国标录像] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
} }
} }
taskQueueHandlerRun = false;
}); });
} }
} }

15
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisAlarmMsgListener.java
View File

@ -38,8 +38,6 @@ public class RedisAlarmMsgListener implements MessageListener {
@Autowired @Autowired
private IVideoManagerStorage storage; private IVideoManagerStorage storage;
private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor") @Qualifier("taskExecutor")
@ -49,15 +47,14 @@ public class RedisAlarmMsgListener implements MessageListener {
@Override @Override
public void onMessage(@NotNull Message message, byte[] bytes) { public void onMessage(@NotNull Message message, byte[] bytes) {
logger.info("收到来自REDIS的ALARM通知： {}", new String(message.getBody())); logger.info("收到来自REDIS的ALARM通知： {}", new String(message.getBody()));
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message); taskQueue.offer(message);
if (!taskQueueHandlerRun) { if (isEmpty) {
taskQueueHandlerRun = true;
logger.info("[线程池信息]活动线程数：{}, 最大线程数： {}", taskExecutor.getActiveCount(), taskExecutor.getMaxPoolSize()); logger.info("[线程池信息]活动线程数：{}, 最大线程数： {}", taskExecutor.getActiveCount(), taskExecutor.getMaxPoolSize());
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll(); Message msg = taskQueue.poll();
try {
AlarmChannelMessage alarmChannelMessage = JSON.parseObject(msg.getBody(), AlarmChannelMessage.class); AlarmChannelMessage alarmChannelMessage = JSON.parseObject(msg.getBody(), AlarmChannelMessage.class);
if (alarmChannelMessage == null) { if (alarmChannelMessage == null) {
logger.warn("[REDIS的ALARM通知]消息解析失败"); logger.warn("[REDIS的ALARM通知]消息解析失败");
@ -107,11 +104,11 @@ public class RedisAlarmMsgListener implements MessageListener {
logger.warn("无法确定" + gbId + "是平台还是设备"); logger.warn("无法确定" + gbId + "是平台还是设备");
} }
} }
}catch (Exception e) {
logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
}
} }
taskQueueHandlerRun = false;
}); });
} }
} }
} }

21
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
View File

@ -88,8 +88,6 @@ public class RedisGbPlayMsgListener implements MessageListener {
@Autowired @Autowired
private ZlmHttpHookSubscribe subscribe; private ZlmHttpHookSubscribe subscribe;
private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor") @Qualifier("taskExecutor")
@ -111,13 +109,13 @@ public class RedisGbPlayMsgListener implements MessageListener {
@Override @Override
public void onMessage(Message message, byte[] bytes) { public void onMessage(Message message, byte[] bytes) {
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message); taskQueue.offer(message);
if (!taskQueueHandlerRun) { if (isEmpty) {
taskQueueHandlerRun = true;
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll(); Message msg = taskQueue.poll();
try {
JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class); JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON); WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) { if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
@ -132,7 +130,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial()); requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
break; break;
case WvpRedisMsgCmd.REQUEST_PUSH_STREAM: case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());; RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());
requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial()); requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
break; break;
default: default:
@ -201,17 +199,14 @@ public class RedisGbPlayMsgListener implements MessageListener {
default: default:
break; break;
} }
}
}catch (Exception e) {
logger.warn("[RedisGbPlayMsg] 发现未处理的异常, {}",e.getMessage());
} }
} }
taskQueueHandlerRun = false;
}); });
} }
} }
/** /**

11
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGpsMsgListener.java
View File

@ -27,8 +27,6 @@ public class RedisGpsMsgListener implements MessageListener {
private final static Logger logger = LoggerFactory.getLogger(RedisGpsMsgListener.class); private final static Logger logger = LoggerFactory.getLogger(RedisGpsMsgListener.class);
private boolean taskQueueHandlerRun = false;
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@ -44,17 +42,20 @@ public class RedisGpsMsgListener implements MessageListener {
@Override @Override
public void onMessage(@NotNull Message message, byte[] bytes) { public void onMessage(@NotNull Message message, byte[] bytes) {
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message); taskQueue.offer(message);
if (!taskQueueHandlerRun) { if (isEmpty) {
taskQueueHandlerRun = true;
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll(); Message msg = taskQueue.poll();
try {
GPSMsgInfo gpsMsgInfo = JSON.parseObject(msg.getBody(), GPSMsgInfo.class); GPSMsgInfo gpsMsgInfo = JSON.parseObject(msg.getBody(), GPSMsgInfo.class);
// 只是放入redis缓存起来 // 只是放入redis缓存起来
redisCatchStorage.updateGpsMsgInfo(gpsMsgInfo); redisCatchStorage.updateGpsMsgInfo(gpsMsgInfo);
}catch (Exception e) {
logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
}
} }
taskQueueHandlerRun = false;
}); });
} }
} }

14
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java
View File

@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.service.redisMsg; package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannelResponse; import com.genersoft.iot.vmp.service.bean.MessageForPushChannelResponse;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -26,8 +25,6 @@ public class RedisPushStreamResponseListener implements MessageListener {
private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamResponseListener.class); private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamResponseListener.class);
private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor") @Qualifier("taskExecutor")
@ -43,13 +40,14 @@ public class RedisPushStreamResponseListener implements MessageListener {
@Override @Override
public void onMessage(Message message, byte[] bytes) { public void onMessage(Message message, byte[] bytes) {
logger.warn("[REDIS消息-请求推流结果]： {}", new String(message.getBody())); logger.info("[REDIS消息-请求推流结果]： {}", new String(message.getBody()));
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message); taskQueue.offer(message);
if (!taskQueueHandlerRun) { if (isEmpty) {
taskQueueHandlerRun = true;
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll(); Message msg = taskQueue.poll();
try {
MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class); MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class);
if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){ if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){
logger.info("[REDIS消息-请求推流结果]：参数不全"); logger.info("[REDIS消息-请求推流结果]：参数不全");
@ -59,8 +57,10 @@ public class RedisPushStreamResponseListener implements MessageListener {
if (responseEvents.get(response.getApp() + response.getStream()) != null) { if (responseEvents.get(response.getApp() + response.getStream()) != null) {
responseEvents.get(response.getApp() + response.getStream()).run(response); responseEvents.get(response.getApp() + response.getStream()).run(response);
} }
}catch (Exception e) {
logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
}
} }
taskQueueHandlerRun = false;
}); });
} }
} }

15
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusListMsgListener.java
View File

@ -6,7 +6,6 @@ import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.IGbStreamService; import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.DateUtil;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -18,7 +17,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.util.*; import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentLinkedQueue;
/** /**
@ -38,7 +38,6 @@ public class RedisPushStreamStatusListMsgListener implements MessageListener {
@Resource @Resource
private IGbStreamService gbStreamService; private IGbStreamService gbStreamService;
private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@ -49,13 +48,13 @@ public class RedisPushStreamStatusListMsgListener implements MessageListener {
@Override @Override
public void onMessage(Message message, byte[] bytes) { public void onMessage(Message message, byte[] bytes) {
logger.info("[REDIS消息-推流设备列表更新]： {}", new String(message.getBody())); logger.info("[REDIS消息-推流设备列表更新]： {}", new String(message.getBody()));
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message); taskQueue.offer(message);
if (!taskQueueHandlerRun) { if (isEmpty) {
taskQueueHandlerRun = true;
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll(); Message msg = taskQueue.poll();
try {
List<StreamPushItem> streamPushItems = JSON.parseArray(new String(msg.getBody()), StreamPushItem.class); List<StreamPushItem> streamPushItems = JSON.parseArray(new String(msg.getBody()), StreamPushItem.class);
//查询全部的app+stream 用于判断是添加还是修改 //查询全部的app+stream 用于判断是添加还是修改
List<String> allAppAndStream = streamPushService.getAllAppAndStream(); List<String> allAppAndStream = streamPushService.getAllAppAndStream();
@ -95,8 +94,10 @@ public class RedisPushStreamStatusListMsgListener implements MessageListener {
logger.info(JSONObject.toJSONString(streamPushItemForUpdate)); logger.info(JSONObject.toJSONString(streamPushItemForUpdate));
gbStreamService.updateGbIdOrName(streamPushItemForUpdate); gbStreamService.updateGbIdOrName(streamPushItemForUpdate);
} }
}catch (Exception e) {
logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
}
} }
taskQueueHandlerRun = false;
}); });
} }
} }

11
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java
View File

@ -29,8 +29,6 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic
private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusMsgListener.class); private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusMsgListener.class);
private boolean taskQueueHandlerRun = false;
@Autowired @Autowired
private IRedisCatchStorage redisCatchStorage; private IRedisCatchStorage redisCatchStorage;
@ -50,14 +48,15 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic
@Override @Override
public void onMessage(Message message, byte[] bytes) { public void onMessage(Message message, byte[] bytes) {
boolean isEmpty = taskQueue.isEmpty();
logger.warn("[REDIS消息-推流设备状态变化]： {}", new String(message.getBody())); logger.warn("[REDIS消息-推流设备状态变化]： {}", new String(message.getBody()));
taskQueue.offer(message); taskQueue.offer(message);
if (!taskQueueHandlerRun) { if (isEmpty) {
taskQueueHandlerRun = true;
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll(); Message msg = taskQueue.poll();
try {
PushStreamStatusChangeFromRedisDto statusChangeFromPushStream = JSON.parseObject(msg.getBody(), PushStreamStatusChangeFromRedisDto.class); PushStreamStatusChangeFromRedisDto statusChangeFromPushStream = JSON.parseObject(msg.getBody(), PushStreamStatusChangeFromRedisDto.class);
if (statusChangeFromPushStream == null) { if (statusChangeFromPushStream == null) {
logger.warn("[REDIS消息]推流设备状态变化消息解析失败"); logger.warn("[REDIS消息]推流设备状态变化消息解析失败");
@ -79,8 +78,10 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic
// 更新部分设备上线 // 更新部分设备上线
streamPushService.online(statusChangeFromPushStream.getOnlineStreams()); streamPushService.online(statusChangeFromPushStream.getOnlineStreams());
} }
}catch (Exception e) {
logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
}
} }
taskQueueHandlerRun = false;
}); });
} }
} }

12
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
View File

@ -33,8 +33,6 @@ public class RedisStreamMsgListener implements MessageListener {
@Autowired @Autowired
private ZLMMediaListManager zlmMediaListManager; private ZLMMediaListManager zlmMediaListManager;
private boolean taskQueueHandlerRun = false;
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>(); private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor") @Qualifier("taskExecutor")
@ -43,13 +41,13 @@ public class RedisStreamMsgListener implements MessageListener {
@Override @Override
public void onMessage(Message message, byte[] bytes) { public void onMessage(Message message, byte[] bytes) {
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message); taskQueue.offer(message);
if (!taskQueueHandlerRun) { if (isEmpty) {
taskQueueHandlerRun = true;
taskExecutor.execute(() -> { taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) { while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll(); Message msg = taskQueue.poll();
try {
JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class); JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
if (steamMsgJson == null) { if (steamMsgJson == null) {
logger.warn("[收到redis 流变化]消息解析失败"); logger.warn("[收到redis 流变化]消息解析失败");
@ -83,8 +81,10 @@ public class RedisStreamMsgListener implements MessageListener {
}else { }else {
zlmMediaListManager.removeMedia(app, stream); zlmMediaListManager.removeMedia(app, stream);
} }
}catch (Exception e) {
logger.warn("[REDIS的ALARM通知] 发现未处理的异常, {}",e.getMessage());
}
} }
taskQueueHandlerRun = false;
}); });
} }
} }

2
src/main/resources/all-application.yml
View File

@ -150,8 +150,6 @@ media:
enable: true enable: true
# [可选] 在此范围内选择端口用于媒体流传输, 必须提前在zlm上配置该属性，不然自动配置此属性可能不成功 # [可选] 在此范围内选择端口用于媒体流传输, 必须提前在zlm上配置该属性，不然自动配置此属性可能不成功
port-range: 30000,30500 # 端口范围 port-range: 30000,30500 # 端口范围
# [可选] 国标级联在此范围内选择端口发送媒体流
send-port-range: 30000,30500 # 端口范围
# 录像辅助服务， 部署此服务可以实现zlm录像的管理与下载， 0 表示不使用 # 录像辅助服务， 部署此服务可以实现zlm录像的管理与下载， 0 表示不使用
record-assist-port: 0 record-assist-port: 0