From 871b4e5359aaaeaa5b398b1a4ea552a435468dcc Mon Sep 17 00:00:00 2001 From: chenjialing <595168663@qq.com> Date: Tue, 9 Aug 2022 17:26:32 +0800 Subject: [PATCH 1/5] =?UTF-8?q?=E4=BC=98=E5=8C=96--=E7=9B=B4=E6=92=AD?= =?UTF-8?q?=E7=BB=93=E6=9D=9F=E5=BC=82=E6=AD=A5=E5=A4=84=E7=90=86=E6=88=AA?= =?UTF-8?q?=E5=9B=BE=E6=93=8D=E4=BD=9C=EF=BC=8C=E6=8F=90=E9=AB=98=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E5=93=8D=E5=BA=94=E9=80=9F=E5=BA=A6=E5=92=8C=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E8=83=BD=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/service/impl/PlayServiceImpl.java | 36 +++++++++++-------- 1 file changed, 22 insertions(+), 14 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java index e6a8f6a7..96acbf6e 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java @@ -11,8 +11,10 @@ import javax.sip.ResponseEvent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; import org.springframework.web.context.request.async.DeferredResult; @@ -104,6 +106,10 @@ public class PlayServiceImpl implements IPlayService { private ZLMHttpHookSubscribe subscribe; + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + @Override @@ -135,21 +141,23 @@ public class PlayServiceImpl implements IPlayService { result.onCompletion(()->{ // 点播结束时调用截图接口 - // TODO 应该在上流时调用更好，结束也可能是错误结束 - String path = "snap"; - String fileName = deviceId + "_" + channelId + ".jpg"; - ResponseEntity responseEntity = (ResponseEntity)result.getResult(); - if (responseEntity != null && responseEntity.getStatusCode() == HttpStatus.OK) { - WVPResult wvpResult = (WVPResult)responseEntity.getBody(); - if (Objects.requireNonNull(wvpResult).getCode() == 0) { - StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData(); - MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId()); - String streamUrl = streamInfoForSuccess.getFmp4(); - // 请求截图 - logger.info("[请求截图]: " + fileName); - zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName); + taskExecutor.execute(()->{ + // TODO 应该在上流时调用更好，结束也可能是错误结束 + String path = "snap"; + String fileName = deviceId + "_" + channelId + ".jpg"; + ResponseEntity responseEntity = (ResponseEntity)result.getResult(); + if (responseEntity != null && responseEntity.getStatusCode() == HttpStatus.OK) { + WVPResult wvpResult = (WVPResult)responseEntity.getBody(); + if (Objects.requireNonNull(wvpResult).getCode() == 0) { + StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData(); + MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId()); + String streamUrl = streamInfoForSuccess.getFmp4(); + // 请求截图 + logger.info("[请求截图]: " + fileName); + zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName); + } } - } + }); }); if (streamInfo != null) { String streamId = streamInfo.getStream(); From 82ab6b1760b8787e31bed4020a0831c64fa05077 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Wed, 10 Aug 2022 18:14:33 +0800 Subject: [PATCH 2/5] =?UTF-8?q?=E5=8E=BB=E9=99=A4jedis=EF=BC=8C=E6=96=B9?= =?UTF-8?q?=E4=BE=BF=E6=94=AF=E6=8C=81redis=E9=9B=86=E7=BE=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pom.xml | 12 --- .../genersoft/iot/vmp/conf/RedisConfig.java | 67 ++++--------- ...edisKeyExpirationEventMessageListener.java | 2 +- .../vmp/media/zlm/ZLMHttpHookListener.java | 3 + .../service/impl/MediaServerServiceImpl.java | 4 - .../iot/vmp/utils/redis/JedisUtil.java | 97 ------------------- .../vmanager/gb28181/play/PlayController.java | 2 + 7 files changed, 25 insertions(+), 162 deletions(-) delete mode 100644 src/main/java/com/genersoft/iot/vmp/utils/redis/JedisUtil.java diff --git a/pom.xml b/pom.xml index 41be5675..2abd9d8c 100644 --- a/pom.xml +++ b/pom.xml @@ -61,13 +61,6 @@ org.springframework.boot spring-boot-starter-data-redis - - - - io.lettuce - lettuce-core - - org.springframework.boot @@ -94,11 +87,6 @@ spring-boot-starter-security - - redis.clients - jedis - - com.alibaba diff --git a/src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java index 6a862ae3..d2e1347e 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java @@ -1,5 +1,6 @@ package com.genersoft.iot.vmp.conf; +import com.alibaba.fastjson.parser.ParserConfig; import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.service.impl.*; import org.apache.commons.lang3.StringUtils; @@ -9,15 +10,14 @@ import org.springframework.cache.annotation.CachingConfigurerSupport; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.serializer.StringRedisSerializer; -import com.alibaba.fastjson.parser.ParserConfig; import com.genersoft.iot.vmp.utils.redis.FastJsonRedisSerializer; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.JedisPoolConfig; + /** * @description:Redis中间件配置类，使用spring-data-redis集成，自动从application.yml中加载redis配置 @@ -28,23 +28,6 @@ import redis.clients.jedis.JedisPoolConfig; @Configuration public class RedisConfig extends CachingConfigurerSupport { - @Value("${spring.redis.host}") - private String host; - @Value("${spring.redis.port}") - private int port; - @Value("${spring.redis.database}") - private int database; - @Value("${spring.redis.password}") - private String password; - @Value("${spring.redis.timeout}") - private int timeout; - @Value("${spring.redis.poolMaxTotal:1000}") - private int poolMaxTotal; - @Value("${spring.redis.poolMaxIdle:500}") - private int poolMaxIdle; - @Value("${spring.redis.poolMaxWait:5}") - private int poolMaxWait; - @Autowired private RedisGpsMsgListener redisGPSMsgListener; @@ -61,36 +44,24 @@ public class RedisConfig extends CachingConfigurerSupport { private RedisPushStreamStatusMsgListener redisPushStreamStatusMsgListener; @Bean - public JedisPool jedisPool() { - if (StringUtils.isBlank(password)) { - password = null; - } - JedisPoolConfig poolConfig = new JedisPoolConfig(); - poolConfig.setMaxIdle(poolMaxIdle); - poolConfig.setMaxTotal(poolMaxTotal); - // 秒转毫秒 - poolConfig.setMaxWaitMillis(poolMaxWait * 1000L); - JedisPool jp = new JedisPool(poolConfig, host, port, timeout * 1000, password, database); - return jp; + public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) { + RedisTemplate redisTemplate = new RedisTemplate<>(); + // 使用fastJson序列化 + FastJsonRedisSerializer fastJsonRedisSerializer = new FastJsonRedisSerializer(Object.class); + // value值的序列化采用fastJsonRedisSerializer + redisTemplate.setValueSerializer(fastJsonRedisSerializer); + redisTemplate.setHashValueSerializer(fastJsonRedisSerializer); + // 全局开启AutoType，不建议使用 + ParserConfig.getGlobalInstance().setAutoTypeSupport(true); + // 建议使用这种方式，小范围指定白名单，需要序列化的类 +// ParserConfig.getGlobalInstance().addAccept("com.avatar"); + // key的序列化采用StringRedisSerializer + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setHashKeySerializer(new StringRedisSerializer()); + redisTemplate.setConnectionFactory(redisConnectionFactory); + return redisTemplate; } - @Bean("redisTemplate") - public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) { - RedisTemplate template = new RedisTemplate<>(); - template.setConnectionFactory(redisConnectionFactory); - // 使用fastjson进行序列化处理，提高解析效率 - FastJsonRedisSerializer serializer = new FastJsonRedisSerializer(Object.class); - // value值的序列化采用fastJsonRedisSerializer - template.setValueSerializer(serializer); - template.setHashValueSerializer(serializer); - // key的序列化采用StringRedisSerializer - template.setKeySerializer(new StringRedisSerializer()); - template.setHashKeySerializer(new StringRedisSerializer()); - template.setConnectionFactory(redisConnectionFactory); - // 使用fastjson时需设置此项，否则会报异常not support type - ParserConfig.getGlobalInstance().setAutoTypeSupport(true); - return template; - } /** * redis消息监听器容器 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器 diff --git a/src/main/java/com/genersoft/iot/vmp/conf/RedisKeyExpirationEventMessageListener.java b/src/main/java/com/genersoft/iot/vmp/conf/RedisKeyExpirationEventMessageListener.java index 18483800..ef4a6172 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/RedisKeyExpirationEventMessageListener.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/RedisKeyExpirationEventMessageListener.java @@ -28,7 +28,7 @@ public class RedisKeyExpirationEventMessageListener extends KeyExpirationEventMe RedisConnection connection = this.listenerContainer.getConnectionFactory().getConnection(); Properties config = connection.getConfig("notify-keyspace-events"); try { - if (!config.getProperty("notify-keyspace-events").equals(keyspaceNotificationsConfigParameter)) { + if (!keyspaceNotificationsConfigParameter.equals(config.getProperty("notify-keyspace-events"))) { connection.setConfig("notify-keyspace-events", keyspaceNotificationsConfigParameter); } } finally { diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java index 0dc6bd38..57ad192f 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -445,12 +445,15 @@ public class ZLMHttpHookListener { if (streamInfo!=null){ redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); + // 如果正在给上级推送，则发送bye + }else{ streamInfo = redisCatchStorage.queryPlayback(null, null, stream, null); if (streamInfo != null) { redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream(), null); } + // 如果正在给上级推送，则发送bye } }else { if (!"rtp".equals(app)){ diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java index 59749189..aded98d5 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java @@ -36,7 +36,6 @@ import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.storager.dao.MediaServerMapper; import com.genersoft.iot.vmp.utils.DateUtil; -import com.genersoft.iot.vmp.utils.redis.JedisUtil; import com.genersoft.iot.vmp.utils.redis.RedisUtil; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; @@ -91,9 +90,6 @@ public class MediaServerServiceImpl implements IMediaServerService { @Autowired private EventPublisher publisher; - @Autowired - JedisUtil jedisUtil; - /** * 初始化 */ diff --git a/src/main/java/com/genersoft/iot/vmp/utils/redis/JedisUtil.java b/src/main/java/com/genersoft/iot/vmp/utils/redis/JedisUtil.java deleted file mode 100644 index 1fb1d17e..00000000 --- a/src/main/java/com/genersoft/iot/vmp/utils/redis/JedisUtil.java +++ /dev/null @@ -1,97 +0,0 @@ -package com.genersoft.iot.vmp.utils.redis; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.stereotype.Component; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisPool; - -import java.util.Set; - -/** - * @description:Jedis工具类 - * @author: wangshaopeng@sunnybs.com - * @date: 2021年03月22日 下午8:27:29 - */ -@Component -public class JedisUtil { - - @Autowired - private JedisPool jedisPool; - - // ============================== Key ============================== - - /** - * 检查给定 key 是否存在。 - * - * @param key - * @return - */ - public Boolean exists(String key) { - Jedis jedis = null; - try { - jedis = jedisPool.getResource(); - Boolean exists = jedis.exists(key); - return exists; - } finally { - returnToPool(jedis); - } - } - - - // ============================== Set ============================== - - /** - * SADD key member [member ...] - * 将一个或多个 member 元素加入到集合 key 当中，已经存在于集合的 member 元素将被忽略。 - * 假如 key 不存在，则创建一个只包含 member 元素作成员的集合。 - * 当 key 不是集合类型时，返回一个错误。 - */ - public Long sadd(String key, String... members) { - Jedis jedis = null; - try { - jedis = jedisPool.getResource(); - Long smove = jedis.sadd(key, members); - return smove; - } finally { - returnToPool(jedis); - } - } - - /** - * SMEMBERS key - * 返回集合 key 中的所有成员。 - * 不存在的 key 被视为空集合。 - */ - public Set smembers(String key) { - Jedis jedis = null; - try { - jedis = jedisPool.getResource(); - Set smembers = jedis.smembers(key); - return smembers; - } finally { - returnToPool(jedis); - } - } - - - /** - * SREM key member1 [member2] - * 移除集合中一个或多个成员 - */ - public Long srem(String key, String... member) { - Jedis jedis = null; - try { - jedis = jedisPool.getResource(); - Long srem = jedis.srem(key, member); - return srem; - } finally { - returnToPool(jedis); - } - } - - private void returnToPool(Jedis jedis) { - if (jedis != null) { - jedis.close(); - } - } -} \ No newline at end of file diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java index db7f8fdc..cc6d4165 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java @@ -148,6 +148,8 @@ public class PlayController { // 超时处理 result.onTimeout(()->{ logger.warn(String.format("设备预览/回放停止超时，deviceId/channelId：%s_%s ", deviceId, channelId)); + redisCatchStorage.stopPlay(streamInfo); + storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); RequestMessage msg = new RequestMessage(); msg.setId(uuid); msg.setKey(key); From dab04fcab717350e4955a78324c49a6ca1cf2f1e Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Thu, 11 Aug 2022 16:37:55 +0800 Subject: [PATCH 3/5] =?UTF-8?q?=E4=BC=98=E5=8C=96=E4=BD=BF=E7=94=A8tcp?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E6=97=B6=E7=9A=84=E8=B6=85=E6=97=B6=E6=97=B6?= =?UTF-8?q?=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../genersoft/iot/vmp/gb28181/SipLayer.java | 13 +++-- .../iot/vmp/service/impl/PlayServiceImpl.java | 53 +++++++++++++++++-- 2 files changed, 58 insertions(+), 8 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java b/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java index a3428b10..c1811bf4 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java @@ -53,10 +53,15 @@ public class SipLayer{ * gov/nist/javax/sip/SipStackImpl.class */ properties.setProperty("gov.nist.javax.sip.LOG_MESSAGE_CONTENT", "true"); - properties.setProperty("gov.nist.javax.sip.DELIVER_UNSOLICITED_NOTIFY", "true"); // 接收所有notify请求，即使没有订阅 - properties.setProperty("gov.nist.javax.sip.DELIVER_TERMINATED_EVENT_FOR_NULL_DIALOG", "true"); // 为_NULL _对话框传递_终止的_事件 - properties.setProperty("gov.nist.javax.sip.RELEASE_REFERENCES_STRATEGY", "Normal"); // 会话清理策略 - properties.setProperty("gov.nist.javax.sip.RELIABLE_CONNECTION_KEEP_ALIVE_TIMEOUT", "10"); + // 接收所有notify请求，即使没有订阅 + properties.setProperty("gov.nist.javax.sip.DELIVER_UNSOLICITED_NOTIFY", "true"); + // 为_NULL _对话框传递_终止的_事件 + properties.setProperty("gov.nist.javax.sip.DELIVER_TERMINATED_EVENT_FOR_NULL_DIALOG", "true"); + // 会话清理策略 + properties.setProperty("gov.nist.javax.sip.RELEASE_REFERENCES_STRATEGY", "Normal"); + // 处理由该服务器处理的基于底层TCP的保持生存超时 + properties.setProperty("gov.nist.javax.sip.RELIABLE_CONNECTION_KEEP_ALIVE_TIMEOUT", "60"); + /** * sip_server_log.log 和 sip_debug_log.log public static final int TRACE_NONE = * 0; public static final int TRACE_MESSAGES = 16; public static final int diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java index 96acbf6e..0c571e98 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java @@ -2,9 +2,7 @@ package com.genersoft.iot.vmp.service.impl; import java.math.BigDecimal; import java.math.RoundingMode; -import java.util.List; -import java.util.Objects; -import java.util.UUID; +import java.util.*; import javax.sip.ResponseEvent; @@ -659,6 +657,53 @@ public class PlayServiceImpl implements IPlayService { @Override public void zlmServerOnline(String mediaServerId) { - // 似乎没啥需要做的 + // TODO 查找之前的点播，流如果不存在则给下级发送bye +// MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); +// zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{ +// Integer code = mediaList.getInteger("code"); +// if (code == 0) { +// JSONArray data = mediaList.getJSONArray("data"); +// if (data == null || data.size() == 0) { +// zlmServerOffline(mediaServerId); +// }else { +// Map mediaListMap = new HashMap<>(); +// for (int i = 0; i < data.size(); i++) { +// JSONObject json = data.getJSONObject(i); +// String app = json.getString("app"); +// if ("rtp".equals(app)) { +// String stream = json.getString("stream"); +// if (mediaListMap.get(stream) != null) { +// continue; +// } +// mediaListMap.put(stream, json); +// // 处理正在观看的国标设备 +// List ssrcTransactions = streamSession.getSsrcTransactionForAll(null, null, null, stream); +// if (ssrcTransactions.size() > 0) { +// for (SsrcTransaction ssrcTransaction : ssrcTransactions) { +// if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) { +// cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), +// ssrcTransaction.getStream(), null); +// } +// } +// } +// } +// } +// if (mediaListMap.size() > 0 ) { +// // 处理正在向上推流的上级平台 +// List sendRtpItems = redisCatchStorage.querySendRTPServer(null); +// if (sendRtpItems.size() > 0) { +// for (SendRtpItem sendRtpItem : sendRtpItems) { +// if (sendRtpItem.getMediaServerId().equals(mediaServerId)) { +// if (mediaListMap.get(sendRtpItem.getStreamId()) == null) { +// ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); +// sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId()); +// } +// } +// } +// } +// } +// } +// } +// })); } } From 2591997dfc8995a788b07dd41f42aac77c6fc4fb Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Fri, 12 Aug 2022 10:00:18 +0800 Subject: [PATCH 4/5] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=96=87=E6=A1=A3?= =?UTF-8?q?=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- doc/_content/introduction/deployment.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/_content/introduction/deployment.md b/doc/_content/introduction/deployment.md index cdc1568e..388fd357 100644 --- a/doc/_content/introduction/deployment.md +++ b/doc/_content/introduction/deployment.md @@ -24,7 +24,7 @@ 7. 启动服务，以linux为例 **启动WVP-PRO** ```shell -nohup java -jar java -jar wvp-pro-*.jar & +nohup java -jar wvp-pro-*.jar & ``` **启动ZLM** From c041aaccb43baf73d1a9f5b50fb5a1f410b88523 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 15 Aug 2022 15:08:51 +0800 Subject: [PATCH 5/5] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BD=95=E5=83=8F?= =?UTF-8?q?=E5=9B=9E=E6=94=BE=E4=B8=AD=E7=9A=84=E4=BF=A1=E4=BB=A4=E9=94=99?= =?UTF-8?q?=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/common/VideoManagerConstants.java | 40 +++++--- .../cmd/SIPRequestHeaderPlarformProvider.java | 6 +- .../cmd/SIPRequestHeaderProvider.java | 98 ++++++++++--------- .../transmit/cmd/impl/SIPCommander.java | 44 ++++++--- .../cmd/impl/SIPCommanderFroPlatform.java | 2 +- .../impl/InviteResponseProcessor.java | 58 +++++++++-- .../vmp/media/zlm/ZLMHttpHookListener.java | 4 +- .../service/impl/MediaServerServiceImpl.java | 21 ++-- .../iot/vmp/storager/IRedisCatchStorage.java | 3 +- .../storager/impl/RedisCatchStorageImpl.java | 4 +- 10 files changed, 179 insertions(+), 101 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index 40a73521..510b5b22 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java @@ -77,38 +77,54 @@ public class VideoManagerConstants { //************************** redis 消息********************************* - // 流变化的通知 + /** + * 流变化的通知 + */ public static final String WVP_MSG_STREAM_CHANGE_PREFIX = "WVP_MSG_STREAM_CHANGE_"; - // 接收推流设备的GPS变化通知 + /** + * 接收推流设备的GPS变化通知 + */ public static final String VM_MSG_GPS = "VM_MSG_GPS"; - // 接收推流设备的GPS变化通知 + /** + * 接收推流设备的GPS变化通知 + */ public static final String VM_MSG_PUSH_STREAM_STATUS_CHANGE = "VM_MSG_PUSH_STREAM_STATUS_CHANGE"; - // redis 消息通知设备推流到平台 + /** + * redis 消息通知设备推流到平台 + */ public static final String VM_MSG_STREAM_PUSH_REQUESTED = "VM_MSG_STREAM_PUSH_REQUESTED"; - // redis 消息请求所有的在线通道 + /** + * redis 消息请求所有的在线通道 + */ public static final String VM_MSG_GET_ALL_ONLINE_REQUESTED = "VM_MSG_GET_ALL_ONLINE_REQUESTED"; - // 移动位置订阅通知 + /** + * 移动位置订阅通知 + */ public static final String VM_MSG_SUBSCRIBE_MOBILE_POSITION = "mobileposition"; - // 报警订阅的通知（收到报警向redis发出通知） + /** + * 报警订阅的通知（收到报警向redis发出通知） + */ public static final String VM_MSG_SUBSCRIBE_ALARM = "alarm"; - // 报警通知的发送 （收到redis发出的通知，转发给其他平台） + /** + * 报警通知的发送 （收到redis发出的通知，转发给其他平台） + */ public static final String VM_MSG_SUBSCRIBE_ALARM_RECEIVE= "alarm_receive"; - // 设备状态订阅的通知 + /** + * 设备状态订阅的通知 + */ public static final String VM_MSG_SUBSCRIBE_DEVICE_STATUS = "device"; - - - //************************** 第三方 **************************************** + public static final String WVP_STREAM_GB_ID_PREFIX = "memberNo_"; public static final String WVP_STREAM_GPS_MSG_PREFIX = "WVP_STREAM_GPS_MSG_"; diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java index 1c2e6118..ad8043f1 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java @@ -62,7 +62,7 @@ public class SIPRequestHeaderPlarformProvider { // Forwards MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70); // ceq - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.MESSAGE), Request.MESSAGE); + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.MESSAGE); request = sipFactory.createMessageFactory().createRequest(requestURI, Request.MESSAGE, callIdHeader, cSeqHeader, fromHeader, toHeader, viaHeaders, maxForwards); @@ -120,7 +120,7 @@ public class SIPRequestHeaderPlarformProvider { String callId, WWWAuthenticateHeader www , CallIdHeader callIdHeader) throws ParseException, PeerUnavailableException, InvalidArgumentException { - Request registerRequest = createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(Request.REGISTER), fromTag, viaTag, callIdHeader); + Request registerRequest = createRegisterRequest(parentPlatform, redisCatchStorage.getCSEQ(), fromTag, viaTag, callIdHeader); SipURI requestURI = sipFactory.createAddressFactory().createSipURI(parentPlatform.getServerGBId(), parentPlatform.getServerIP() + ":" + parentPlatform.getServerPort()); if (www == null) { AuthorizationHeader authorizationHeader = sipFactory.createHeaderFactory().createAuthorizationHeader("Digest"); @@ -213,7 +213,7 @@ public class SIPRequestHeaderPlarformProvider { // Forwards MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70); // ceq - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.MESSAGE), Request.MESSAGE); + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.MESSAGE); MessageFactoryImpl messageFactory = (MessageFactoryImpl) sipFactory.createMessageFactory(); // 设置编码， 防止中文乱码 messageFactory.setDefaultContentEncodingCharset(parentPlatform.getCharacterSet()); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java index 349732b8..b89fd8ea 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java @@ -2,11 +2,9 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd; import java.text.ParseException; import java.util.ArrayList; +import java.util.List; -import javax.sip.Dialog; -import javax.sip.InvalidArgumentException; -import javax.sip.PeerUnavailableException; -import javax.sip.SipFactory; +import javax.sip.*; import javax.sip.address.Address; import javax.sip.address.SipURI; import javax.sip.header.*; @@ -15,7 +13,11 @@ import javax.sip.message.Request; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import gov.nist.javax.sip.SipProviderImpl; +import gov.nist.javax.sip.SipStackImpl; +import gov.nist.javax.sip.stack.SIPDialog; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import com.genersoft.iot.vmp.conf.SipConfig; @@ -40,6 +42,14 @@ public class SIPRequestHeaderProvider { @Autowired private VideoStreamSessionManager streamSession; + + @Autowired + @Qualifier(value="tcpSipProvider") + private SipProviderImpl tcpSipProvider; + + @Autowired + @Qualifier(value="udpSipProvider") + private SipProviderImpl udpSipProvider; public Request createMessageRequest(Device device, String content, String viaTag, String fromTag, String toTag, CallIdHeader callIdHeader) throws ParseException, InvalidArgumentException, PeerUnavailableException { Request request = null; @@ -95,7 +105,7 @@ public class SIPRequestHeaderProvider { MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70); //ceq - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.INVITE), Request.INVITE); + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.INVITE); request = sipFactory.createMessageFactory().createRequest(requestLine, Request.INVITE, callIdHeader, cSeqHeader,fromHeader, toHeader, viaHeaders, maxForwards); Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort())); @@ -131,7 +141,7 @@ public class SIPRequestHeaderProvider { MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70); //ceq - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.INVITE), Request.INVITE); + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.INVITE); request = sipFactory.createMessageFactory().createRequest(requestLine, Request.INVITE, callIdHeader, cSeqHeader,fromHeader, toHeader, viaHeaders, maxForwards); Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort())); @@ -200,7 +210,7 @@ public class SIPRequestHeaderProvider { MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70); // ceq - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(Request.SUBSCRIBE), Request.SUBSCRIBE); + CSeqHeader cSeqHeader = sipFactory.createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.SUBSCRIBE); request = sipFactory.createMessageFactory().createRequest(requestURI, Request.SUBSCRIBE, callIdHeader, cSeqHeader, fromHeader, toHeader, viaHeaders, maxForwards); @@ -226,55 +236,55 @@ public class SIPRequestHeaderProvider { } public Request createInfoRequest(Device device, StreamInfo streamInfo, String content) - throws PeerUnavailableException, ParseException, InvalidArgumentException { - Request request = null; + throws SipException, ParseException, InvalidArgumentException { if (streamInfo == null) { return null; } - Dialog dialog = streamSession.getDialogByStream(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream()); + Request request = null; + SIPDialog dialog = streamSession.getDialogByStream(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream()); if (dialog == null) { return null; } - SipURI requestLine = sipFactory.createAddressFactory().createSipURI(device.getDeviceId(), - device.getHostAddress()); - // via - ArrayList viaHeaders = new ArrayList(); - ViaHeader viaHeader = sipFactory.createHeaderFactory().createViaHeader(device.getIp(), device.getPort(), - device.getTransport(), null); + SipStack sipStack = udpSipProvider.getSipStack(); + SIPDialog sipDialog = ((SipStackImpl) sipStack).putDialog(dialog); + if (dialog != sipDialog) { + dialog = sipDialog; + }else { + dialog.setSipProvider(udpSipProvider); + } + streamSession.put(streamInfo.getDeviceID(), streamInfo.getChannelId(), dialog.getCallId().getCallId(), dialog); + Request infoRequest = dialog.createRequest(Request.INFO); + SipURI sipURI = (SipURI) infoRequest.getRequestURI(); + sipURI.setHost(device.getIp()); + sipURI.setPort(device.getPort()); + sipURI.setUser(streamInfo.getChannelId()); + + ViaHeader viaHeader = (ViaHeader) infoRequest.getHeader(ViaHeader.NAME); viaHeader.setRPort(); - viaHeaders.add(viaHeader); - // from - SipURI fromSipURI = sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), - sipConfig.getDomain()); - Address fromAddress = sipFactory.createAddressFactory().createAddress(fromSipURI); - FromHeader fromHeader = sipFactory.createHeaderFactory().createFromHeader(fromAddress, dialog.getLocalTag()); - // to - SipURI toSipURI = sipFactory.createAddressFactory().createSipURI(streamInfo.getChannelId(), - sipConfig.getDomain()); - Address toAddress = sipFactory.createAddressFactory().createAddress(toSipURI); - ToHeader toHeader = sipFactory.createHeaderFactory().createToHeader(toAddress, dialog.getRemoteTag()); - - // callid - CallIdHeader callIdHeader = dialog.getCallId(); - - // Forwards - MaxForwardsHeader maxForwards = sipFactory.createHeaderFactory().createMaxForwardsHeader(70); - - Long cseq = redisCatchStorage.getCSEQ(Request.INVITE); - // ceq - CSeqHeader cSeqHeader = sipFactory.createHeaderFactory() - .createCSeqHeader(cseq, Request.INFO); - - request = sipFactory.createMessageFactory().createRequest(requestLine, Request.INFO, callIdHeader, cSeqHeader, - fromHeader, toHeader, viaHeaders, maxForwards); + // 增加Contact header Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory() .createSipURI(sipConfig.getId(), sipConfig.getIp() + ":" + sipConfig.getPort())); - request.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress)); + infoRequest.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress)); + List agentParam = new ArrayList<>(); + agentParam.add("wvp-pro"); + // TODO 添加版本信息以及日期 + UserAgentHeader userAgentHeader = null; + try { + userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam); + } catch (ParseException e) { + throw new RuntimeException(e); + } + infoRequest.addHeader(userAgentHeader); ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSRTSP"); - request.setContent(content, contentTypeHeader); - return request; + infoRequest.setContent(content, contentTypeHeader); + + CSeqHeader cSeqHeader = (CSeqHeader)infoRequest.getHeader(CSeqHeader.NAME); + cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ()); + // ceq + infoRequest.addHeader(cSeqHeader); + return infoRequest; } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java index feb66b47..cb6038b4 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java @@ -732,8 +732,23 @@ public class SIPCommander implements ISIPCommander { SIPRequest request = (SIPRequest)transaction.getRequest(); byeURI.setHost(request.getRemoteAddress().getHostAddress()); byeURI.setPort(request.getRemotePort()); + byeURI.setUser(channelId); ViaHeader viaHeader = (ViaHeader) byeRequest.getHeader(ViaHeader.NAME); String protocol = viaHeader.getTransport().toUpperCase(); + viaHeader.setRPort(); + // 增加Contact header + Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort())); + byeRequest.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress)); + List agentParam = new ArrayList<>(); + agentParam.add("wvp-pro"); + // TODO 添加版本信息以及日期 + UserAgentHeader userAgentHeader = null; + try { + userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam); + } catch (ParseException e) { + throw new RuntimeException(e); + } + byeRequest.addHeader(userAgentHeader); ClientTransaction clientTransaction = null; if("TCP".equals(protocol)) { clientTransaction = tcpSipProvider.getNewClientTransaction(byeRequest); @@ -745,11 +760,14 @@ public class SIPCommander implements ISIPCommander { if (okEvent != null) { sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), okEvent); } - + CSeqHeader cSeqHeader = (CSeqHeader)byeRequest.getHeader(CSeqHeader.NAME); + cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ()); dialog.sendRequest(clientTransaction); } catch (SipException | ParseException e) { e.printStackTrace(); + } catch (InvalidArgumentException e) { + throw new RuntimeException(e); } } @@ -1483,7 +1501,7 @@ public class SIPCommander implements ISIPCommander { request.setContent(subscribePostitionXml.toString(), contentTypeHeader); CSeqHeader cSeqHeader = (CSeqHeader)request.getHeader(CSeqHeader.NAME); - cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ(Request.SUBSCRIBE)); + cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ()); request.removeHeader(CSeqHeader.NAME); request.addHeader(cSeqHeader); }else { @@ -1587,7 +1605,7 @@ public class SIPCommander implements ISIPCommander { request.setContent(cmdXml.toString(), contentTypeHeader); CSeqHeader cSeqHeader = (CSeqHeader)request.getHeader(CSeqHeader.NAME); - cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ(Request.SUBSCRIBE)); + cSeqHeader.setSeqNumber(redisCatchStorage.getCSEQ()); request.removeHeader(CSeqHeader.NAME); request.addHeader(cSeqHeader); @@ -1697,10 +1715,9 @@ public class SIPCommander implements ISIPCommander { @Override public void playPauseCmd(Device device, StreamInfo streamInfo) { try { - Long cseq = redisCatchStorage.getCSEQ(Request.INFO); StringBuffer content = new StringBuffer(200); content.append("PAUSE RTSP/1.0\r\n"); - content.append("CSeq: " + cseq + "\r\n"); + content.append("CSeq: " + getInfoCseq() + "\r\n"); content.append("PauseTime: now\r\n"); Request request = headerProvider.createInfoRequest(device, streamInfo, content.toString()); if (request == null) { @@ -1728,10 +1745,9 @@ public class SIPCommander implements ISIPCommander { @Override public void playResumeCmd(Device device, StreamInfo streamInfo) { try { - Long cseq = redisCatchStorage.getCSEQ(Request.INFO); StringBuffer content = new StringBuffer(200); content.append("PLAY RTSP/1.0\r\n"); - content.append("CSeq: " + cseq + "\r\n"); + content.append("CSeq: " + getInfoCseq() + "\r\n"); content.append("Range: npt=now-\r\n"); Request request = headerProvider.createInfoRequest(device, streamInfo, content.toString()); if (request == null) { @@ -1758,10 +1774,9 @@ public class SIPCommander implements ISIPCommander { @Override public void playSeekCmd(Device device, StreamInfo streamInfo, long seekTime) { try { - Long cseq = redisCatchStorage.getCSEQ(Request.INFO); StringBuffer content = new StringBuffer(200); content.append("PLAY RTSP/1.0\r\n"); - content.append("CSeq: " + cseq + "\r\n"); + content.append("CSeq: " + getInfoCseq() + "\r\n"); content.append("Range: npt=" + Math.abs(seekTime) + "-\r\n"); Request request = headerProvider.createInfoRequest(device, streamInfo, content.toString()); @@ -1789,11 +1804,11 @@ public class SIPCommander implements ISIPCommander { @Override public void playSpeedCmd(Device device, StreamInfo streamInfo, Double speed) { try { - Long cseq = redisCatchStorage.getCSEQ(Request.INFO); + StringBuffer content = new StringBuffer(200); content.append("PLAY RTSP/1.0\r\n"); - content.append("CSeq: " + cseq + "\r\n"); - content.append("Scale: " + String.format("%.1f",speed) + "\r\n"); + content.append("CSeq: " + getInfoCseq() + "\r\n"); + content.append("Scale: " + String.format("%.6f",speed) + "\r\n"); Request request = headerProvider.createInfoRequest(device, streamInfo, content.toString()); if (request == null) { return; @@ -1812,6 +1827,10 @@ public class SIPCommander implements ISIPCommander { e.printStackTrace(); } } + + private int getInfoCseq() { + return (int) ((Math.random() * 9 + 1) * Math.pow(10, 8)); + } @Override public void playbackControlCmd(Device device, StreamInfo streamInfo, String content,SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) { @@ -1820,7 +1839,6 @@ public class SIPCommander implements ISIPCommander { if (request == null) { return; } - logger.info(request.toString()); ClientTransaction clientTransaction = null; if ("TCP".equals(device.getTransport())) { clientTransaction = tcpSipProvider.getNewClientTransaction(request); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java index 10f428b0..36d9298e 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java @@ -105,7 +105,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform { } request = headerProviderPlarformProvider.createRegisterRequest(parentPlatform, - redisCatchStorage.getCSEQ(Request.REGISTER), "FromRegister" + tm, + redisCatchStorage.getCSEQ(), "FromRegister" + tm, "z9hG4bK-" + UUID.randomUUID().toString().replace("-", ""), callIdHeader); // 将 callid 写入缓存， 等注册成功可以更新状态 String callIdFromHeader = callIdHeader.getCallId(); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/InviteResponseProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/InviteResponseProcessor.java index 89958e92..04a11b93 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/InviteResponseProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/InviteResponseProcessor.java @@ -2,24 +2,32 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.response.impl; import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.gb28181.SipLayer; +import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; import com.genersoft.iot.vmp.gb28181.transmit.event.response.SIPResponseProcessorAbstract; import gov.nist.javax.sip.ResponseEventExt; +import gov.nist.javax.sip.message.SIPResponse; import gov.nist.javax.sip.stack.SIPDialog; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import javax.sip.InvalidArgumentException; -import javax.sip.ResponseEvent; -import javax.sip.SipException; +import javax.sdp.SdpFactory; +import javax.sdp.SdpParseException; +import javax.sdp.SessionDescription; +import javax.sip.*; +import javax.sip.address.Address; import javax.sip.address.SipURI; import javax.sip.header.CSeqHeader; +import javax.sip.header.CallIdHeader; +import javax.sip.header.UserAgentHeader; import javax.sip.message.Request; import javax.sip.message.Response; import java.text.ParseException; +import java.util.ArrayList; +import java.util.List; /** @@ -34,23 +42,24 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract { private final String method = "INVITE"; @Autowired - private SipLayer sipLayer; - - @Autowired - private SipConfig config; - + private VideoStreamSessionManager streamSession; @Autowired private SIPProcessorObserver sipProcessorObserver; + @Autowired + private SipConfig sipConfig; + + @Autowired + private SipFactory sipFactory; + @Override public void afterPropertiesSet() throws Exception { // 添加消息处理的订阅 sipProcessorObserver.addResponseProcessor(method, this); } - @Autowired - private VideoStreamSessionManager streamSession; + /** * 处理invite响应 @@ -74,6 +83,19 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract { CSeqHeader cseq = (CSeqHeader) response.getHeader(CSeqHeader.NAME); Request reqAck = dialog.createAck(cseq.getSeqNumber()); SipURI requestURI = (SipURI) reqAck.getRequestURI(); + String contentString = new String(response.getRawContent()); + // jainSip不支持y=字段， 移除以解析。 + int ssrcIndex = contentString.indexOf("y="); + // 检查是否有y字段 + SessionDescription sdp; + if (ssrcIndex >= 0) { + //ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段 + String substring = contentString.substring(0, contentString.indexOf("y=")); + sdp = SdpFactory.getInstance().createSessionDescription(substring); + } else { + sdp = SdpFactory.getInstance().createSessionDescription(contentString); + } + requestURI.setUser(sdp.getOrigin().getUsername()); try { requestURI.setHost(event.getRemoteIpAddress()); } catch (ParseException e) { @@ -81,6 +103,18 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract { } requestURI.setPort(event.getRemotePort()); reqAck.setRequestURI(requestURI); + List agentParam = new ArrayList<>(); + agentParam.add("wvp-pro"); + // TODO 添加版本信息以及日期 + UserAgentHeader userAgentHeader = null; + try { + userAgentHeader = sipFactory.createHeaderFactory().createUserAgentHeader(agentParam); + } catch (ParseException e) { + throw new RuntimeException(e); + } + reqAck.addHeader(userAgentHeader); + Address concatAddress = sipFactory.createAddressFactory().createAddress(sipFactory.createAddressFactory().createSipURI(sipConfig.getId(), sipConfig.getIp()+":"+sipConfig.getPort())); + reqAck.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress)); logger.info("[回复ack] {}-> {}:{} ",requestURI, event.getRemoteIpAddress(), event.getRemotePort()); dialog.sendAck(reqAck); @@ -88,6 +122,10 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract { } } catch (InvalidArgumentException | SipException e) { e.printStackTrace(); + } catch (ParseException e) { + throw new RuntimeException(e); + } catch (SdpParseException e) { + throw new RuntimeException(e); } } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java index 57ad192f..e934a26f 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -98,9 +98,7 @@ public class ZLMHttpHookListener { @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8") public ResponseEntity onServerKeepalive(@RequestBody JSONObject json){ - if (logger.isDebugEnabled()) { - logger.debug("[ ZLM HOOK ] on_server_keepalive API调用，参数：" + json.toString()); - } + logger.info("[ ZLM HOOK ] on_server_keepalive API调用，参数：" + json.toString()); String mediaServerId = json.getString("mediaServerId"); List subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive); if (subscribes != null && subscribes.size() > 0) { diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java index aded98d5..df66bbac 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java @@ -277,13 +277,7 @@ public class MediaServerServiceImpl implements IMediaServerService { return null; } String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId; - MediaServerItem serverItem=(MediaServerItem)redisUtil.get(key); - if(null==serverItem){ - //zlm服务不在线，启动重连 - reloadZlm(); - serverItem=(MediaServerItem)redisUtil.get(key); - } - return serverItem; + return (MediaServerItem)redisUtil.get(key); } @Override @@ -412,7 +406,6 @@ public class MediaServerServiceImpl implements IMediaServerService { } redisUtil.set(key, serverItem); resetOnlineServerItem(serverItem); - updateMediaServerKeepalive(serverItem.getId(), null); if (serverItem.isAutoConfig()) { setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable())); } @@ -476,9 +469,6 @@ public class MediaServerServiceImpl implements IMediaServerService { String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(); if (redisUtil.zSize(key) == null || redisUtil.zSize(key) == 0) { - logger.info("获取负载最低的节点时无在线节点，启动重连机制"); - //启动重连 - reloadZlm(); if (redisUtil.zSize(key) == null || redisUtil.zSize(key) == 0) { logger.info("获取负载最低的节点时无在线节点"); return null; @@ -643,6 +633,11 @@ public class MediaServerServiceImpl implements IMediaServerService { public void updateMediaServerKeepalive(String mediaServerId, JSONObject data) { MediaServerItem mediaServerItem = getOne(mediaServerId); if (mediaServerItem == null) { + // 缓存不存在，从数据库查询，如果数据库不存在则是错误的 + MediaServerItem mediaServerItemFromDatabase = getOneFromDatabase(mediaServerId); + if (mediaServerItemFromDatabase == null) { + return; + } // zlm连接重试 logger.warn("[更新ZLM 保活信息]失败，未找到流媒体信息,尝试重连zlm"); reloadZlm(); @@ -658,6 +653,10 @@ public class MediaServerServiceImpl implements IMediaServerService { redisUtil.set(key, data, hookAliveInterval); } + private MediaServerItem getOneFromDatabase(String mediaServerId) { + return mediaServerMapper.queryOne(mediaServerId); + } + @Override public void syncCatchFromDatabase() { List allInCatch = getAll(); diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index b9811da1..d10de489 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -17,10 +17,9 @@ public interface IRedisCatchStorage { /** * 计数器。为cseq进行计数 * - * @param method sip 方法 * @return */ - Long getCSEQ(String method); + Long getCSEQ(); /** * 开始播放时将流存入 diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java index 31294a08..81920b8f 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java @@ -42,8 +42,8 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { private UserSetting userSetting; @Override - public Long getCSEQ(String method) { - String key = VideoManagerConstants.SIP_CSEQ_PREFIX + userSetting.getServerId() + "_" + method; + public Long getCSEQ() { + String key = VideoManagerConstants.SIP_CSEQ_PREFIX + userSetting.getServerId(); long result = redis.incr(key, 1L); if (result > Integer.MAX_VALUE) {