guzijian / wvp-GB28181-pro · Commits

Commit cb16cabb64e99d7f505822a49665725ff5b61ff6

Authored by 648540858
2 parents 116d979d c041aacc

Merge branch 'wvp-28181-2.0' 

# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
Inline Side-by-side
Showing 18 changed files with 301 additions and 306 deletions
doc/_content/introduction/deployment.md
  View file @cb16cab
... ... @@ -24,7 +24,7 @@
24 24 7. 启动服务，以linux为例
25 25 **启动WVP-PRO**
26 26 ```shell
27   -nohup java -jar java -jar wvp-pro-*.jar &
  27 +nohup java -jar wvp-pro-*.jar &
28 28 ```
29 29  
30 30 **启动ZLM**
... ...
... ... @@ -61,13 +61,6 @@
61 61 <dependency>
62 62 <groupId>org.springframework.boot</groupId>
63 63 <artifactId>spring-boot-starter-data-redis</artifactId>
64   - <exclusions>
65   - <!-- 去掉 Lettuce 的依赖， Spring Boot 优先使用 Lettuce 作为 Redis 客户端 -->
66   - <exclusion>
67   - <groupId>io.lettuce</groupId>
68   - <artifactId>lettuce-core</artifactId>
69   - </exclusion>
70   - </exclusions>
71 64 </dependency>
72 65 <dependency>
73 66 <groupId>org.springframework.boot</groupId>
... ... @@ -94,11 +87,6 @@
94 87 <artifactId>spring-boot-starter-security</artifactId>
95 88 </dependency>
96 89  
97   - <dependency>
98   - <groupId>redis.clients</groupId>
99   - <artifactId>jedis</artifactId>
100   - </dependency>
101   -
102 90 <!-- druid数据库连接池 -->
103 91 <dependency>
104 92 <groupId>com.alibaba</groupId>
... ...
src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
  View file @cb16cab
... ... @@ -77,38 +77,54 @@ public class VideoManagerConstants {
77 77  
78 78 //************************** redis 消息*********************************
79 79  
80   - // 流变化的通知
  80 + /**
  81 + * 流变化的通知
  82 + */
81 83 public static final String WVP_MSG_STREAM_CHANGE_PREFIX = "WVP_MSG_STREAM_CHANGE_";
82 84  
83   - // 接收推流设备的GPS变化通知
  85 + /**
  86 + * 接收推流设备的GPS变化通知
  87 + */
84 88 public static final String VM_MSG_GPS = "VM_MSG_GPS";
85 89  
86   - // 接收推流设备的GPS变化通知
  90 + /**
  91 + * 接收推流设备的GPS变化通知
  92 + */
87 93 public static final String VM_MSG_PUSH_STREAM_STATUS_CHANGE = "VM_MSG_PUSH_STREAM_STATUS_CHANGE";
88 94  
89   - // redis 消息通知设备推流到平台
  95 + /**
  96 + * redis 消息通知设备推流到平台
  97 + */
90 98 public static final String VM_MSG_STREAM_PUSH_REQUESTED = "VM_MSG_STREAM_PUSH_REQUESTED";
91 99  
92   - // redis 消息请求所有的在线通道
  100 + /**
  101 + * redis 消息请求所有的在线通道
  102 + */
93 103 public static final String VM_MSG_GET_ALL_ONLINE_REQUESTED = "VM_MSG_GET_ALL_ONLINE_REQUESTED";
94 104  
95   - // 移动位置订阅通知
  105 + /**
  106 + * 移动位置订阅通知
  107 + */
96 108 public static final String VM_MSG_SUBSCRIBE_MOBILE_POSITION = "mobileposition";
97 109  
98   - // 报警订阅的通知（收到报警向redis发出通知）
  110 + /**
  111 + * 报警订阅的通知（收到报警向redis发出通知）
  112 + */
99 113 public static final String VM_MSG_SUBSCRIBE_ALARM = "alarm";
100 114  
101   - // 报警通知的发送 （收到redis发出的通知，转发给其他平台）
  115 + /**
  116 + * 报警通知的发送 （收到redis发出的通知，转发给其他平台）
  117 + */
102 118 public static final String VM_MSG_SUBSCRIBE_ALARM_RECEIVE= "alarm_receive";
103 119  
104   - // 设备状态订阅的通知
  120 + /**
  121 + * 设备状态订阅的通知
  122 + */
105 123 public static final String VM_MSG_SUBSCRIBE_DEVICE_STATUS = "device";
106 124  
107 125  
108   -
109   -
110   -
111 126 //************************** 第三方 ****************************************
  127 +
112 128 public static final String WVP_STREAM_GB_ID_PREFIX = "memberNo_";
113 129 public static final String WVP_STREAM_GPS_MSG_PREFIX = "WVP_STREAM_GPS_MSG_";
114 130  
... ...
src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java
  View file @cb16cab
1 1 package com.genersoft.iot.vmp.conf;
2 2  
  3 +import com.alibaba.fastjson.parser.ParserConfig;
3 4 import com.genersoft.iot.vmp.common.VideoManagerConstants;
4 5 import com.genersoft.iot.vmp.service.impl.*;
5 6 import org.apache.commons.lang3.StringUtils;
... ... @@ -9,15 +10,14 @@ import org.springframework.cache.annotation.CachingConfigurerSupport;
9 10 import org.springframework.context.annotation.Bean;
10 11 import org.springframework.context.annotation.Configuration;
11 12 import org.springframework.data.redis.connection.RedisConnectionFactory;
  13 +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
12 14 import org.springframework.data.redis.core.RedisTemplate;
13 15 import org.springframework.data.redis.listener.PatternTopic;
14 16 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
15 17 import org.springframework.data.redis.serializer.StringRedisSerializer;
16 18  
17   -import com.alibaba.fastjson.parser.ParserConfig;
18 19 import com.genersoft.iot.vmp.utils.redis.FastJsonRedisSerializer;
19   -import redis.clients.jedis.JedisPool;
20   -import redis.clients.jedis.JedisPoolConfig;
  20 +
21 21  
22 22 /**
23 23 * @description:Redis中间件配置类，使用spring-data-redis集成，自动从application.yml中加载redis配置
... ... @@ -28,23 +28,6 @@ import redis.clients.jedis.JedisPoolConfig;
28 28 @Configuration
29 29 public class RedisConfig extends CachingConfigurerSupport {
30 30  
31   - @Value("${spring.redis.host}")
32   - private String host;
33   - @Value("${spring.redis.port}")
34   - private int port;
35   - @Value("${spring.redis.database}")
36   - private int database;
37   - @Value("${spring.redis.password}")
38   - private String password;
39   - @Value("${spring.redis.timeout}")
40   - private int timeout;
41   - @Value("${spring.redis.poolMaxTotal:1000}")
42   - private int poolMaxTotal;
43   - @Value("${spring.redis.poolMaxIdle:500}")
44   - private int poolMaxIdle;
45   - @Value("${spring.redis.poolMaxWait:5}")
46   - private int poolMaxWait;
47   -
48 31 @Autowired
49 32 private RedisGpsMsgListener redisGPSMsgListener;
50 33  
... ... @@ -61,37 +44,25 @@ public class RedisConfig extends CachingConfigurerSupport {
61 44 private RedisPushStreamStatusMsgListener redisPushStreamStatusMsgListener;
62 45  
63 46 @Bean
64   - public JedisPool jedisPool() {
65   - if (StringUtils.isBlank(password)) {
66   - password = null;
67   - }
68   - JedisPoolConfig poolConfig = new JedisPoolConfig();
69   - poolConfig.setMaxIdle(poolMaxIdle);
70   - poolConfig.setMaxTotal(poolMaxTotal);
71   - // 秒转毫秒
72   - poolConfig.setMaxWaitMillis(poolMaxWait * 1000L);
73   - JedisPool jp = new JedisPool(poolConfig, host, port, timeout * 1000, password, database);
74   - return jp;
75   - }
76   -
77   - @Bean("redisTemplate")
78 47 public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
79   - RedisTemplate<Object, Object> template = new RedisTemplate<>();
80   - template.setConnectionFactory(redisConnectionFactory);
81   - // 使用fastjson进行序列化处理，提高解析效率
82   - FastJsonRedisSerializer<Object> serializer = new FastJsonRedisSerializer<Object>(Object.class);
  48 + RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
  49 + // 使用fastJson序列化
  50 + FastJsonRedisSerializer fastJsonRedisSerializer = new FastJsonRedisSerializer(Object.class);
83 51 // value值的序列化采用fastJsonRedisSerializer
84   - template.setValueSerializer(serializer);
85   - template.setHashValueSerializer(serializer);
  52 + redisTemplate.setValueSerializer(fastJsonRedisSerializer);
  53 + redisTemplate.setHashValueSerializer(fastJsonRedisSerializer);
  54 + // 全局开启AutoType，不建议使用
  55 + ParserConfig.getGlobalInstance().setAutoTypeSupport(true);
  56 + // 建议使用这种方式，小范围指定白名单，需要序列化的类
  57 +// ParserConfig.getGlobalInstance().addAccept("com.avatar");
86 58 // key的序列化采用StringRedisSerializer
87   - template.setKeySerializer(new StringRedisSerializer());
88   - template.setHashKeySerializer(new StringRedisSerializer());
89   - template.setConnectionFactory(redisConnectionFactory);
90   - // 使用fastjson时需设置此项，否则会报异常not support type
91   - ParserConfig.getGlobalInstance().setAutoTypeSupport(true);
92   - return template;
  59 + redisTemplate.setKeySerializer(new StringRedisSerializer());
  60 + redisTemplate.setHashKeySerializer(new StringRedisSerializer());
  61 + redisTemplate.setConnectionFactory(redisConnectionFactory);
  62 + return redisTemplate;
93 63 }
94 64  
  65 +
95 66 /**
96 67 * redis消息监听器容器 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
97 68 * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理
... ...
src/main/java/com/genersoft/iot/vmp/conf/RedisKeyExpirationEventMessageListener.java
  View file @cb16cab
... ... @@ -28,7 +28,7 @@ public class RedisKeyExpirationEventMessageListener extends KeyExpirationEventMe
28 28 RedisConnection connection = this.listenerContainer.getConnectionFactory().getConnection();
29 29 Properties config = connection.getConfig("notify-keyspace-events");
30 30 try {
31   - if (!config.getProperty("notify-keyspace-events").equals(keyspaceNotificationsConfigParameter)) {
  31 + if (!keyspaceNotificationsConfigParameter.equals(config.getProperty("notify-keyspace-events"))) {
32 32 connection.setConfig("notify-keyspace-events", keyspaceNotificationsConfigParameter);
33 33 }
34 34 } finally {
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
  View file @cb16cab
... ... @@ -53,10 +53,15 @@ public class SipLayer{
53 53 * gov/nist/javax/sip/SipStackImpl.class
54 54 */
55 55 properties.setProperty("gov.nist.javax.sip.LOG_MESSAGE_CONTENT", "true");
56   - properties.setProperty("gov.nist.javax.sip.DELIVER_UNSOLICITED_NOTIFY", "true"); // 接收所有notify请求，即使没有订阅
57   - properties.setProperty("gov.nist.javax.sip.DELIVER_TERMINATED_EVENT_FOR_NULL_DIALOG", "true"); // 为_NULL _对话框传递_终止的_事件
58   - properties.setProperty("gov.nist.javax.sip.RELEASE_REFERENCES_STRATEGY", "Normal"); // 会话清理策略
59   - properties.setProperty("gov.nist.javax.sip.RELIABLE_CONNECTION_KEEP_ALIVE_TIMEOUT", "10");
  56 + // 接收所有notify请求，即使没有订阅
  57 + properties.setProperty("gov.nist.javax.sip.DELIVER_UNSOLICITED_NOTIFY", "true");
  58 + // 为_NULL _对话框传递_终止的_事件
  59 + properties.setProperty("gov.nist.javax.sip.DELIVER_TERMINATED_EVENT_FOR_NULL_DIALOG", "true");
  60 + // 会话清理策略
  61 + properties.setProperty("gov.nist.javax.sip.RELEASE_REFERENCES_STRATEGY", "Normal");
  62 + // 处理由该服务器处理的基于底层TCP的保持生存超时
  63 + properties.setProperty("gov.nist.javax.sip.RELIABLE_CONNECTION_KEEP_ALIVE_TIMEOUT", "60");
  64 +
60 65 /**
61 66 * sip_server_log.log 和 sip_debug_log.log public static final int TRACE_NONE =
62 67 * 0; public static final int TRACE_MESSAGES = 16; public static final int
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java
  View file @cb16cab
... ... @@ -62,7 +62,7 @@ public class SIPRequestHeaderPlarformProvider {
62 62 // Forwards
63 63 MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70);
64 64 // ceq
65   - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.MESSAGE), Request.MESSAGE);
  65 + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.MESSAGE);
66 66  
67 67 request = sipFactory.createMessageFactory().createRequest(requestURI, Request.MESSAGE, callIdHeader, cSeqHeader, fromHeader,
68 68 toHeader, viaHeaders, maxForwards);
... ... @@ -120,7 +120,7 @@ public class SIPRequestHeaderPlarformProvider {
120 120 String callId, WWWAuthenticateHeader www , CallIdHeader callIdHeader) throws ParseException, PeerUnavailableException, InvalidArgumentException {
121 121  
122 122  
123   - Request registerRequest = createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(Request.REGISTER), fromTag, viaTag, callIdHeader);
  123 + Request registerRequest = createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(), fromTag, viaTag, callIdHeader);
124 124 SipURI requestURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerIP() + ":" + parentPlatform.getServerPort());
125 125 if (www == null) {
126 126 AuthorizationHeader authorizationHeader = sipFactory.createHeaderFactory().createAuthorizationHeader("Digest");
... ... @@ -213,7 +213,7 @@ public class SIPRequestHeaderPlarformProvider {
213 213 // Forwards
214 214 MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70);
215 215 // ceq
216   - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.MESSAGE), Request.MESSAGE);
  216 + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.MESSAGE);
217 217 MessageFactoryImpl messageFactory = (MessageFactoryImpl) sipFactory.createMessageFactory();
218 218 // 设置编码， 防止中文乱码
219 219 messageFactory.setDefaultContentEncodingCharset(parentPlatform.getCharacterSet());
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java
  View file @cb16cab
... ... @@ -2,11 +2,9 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd;
2 2  
3 3 import java.text.ParseException;
4 4 import java.util.ArrayList;
  5 +import java.util.List;
5 6  
6   -import javax.sip.Dialog;
7   -import javax.sip.InvalidArgumentException;
8   -import javax.sip.PeerUnavailableException;
9   -import javax.sip.SipFactory;
  7 +import javax.sip.*;
10 8 import javax.sip.address.Address;
11 9 import javax.sip.address.SipURI;
12 10 import javax.sip.header.*;
... ... @@ -15,7 +13,11 @@ import javax.sip.message.Request;
15 13 import com.genersoft.iot.vmp.common.StreamInfo;
16 14 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
17 15 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
  16 +import gov.nist.javax.sip.SipProviderImpl;
  17 +import gov.nist.javax.sip.SipStackImpl;
  18 +import gov.nist.javax.sip.stack.SIPDialog;
18 19 import org.springframework.beans.factory.annotation.Autowired;
  20 +import org.springframework.beans.factory.annotation.Qualifier;
19 21 import org.springframework.stereotype.Component;
20 22  
21 23 import com.genersoft.iot.vmp.conf.SipConfig;
... ... @@ -40,6 +42,14 @@ public class SIPRequestHeaderProvider {
40 42  
41 43 @Autowired
42 44 private VideoStreamSessionManager streamSession;
  45 +
  46 + @Autowired
  47 + @Qualifier(value="tcpSipProvider")
  48 + private SipProviderImpl tcpSipProvider;
  49 +
  50 + @Autowired
  51 + @Qualifier(value="udpSipProvider")
  52 + private SipProviderImpl udpSipProvider;
43 53  
44 54 public Request createMessageRequest(Device device, String content, String viaTag, String fromTag, String toTag, CallIdHeader callIdHeader) throws ParseException, InvalidArgumentException, PeerUnavailableException {
45 55 Request request = null;
... ... @@ -95,7 +105,7 @@ public class SIPRequestHeaderProvider {
95 105 MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70);
96 106  
97 107 //ceq
98   - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.INVITE), Request.INVITE);
  108 + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.INVITE);
99 109 request = sipFactory.createMessageFactory().createRequest(requestLine, Request.INVITE, callIdHeader, cSeqHeader,fromHeader, toHeader, viaHeaders, maxForwards);
100 110  
101 111 Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort()));
... ... @@ -131,7 +141,7 @@ public class SIPRequestHeaderProvider {
131 141 MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70);
132 142  
133 143 //ceq
134   - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.INVITE), Request.INVITE);
  144 + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.INVITE);
135 145 request = sipFactory.createMessageFactory().createRequest(requestLine, Request.INVITE, callIdHeader, cSeqHeader,fromHeader, toHeader, viaHeaders, maxForwards);
136 146  
137 147 Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort()));
... ... @@ -200,7 +210,7 @@ public class SIPRequestHeaderProvider {
200 210 MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70);
201 211  
202 212 // ceq
203   - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.SUBSCRIBE), Request.SUBSCRIBE);
  213 + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.SUBSCRIBE);
204 214  
205 215 request = sipFactory.createMessageFactory().createRequest(requestURI, Request.SUBSCRIBE, callIdHeader, cSeqHeader, fromHeader,
206 216 toHeader, viaHeaders, maxForwards);
... ... @@ -226,55 +236,55 @@ public class SIPRequestHeaderProvider {
226 236 }
227 237  
228 238 public Request createInfoRequest(Device device, StreamInfo streamInfo, String content)
229   - throws PeerUnavailableException, ParseException, InvalidArgumentException {
230   - Request request = null;
  239 + throws SipException, ParseException, InvalidArgumentException {
231 240 if (streamInfo == null) {
232 241 return null;
233 242 }
234   - Dialog dialog = streamSession.getDialogByStream(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream());
  243 + Request request = null;
  244 + SIPDialog dialog = streamSession.getDialogByStream(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream());
235 245 if (dialog == null) {
236 246 return null;
237 247 }
238 248  
239   - SipURI requestLine = sipFactory.createAddressFactory().createSipURI(device.getDeviceId(),
240   - device.getHostAddress());
241   - // via
242   - ArrayList<ViaHeader> viaHeaders = new ArrayList<ViaHeader>();
243   - ViaHeader viaHeader = sipFactory.createHeaderFactory().createViaHeader(device.getIp(), device.getPort(),
244   - device.getTransport(), null);
  249 + SipStack sipStack = udpSipProvider.getSipStack();
  250 + SIPDialog sipDialog = ((SipStackImpl) sipStack).putDialog(dialog);
  251 + if (dialog != sipDialog) {
  252 + dialog = sipDialog;
  253 + }else {
  254 + dialog.setSipProvider(udpSipProvider);
  255 + }
  256 + streamSession.put(streamInfo.getDeviceID(), streamInfo.getChannelId(), dialog.getCallId().getCallId(), dialog);
  257 + Request infoRequest = dialog.createRequest(Request.INFO);
  258 + SipURI sipURI = (SipURI) infoRequest.getRequestURI();
  259 + sipURI.setHost(device.getIp());
  260 + sipURI.setPort(device.getPort());
  261 + sipURI.setUser(streamInfo.getChannelId());
  262 +
  263 + ViaHeader viaHeader = (ViaHeader) infoRequest.getHeader(ViaHeader.NAME);
245 264 viaHeader.setRPort();
246   - viaHeaders.add(viaHeader);
247   - // from
248   - SipURI fromSipURI = sipFactory.createAddressFactory().createSipURI(sipConfig.getId(),
249   - sipConfig.getDomain());
250   - Address fromAddress = sipFactory.createAddressFactory().createAddress(fromSipURI);
251   - FromHeader fromHeader = sipFactory.createHeaderFactory().createFromHeader(fromAddress, dialog.getLocalTag());
252   - // to
253   - SipURI toSipURI = sipFactory.createAddressFactory().createSipURI(streamInfo.getChannelId(),
254   - sipConfig.getDomain());
255   - Address toAddress = sipFactory.createAddressFactory().createAddress(toSipURI);
256   - ToHeader toHeader = sipFactory.createHeaderFactory().createToHeader(toAddress, dialog.getRemoteTag());
257   -
258   - // callid
259   - CallIdHeader callIdHeader = dialog.getCallId();
260   -
261   - // Forwards
262   - MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70);
263   -
264   - Long cseq = redisCatchStorage.getCSEQ(Request.INVITE);
265   - // ceq
266   - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory()
267   - .createCSeqHeader(cseq, Request.INFO);
268   -
269   - request = sipFactory.createMessageFactory().createRequest(requestLine, Request.INFO, callIdHeader, cSeqHeader,
270   - fromHeader, toHeader, viaHeaders, maxForwards);
  265 + // 增加Contact header
271 266 Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory()
272 267 .createSipURI(sipConfig.getId(), sipConfig.getIp() + ":" + sipConfig.getPort()));
273   - request.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
  268 + infoRequest.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
  269 + List<String> agentParam = new ArrayList<>();
  270 + agentParam.add("wvp-pro");
  271 + // TODO 添加版本信息以及日期
  272 + UserAgentHeader userAgentHeader = null;
  273 + try {
  274 + userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam);
  275 + } catch (ParseException e) {
  276 + throw new RuntimeException(e);
  277 + }
  278 + infoRequest.addHeader(userAgentHeader);
274 279  
275 280 ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application",
276 281 "MANSRTSP");
277   - request.setContent(content, contentTypeHeader);
278   - return request;
  282 + infoRequest.setContent(content, contentTypeHeader);
  283 +
  284 + CSeqHeader cSeqHeader = (CSeqHeader)infoRequest.getHeader(CSeqHeader.NAME);
  285 + cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ());
  286 + // ceq
  287 + infoRequest.addHeader(cSeqHeader);
  288 + return infoRequest;
279 289 }
280 290 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
  View file @cb16cab
... ... @@ -727,36 +727,50 @@ public class SIPCommander implements ISIPCommander {
727 727 }
728 728 }
729 729  
730   - streamByeCmd(dialog, (SIPRequest)transaction.getRequest(), okEvent);
  730 + Request byeRequest = dialog.createRequest(Request.BYE);
  731 + SipURI byeURI = (SipURI) byeRequest.getRequestURI();
  732 + SIPRequest request = (SIPRequest)transaction.getRequest();
  733 + byeURI.setHost(request.getRemoteAddress().getHostAddress());
  734 + byeURI.setPort(request.getRemotePort());
  735 + byeURI.setUser(channelId);
  736 + ViaHeader viaHeader = (ViaHeader) byeRequest.getHeader(ViaHeader.NAME);
  737 + String protocol = viaHeader.getTransport().toUpperCase();
  738 + viaHeader.setRPort();
  739 + // 增加Contact header
  740 + Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort()));
  741 + byeRequest.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
  742 + List<String> agentParam = new ArrayList<>();
  743 + agentParam.add("wvp-pro");
  744 + // TODO 添加版本信息以及日期
  745 + UserAgentHeader userAgentHeader = null;
  746 + try {
  747 + userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam);
  748 + } catch (ParseException e) {
  749 + throw new RuntimeException(e);
  750 + }
  751 + byeRequest.addHeader(userAgentHeader);
  752 + ClientTransaction clientTransaction = null;
  753 + if("TCP".equals(protocol)) {
  754 + clientTransaction = tcpSipProvider.getNewClientTransaction(byeRequest);
  755 + } else if("UDP".equals(protocol)) {
  756 + clientTransaction = udpSipProvider.getNewClientTransaction(byeRequest);
  757 + }
  758 +
  759 + CallIdHeader callIdHeader = (CallIdHeader) byeRequest.getHeader(CallIdHeader.NAME);
  760 + if (okEvent != null) {
  761 + sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), okEvent);
  762 + }
  763 + CSeqHeader cSeqHeader = (CSeqHeader)byeRequest.getHeader(CSeqHeader.NAME);
  764 + cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ());
  765 + dialog.sendRequest(clientTransaction);
731 766  
732 767 } catch (SipException | ParseException e) {
733 768 e.printStackTrace();
  769 + } catch (InvalidArgumentException e) {
  770 + throw new RuntimeException(e);
734 771 }
735 772 }
736 773  
737   - @Override
738   - public void streamByeCmd(SIPDialog dialog, SIPRequest request, SipSubscribe.Event okEvent) throws SipException, ParseException {
739   - Request byeRequest = dialog.createRequest(Request.BYE);
740   - SipURI byeURI = (SipURI) byeRequest.getRequestURI();
741   - byeURI.setHost(request.getRemoteAddress().getHostAddress());
742   - byeURI.setPort(request.getRemotePort());
743   - ViaHeader viaHeader = (ViaHeader) byeRequest.getHeader(ViaHeader.NAME);
744   - String protocol = viaHeader.getTransport().toUpperCase();
745   - ClientTransaction clientTransaction = null;
746   - if("TCP".equals(protocol)) {
747   - clientTransaction = tcpSipProvider.getNewClientTransaction(byeRequest);
748   - } else if("UDP".equals(protocol)) {
749   - clientTransaction = udpSipProvider.getNewClientTransaction(byeRequest);
750   - }
751   -
752   - CallIdHeader callIdHeader = (CallIdHeader) byeRequest.getHeader(CallIdHeader.NAME);
753   - if (okEvent != null) {
754   - sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), okEvent);
755   - }
756   -
757   - dialog.sendRequest(clientTransaction);
758   - }
759   -
760 774 /**
761 775 * 语音广播
762 776 *
... ... @@ -1450,7 +1464,7 @@ public class SIPCommander implements ISIPCommander {
1450 1464 request.setContent(subscribePostitionXml.toString(), contentTypeHeader);
1451 1465  
1452 1466 CSeqHeader cSeqHeader = (CSeqHeader)request.getHeader(CSeqHeader.NAME);
1453   - cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ(Request.SUBSCRIBE));
  1467 + cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ());
1454 1468 request.removeHeader(CSeqHeader.NAME);
1455 1469 request.addHeader(cSeqHeader);
1456 1470 }else {
... ... @@ -1554,7 +1568,7 @@ public class SIPCommander implements ISIPCommander {
1554 1568 request.setContent(cmdXml.toString(), contentTypeHeader);
1555 1569  
1556 1570 CSeqHeader cSeqHeader = (CSeqHeader)request.getHeader(CSeqHeader.NAME);
1557   - cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ(Request.SUBSCRIBE));
  1571 + cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ());
1558 1572 request.removeHeader(CSeqHeader.NAME);
1559 1573 request.addHeader(cSeqHeader);
1560 1574  
... ... @@ -1664,10 +1678,9 @@ public class SIPCommander implements ISIPCommander {
1664 1678 @Override
1665 1679 public void playPauseCmd(Device device, StreamInfo streamInfo) {
1666 1680 try {
1667   - Long cseq = redisCatchStorage.getCSEQ(Request.INFO);
1668 1681 StringBuffer content = new StringBuffer(200);
1669 1682 content.append("PAUSE RTSP/1.0\r\n");
1670   - content.append("CSeq: " + cseq + "\r\n");
  1683 + content.append("CSeq: " + getInfoCseq() + "\r\n");
1671 1684 content.append("PauseTime: now\r\n");
1672 1685 Request request = headerProvider.createInfoRequest(device, streamInfo, content.toString());
1673 1686 if (request == null) {
... ... @@ -1695,10 +1708,9 @@ public class SIPCommander implements ISIPCommander {
1695 1708 @Override
1696 1709 public void playResumeCmd(Device device, StreamInfo streamInfo) {
1697 1710 try {
1698   - Long cseq = redisCatchStorage.getCSEQ(Request.INFO);
1699 1711 StringBuffer content = new StringBuffer(200);
1700 1712 content.append("PLAY RTSP/1.0\r\n");
1701   - content.append("CSeq: " + cseq + "\r\n");
  1713 + content.append("CSeq: " + getInfoCseq() + "\r\n");
1702 1714 content.append("Range: npt=now-\r\n");
1703 1715 Request request = headerProvider.createInfoRequest(device, streamInfo, content.toString());
1704 1716 if (request == null) {
... ... @@ -1725,10 +1737,9 @@ public class SIPCommander implements ISIPCommander {
1725 1737 @Override
1726 1738 public void playSeekCmd(Device device, StreamInfo streamInfo, long seekTime) {
1727 1739 try {
1728   - Long cseq = redisCatchStorage.getCSEQ(Request.INFO);
1729 1740 StringBuffer content = new StringBuffer(200);
1730 1741 content.append("PLAY RTSP/1.0\r\n");
1731   - content.append("CSeq: " + cseq + "\r\n");
  1742 + content.append("CSeq: " + getInfoCseq() + "\r\n");
1732 1743 content.append("Range: npt=" + Math.abs(seekTime) + "-\r\n");
1733 1744  
1734 1745 Request request = headerProvider.createInfoRequest(device, streamInfo, content.toString());
... ... @@ -1756,11 +1767,11 @@ public class SIPCommander implements ISIPCommander {
1756 1767 @Override
1757 1768 public void playSpeedCmd(Device device, StreamInfo streamInfo, Double speed) {
1758 1769 try {
1759   - Long cseq = redisCatchStorage.getCSEQ(Request.INFO);
  1770 +
1760 1771 StringBuffer content = new StringBuffer(200);
1761 1772 content.append("PLAY RTSP/1.0\r\n");
1762   - content.append("CSeq: " + cseq + "\r\n");
1763   - content.append("Scale: " + String.format("%.1f",speed) + "\r\n");
  1773 + content.append("CSeq: " + getInfoCseq() + "\r\n");
  1774 + content.append("Scale: " + String.format("%.6f",speed) + "\r\n");
1764 1775 Request request = headerProvider.createInfoRequest(device, streamInfo, content.toString());
1765 1776 if (request == null) {
1766 1777 return;
... ... @@ -1779,7 +1790,11 @@ public class SIPCommander implements ISIPCommander {
1779 1790 e.printStackTrace();
1780 1791 }
1781 1792 }
1782   -
  1793 +
  1794 + private int getInfoCseq() {
  1795 + return (int) ((Math.random() * 9 + 1) * Math.pow(10, 8));
  1796 + }
  1797 +
1783 1798 @Override
1784 1799 public void playbackControlCmd(Device device, StreamInfo streamInfo, String content,SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) {
1785 1800 try {
... ... @@ -1787,7 +1802,6 @@ public class SIPCommander implements ISIPCommander {
1787 1802 if (request == null) {
1788 1803 return;
1789 1804 }
1790   - logger.info(request.toString());
1791 1805 ClientTransaction clientTransaction = null;
1792 1806 if ("TCP".equals(device.getTransport())) {
1793 1807 clientTransaction = tcpSipProvider.getNewClientTransaction(request);
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
  View file @cb16cab
... ... @@ -105,7 +105,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
105 105 }
106 106  
107 107 request = headerProviderPlarformProvider.createRegisterRequest(parentPlatform,
108   - redisCatchStorage.getCSEQ(Request.REGISTER), "FromRegister" + tm,
  108 + redisCatchStorage.getCSEQ(), "FromRegister" + tm,
109 109 "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), callIdHeader);
110 110 // 将 callid 写入缓存， 等注册成功可以更新状态
111 111 String callIdFromHeader = callIdHeader.getCallId();
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/InviteResponseProcessor.java
  View file @cb16cab
... ... @@ -2,24 +2,32 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.response.impl;
2 2  
3 3 import com.genersoft.iot.vmp.conf.SipConfig;
4 4 import com.genersoft.iot.vmp.gb28181.SipLayer;
  5 +import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
5 6 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
6 7 import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
7 8 import com.genersoft.iot.vmp.gb28181.transmit.event.response.SIPResponseProcessorAbstract;
8 9 import gov.nist.javax.sip.ResponseEventExt;
  10 +import gov.nist.javax.sip.message.SIPResponse;
9 11 import gov.nist.javax.sip.stack.SIPDialog;
10 12 import org.slf4j.Logger;
11 13 import org.slf4j.LoggerFactory;
12 14 import org.springframework.beans.factory.annotation.Autowired;
13 15 import org.springframework.stereotype.Component;
14 16  
15   -import javax.sip.InvalidArgumentException;
16   -import javax.sip.ResponseEvent;
17   -import javax.sip.SipException;
  17 +import javax.sdp.SdpFactory;
  18 +import javax.sdp.SdpParseException;
  19 +import javax.sdp.SessionDescription;
  20 +import javax.sip.*;
  21 +import javax.sip.address.Address;
18 22 import javax.sip.address.SipURI;
19 23 import javax.sip.header.CSeqHeader;
  24 +import javax.sip.header.CallIdHeader;
  25 +import javax.sip.header.UserAgentHeader;
20 26 import javax.sip.message.Request;
21 27 import javax.sip.message.Response;
22 28 import java.text.ParseException;
  29 +import java.util.ArrayList;
  30 +import java.util.List;
23 31  
24 32  
25 33 /**
... ... @@ -34,14 +42,16 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
34 42 private final String method = "INVITE";
35 43  
36 44 @Autowired
37   - private SipLayer sipLayer;
  45 + private VideoStreamSessionManager streamSession;
38 46  
39 47 @Autowired
40   - private SipConfig config;
  48 + private SIPProcessorObserver sipProcessorObserver;
41 49  
  50 + @Autowired
  51 + private SipConfig sipConfig;
42 52  
43 53 @Autowired
44   - private SIPProcessorObserver sipProcessorObserver;
  54 + private SipFactory sipFactory;
45 55  
46 56 @Override
47 57 public void afterPropertiesSet() throws Exception {
... ... @@ -49,8 +59,7 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
49 59 sipProcessorObserver.addResponseProcessor(method, this);
50 60 }
51 61  
52   - @Autowired
53   - private VideoStreamSessionManager streamSession;
  62 +
54 63  
55 64 /**
56 65 * 处理invite响应
... ... @@ -74,6 +83,19 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
74 83 CSeqHeader cseq = (CSeqHeader) response.getHeader(CSeqHeader.NAME);
75 84 Request reqAck = dialog.createAck(cseq.getSeqNumber());
76 85 SipURI requestURI = (SipURI) reqAck.getRequestURI();
  86 + String contentString = new String(response.getRawContent());
  87 + // jainSip不支持y=字段， 移除以解析。
  88 + int ssrcIndex = contentString.indexOf("y=");
  89 + // 检查是否有y字段
  90 + SessionDescription sdp;
  91 + if (ssrcIndex >= 0) {
  92 + //ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段
  93 + String substring = contentString.substring(0, contentString.indexOf("y="));
  94 + sdp = SdpFactory.getInstance().createSessionDescription(substring);
  95 + } else {
  96 + sdp = SdpFactory.getInstance().createSessionDescription(contentString);
  97 + }
  98 + requestURI.setUser(sdp.getOrigin().getUsername());
77 99 try {
78 100 requestURI.setHost(event.getRemoteIpAddress());
79 101 } catch (ParseException e) {
... ... @@ -81,6 +103,18 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
81 103 }
82 104 requestURI.setPort(event.getRemotePort());
83 105 reqAck.setRequestURI(requestURI);
  106 + List<String> agentParam = new ArrayList<>();
  107 + agentParam.add("wvp-pro");
  108 + // TODO 添加版本信息以及日期
  109 + UserAgentHeader userAgentHeader = null;
  110 + try {
  111 + userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam);
  112 + } catch (ParseException e) {
  113 + throw new RuntimeException(e);
  114 + }
  115 + reqAck.addHeader(userAgentHeader);
  116 + Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort()));
  117 + reqAck.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
84 118 logger.info("[回复ack] {}-> {}:{} ",requestURI, event.getRemoteIpAddress(), event.getRemotePort());
85 119  
86 120 dialog.sendAck(reqAck);
... ... @@ -88,6 +122,10 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
88 122 }
89 123 } catch (InvalidArgumentException | SipException e) {
90 124 e.printStackTrace();
  125 + } catch (ParseException e) {
  126 + throw new RuntimeException(e);
  127 + } catch (SdpParseException e) {
  128 + throw new RuntimeException(e);
91 129 }
92 130 }
93 131  
... ...
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
  View file @cb16cab
... ... @@ -98,9 +98,7 @@ public class ZLMHttpHookListener {
98 98 @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
99 99 public ResponseEntity<String> onServerKeepalive(@RequestBody JSONObject json){
100 100  
101   - if (logger.isDebugEnabled()) {
102   - logger.debug("[ ZLM HOOK ] on_server_keepalive API调用，参数：" + json.toString());
103   - }
  101 + logger.info("[ ZLM HOOK ] on_server_keepalive API调用，参数：" + json.toString());
104 102 String mediaServerId = json.getString("mediaServerId");
105 103 List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive);
106 104 if (subscribes != null && subscribes.size() > 0) {
... ... @@ -445,12 +443,15 @@ public class ZLMHttpHookListener {
445 443 if (streamInfo!=null){
446 444 redisCatchStorage.stopPlay(streamInfo);
447 445 storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
  446 + // 如果正在给上级推送，则发送bye
  447 +
448 448 }else{
449 449 streamInfo = redisCatchStorage.queryPlayback(null, null, stream, null);
450 450 if (streamInfo != null) {
451 451 redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
452 452 streamInfo.getStream(), null);
453 453 }
  454 + // 如果正在给上级推送，则发送bye
454 455 }
455 456 }else {
456 457 if (!"rtp".equals(app)){
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
  View file @cb16cab
... ... @@ -38,7 +38,6 @@ import com.genersoft.iot.vmp.service.IMediaServerService;
38 38 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
39 39 import com.genersoft.iot.vmp.storager.dao.MediaServerMapper;
40 40 import com.genersoft.iot.vmp.utils.DateUtil;
41   -import com.genersoft.iot.vmp.utils.redis.JedisUtil;
42 41 import com.genersoft.iot.vmp.utils.redis.RedisUtil;
43 42 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
44 43  
... ... @@ -101,9 +100,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
101 100 @Autowired
102 101 private EventPublisher publisher;
103 102  
104   - @Autowired
105   - JedisUtil jedisUtil;
106   -
107 103 /**
108 104 * 初始化
109 105 */
... ... @@ -291,13 +287,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
291 287 return null;
292 288 }
293 289 String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;
294   - MediaServerItem serverItem=(MediaServerItem)redisUtil.get(key);
295   - if(null==serverItem){
296   - //zlm服务不在线，启动重连
297   - reloadZlm();
298   - serverItem=(MediaServerItem)redisUtil.get(key);
299   - }
300   - return serverItem;
  290 + return (MediaServerItem)redisUtil.get(key);
301 291 }
302 292  
303 293 @Override
... ... @@ -426,7 +416,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
426 416 }
427 417 redisUtil.set(key, serverItem);
428 418 resetOnlineServerItem(serverItem);
429   - updateMediaServerKeepalive(serverItem.getId(), null);
430 419 if (serverItem.isAutoConfig()) {
431 420 setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable()));
432 421 }
... ... @@ -490,9 +479,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
490 479 String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
491 480  
492 481 if (redisUtil.zSize(key) == null || redisUtil.zSize(key) == 0) {
493   - logger.info("获取负载最低的节点时无在线节点，启动重连机制");
494   - //启动重连
495   - reloadZlm();
496 482 if (redisUtil.zSize(key) == null || redisUtil.zSize(key) == 0) {
497 483 logger.info("获取负载最低的节点时无在线节点");
498 484 return null;
... ... @@ -657,6 +643,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
657 643 public void updateMediaServerKeepalive(String mediaServerId, JSONObject data) {
658 644 MediaServerItem mediaServerItem = getOne(mediaServerId);
659 645 if (mediaServerItem == null) {
  646 + // 缓存不存在，从数据库查询，如果数据库不存在则是错误的
  647 + MediaServerItem mediaServerItemFromDatabase = getOneFromDatabase(mediaServerId);
  648 + if (mediaServerItemFromDatabase == null) {
  649 + return;
  650 + }
660 651 // zlm连接重试
661 652 logger.warn("[更新ZLM 保活信息]失败，未找到流媒体信息,尝试重连zlm");
662 653 reloadZlm();
... ... @@ -672,6 +663,10 @@ public class MediaServerServiceImpl implements IMediaServerService {
672 663 redisUtil.set(key, data, hookAliveInterval);
673 664 }
674 665  
  666 + private MediaServerItem getOneFromDatabase(String mediaServerId) {
  667 + return mediaServerMapper.queryOne(mediaServerId);
  668 + }
  669 +
675 670 @Override
676 671 public void syncCatchFromDatabase() {
677 672 List<MediaServerItem> allInCatch = getAll();
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
  View file @cb16cab
... ... @@ -2,9 +2,7 @@ package com.genersoft.iot.vmp.service.impl;
2 2  
3 3 import java.math.BigDecimal;
4 4 import java.math.RoundingMode;
5   -import java.util.List;
6   -import java.util.Objects;
7   -import java.util.UUID;
  5 +import java.util.*;
8 6  
9 7 import javax.sip.ResponseEvent;
10 8  
... ... @@ -12,8 +10,10 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
12 10 import org.slf4j.Logger;
13 11 import org.slf4j.LoggerFactory;
14 12 import org.springframework.beans.factory.annotation.Autowired;
  13 +import org.springframework.beans.factory.annotation.Qualifier;
15 14 import org.springframework.http.HttpStatus;
16 15 import org.springframework.http.ResponseEntity;
  16 +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
17 17 import org.springframework.stereotype.Service;
18 18 import org.springframework.web.context.request.async.DeferredResult;
19 19  
... ... @@ -131,6 +131,10 @@ public class PlayServiceImpl implements IPlayService {
131 131 private ZLMHttpHookSubscribe subscribe;
132 132  
133 133  
  134 + @Qualifier("taskExecutor")
  135 + @Autowired
  136 + private ThreadPoolTaskExecutor taskExecutor;
  137 +
134 138  
135 139  
136 140 @Override
... ... @@ -162,21 +166,23 @@ public class PlayServiceImpl implements IPlayService {
162 166  
163 167 result.onCompletion(()->{
164 168 // 点播结束时调用截图接口
165   - // TODO 应该在上流时调用更好，结束也可能是错误结束
166   - String path = "snap";
167   - String fileName = deviceId + "_" + channelId + ".jpg";
168   - ResponseEntity responseEntity = (ResponseEntity)result.getResult();
169   - if (responseEntity != null && responseEntity.getStatusCode() == HttpStatus.OK) {
170   - WVPResult wvpResult = (WVPResult)responseEntity.getBody();
171   - if (Objects.requireNonNull(wvpResult).getCode() == 0) {
172   - StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData();
173   - MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId());
174   - String streamUrl = streamInfoForSuccess.getFmp4();
175   - // 请求截图
176   - logger.info("[请求截图]: " + fileName);
177   - zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName);
  169 + taskExecutor.execute(()->{
  170 + // TODO 应该在上流时调用更好，结束也可能是错误结束
  171 + String path = "snap";
  172 + String fileName = deviceId + "_" + channelId + ".jpg";
  173 + ResponseEntity responseEntity = (ResponseEntity)result.getResult();
  174 + if (responseEntity != null && responseEntity.getStatusCode() == HttpStatus.OK) {
  175 + WVPResult wvpResult = (WVPResult)responseEntity.getBody();
  176 + if (Objects.requireNonNull(wvpResult).getCode() == 0) {
  177 + StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData();
  178 + MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId());
  179 + String streamUrl = streamInfoForSuccess.getFmp4();
  180 + // 请求截图
  181 + logger.info("[请求截图]: " + fileName);
  182 + zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName);
  183 + }
178 184 }
179   - }
  185 + });
180 186 });
181 187 if (streamInfo != null) {
182 188 String streamId = streamInfo.getStream();
... ... @@ -759,6 +765,53 @@ public class PlayServiceImpl implements IPlayService {
759 765  
760 766 @Override
761 767 public void zlmServerOnline(String mediaServerId) {
762   - // 似乎没啥需要做的
  768 + // TODO 查找之前的点播，流如果不存在则给下级发送bye
  769 +// MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
  770 +// zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
  771 +// Integer code = mediaList.getInteger("code");
  772 +// if (code == 0) {
  773 +// JSONArray data = mediaList.getJSONArray("data");
  774 +// if (data == null || data.size() == 0) {
  775 +// zlmServerOffline(mediaServerId);
  776 +// }else {
  777 +// Map<String, JSONObject> mediaListMap = new HashMap<>();
  778 +// for (int i = 0; i < data.size(); i++) {
  779 +// JSONObject json = data.getJSONObject(i);
  780 +// String app = json.getString("app");
  781 +// if ("rtp".equals(app)) {
  782 +// String stream = json.getString("stream");
  783 +// if (mediaListMap.get(stream) != null) {
  784 +// continue;
  785 +// }
  786 +// mediaListMap.put(stream, json);
  787 +// // 处理正在观看的国标设备
  788 +// List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(null, null, null, stream);
  789 +// if (ssrcTransactions.size() > 0) {
  790 +// for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
  791 +// if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
  792 +// cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(),
  793 +// ssrcTransaction.getStream(), null);
  794 +// }
  795 +// }
  796 +// }
  797 +// }
  798 +// }
  799 +// if (mediaListMap.size() > 0 ) {
  800 +// // 处理正在向上推流的上级平台
  801 +// List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(null);
  802 +// if (sendRtpItems.size() > 0) {
  803 +// for (SendRtpItem sendRtpItem : sendRtpItems) {
  804 +// if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
  805 +// if (mediaListMap.get(sendRtpItem.getStreamId()) == null) {
  806 +// ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
  807 +// sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
  808 +// }
  809 +// }
  810 +// }
  811 +// }
  812 +// }
  813 +// }
  814 +// }
  815 +// }));
763 816 }
764 817 }
... ...
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
  View file @cb16cab
... ... @@ -17,10 +17,9 @@ public interface IRedisCatchStorage {
17 17 /**
18 18 * 计数器。为cseq进行计数
19 19 *
20   - * @param method sip 方法
21 20 * @return
22 21 */
23   - Long getCSEQ(String method);
  22 + Long getCSEQ();
24 23  
25 24 /**
26 25 * 开始播放时将流存入
... ...
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
  View file @cb16cab
... ... @@ -42,8 +42,8 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
42 42 private UserSetting userSetting;
43 43  
44 44 @Override
45   - public Long getCSEQ(String method) {
46   - String key = VideoManagerConstants.SIP_CSEQ_PREFIX + userSetting.getServerId() + "_" + method;
  45 + public Long getCSEQ() {
  46 + String key = VideoManagerConstants.SIP_CSEQ_PREFIX + userSetting.getServerId();
47 47  
48 48 long result = redis.incr(key, 1L);
49 49 if (result > Integer.MAX_VALUE) {
... ...
src/main/java/com/genersoft/iot/vmp/utils/redis/JedisUtil.java deleted 100644 → 0
  View file @116d979
1   -package com.genersoft.iot.vmp.utils.redis;
2   -
3   -import org.springframework.beans.factory.annotation.Autowired;
4   -import org.springframework.stereotype.Component;
5   -import redis.clients.jedis.Jedis;
6   -import redis.clients.jedis.JedisPool;
7   -
8   -import java.util.Set;
9   -
10   -/**
11   - * @description:Jedis工具类
12   - * @author: wangshaopeng@sunnybs.com
13   - * @date: 2021年03月22日 下午8:27:29
14   - */
15   -@Component
16   -public class JedisUtil {
17   -
18   - @Autowired
19   - private JedisPool jedisPool;
20   -
21   - // ============================== Key ==============================
22   -
23   - /**
24   - * 检查给定 key 是否存在。
25   - *
26   - * @param key
27   - * @return
28   - */
29   - public Boolean exists(String key) {
30   - Jedis jedis = null;
31   - try {
32   - jedis = jedisPool.getResource();
33   - Boolean exists = jedis.exists(key);
34   - return exists;
35   - } finally {
36   - returnToPool(jedis);
37   - }
38   - }
39   -
40   -
41   - // ============================== Set ==============================
42   -
43   - /**
44   - * SADD key member [member ...]
45   - * 将一个或多个 member 元素加入到集合 key 当中，已经存在于集合的 member 元素将被忽略。
46   - * 假如 key 不存在，则创建一个只包含 member 元素作成员的集合。
47   - * 当 key 不是集合类型时，返回一个错误。
48   - */
49   - public Long sadd(String key, String... members) {
50   - Jedis jedis = null;
51   - try {
52   - jedis = jedisPool.getResource();
53   - Long smove = jedis.sadd(key, members);
54   - return smove;
55   - } finally {
56   - returnToPool(jedis);
57   - }
58   - }
59   -
60   - /**
61   - * SMEMBERS key
62   - * 返回集合 key 中的所有成员。
63   - * 不存在的 key 被视为空集合。
64   - */
65   - public Set<String> smembers(String key) {
66   - Jedis jedis = null;
67   - try {
68   - jedis = jedisPool.getResource();
69   - Set<String> smembers = jedis.smembers(key);
70   - return smembers;
71   - } finally {
72   - returnToPool(jedis);
73   - }
74   - }
75   -
76   -
77   - /**
78   - * SREM key member1 [member2]
79   - * 移除集合中一个或多个成员
80   - */
81   - public Long srem(String key, String... member) {
82   - Jedis jedis = null;
83   - try {
84   - jedis = jedisPool.getResource();
85   - Long srem = jedis.srem(key, member);
86   - return srem;
87   - } finally {
88   - returnToPool(jedis);
89   - }
90   - }
91   -
92   - private void returnToPool(Jedis jedis) {
93   - if (jedis != null) {
94   - jedis.close();
95   - }
96   - }
97   -}
98 0 \ No newline at end of file
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
  View file @cb16cab
... ... @@ -152,6 +152,8 @@ public class PlayController {
152 152 // 超时处理
153 153 result.onTimeout(()->{
154 154 logger.warn(String.format("设备预览/回放停止超时，deviceId/channelId：%s_%s ", deviceId, channelId));
  155 + redisCatchStorage.stopPlay(streamInfo);
  156 + storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
155 157 RequestMessage msg = new RequestMessage();
156 158 msg.setId(uuid);
157 159 msg.setKey(key);
... ...