Commit 39078225f1104eeaed21b24a024457b892fa7f7a

Authored by panlinlin
1 parent bc2b2885

优化级联时的异常处理

Showing 25 changed files with 226 additions and 31 deletions
src/main/java/com/genersoft/iot/vmp/conf/ApplicationCheckRunner.java
@@ -55,7 +55,7 @@ public class ApplicationCheckRunner implements CommandLineRunner { @@ -55,7 +55,7 @@ public class ApplicationCheckRunner implements CommandLineRunner {
55 System.exit(1); 55 System.exit(1);
56 } 56 }
57 57
58 - if (mediaIp.equals("localhost") || mediaIp.equals("127.0.0.1")) { 58 + if (mediaIp.equals("localhost") || (mediaIp.equals("127.0.0.1") && mediaWanIp == null)) {
59 logger.warn("mediaIp.ip使用 {} ,将无法收到网络内其他设备的推流!!!", mediaIp ); 59 logger.warn("mediaIp.ip使用 {} ,将无法收到网络内其他设备的推流!!!", mediaIp );
60 } 60 }
61 61
src/main/java/com/genersoft/iot/vmp/conf/SipDeviceRunner.java 0 → 100644
  1 +package com.genersoft.iot.vmp.conf;
  2 +
  3 +import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
  4 +import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
  5 +import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
  6 +import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
  7 +import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
  8 +import org.springframework.beans.factory.annotation.Autowired;
  9 +import org.springframework.boot.CommandLineRunner;
  10 +import org.springframework.core.annotation.Order;
  11 +import org.springframework.stereotype.Component;
  12 +
  13 +import java.util.List;
  14 +
  15 +/**
  16 + * 系统启动时控制设备离线
  17 + */
  18 +@Component
  19 +@Order(value=4)
  20 +public class SipDeviceRunner implements CommandLineRunner {
  21 +
  22 + @Autowired
  23 + private IVideoManagerStorager storager;
  24 +
  25 + @Autowired
  26 + private IRedisCatchStorage redisCatchStorage;
  27 +
  28 + @Override
  29 + public void run(String... args) throws Exception {
  30 + // 设置所有设备离线
  31 + storager.outlineForAll();
  32 + }
  33 +}
src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java
@@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.conf; @@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.conf;
3 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; 3 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
4 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch; 4 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
5 import com.genersoft.iot.vmp.gb28181.event.EventPublisher; 5 import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
  6 +import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
6 import com.genersoft.iot.vmp.storager.IRedisCatchStorage; 7 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
7 import com.genersoft.iot.vmp.storager.IVideoManagerStorager; 8 import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
8 import org.springframework.beans.factory.annotation.Autowired; 9 import org.springframework.beans.factory.annotation.Autowired;
@@ -28,6 +29,10 @@ public class SipPlatformRunner implements CommandLineRunner { @@ -28,6 +29,10 @@ public class SipPlatformRunner implements CommandLineRunner {
28 @Autowired 29 @Autowired
29 private EventPublisher publisher; 30 private EventPublisher publisher;
30 31
  32 + @Autowired
  33 + private ZLMRTPServerFactory zlmrtpServerFactory;
  34 +
  35 +
31 @Override 36 @Override
32 public void run(String... args) throws Exception { 37 public void run(String... args) throws Exception {
33 // 设置所有平台离线 38 // 设置所有平台离线
@@ -36,6 +41,9 @@ public class SipPlatformRunner implements CommandLineRunner { @@ -36,6 +41,9 @@ public class SipPlatformRunner implements CommandLineRunner {
36 // 清理所有平台注册缓存 41 // 清理所有平台注册缓存
37 redisCatchStorage.cleanPlatformRegisterInfos(); 42 redisCatchStorage.cleanPlatformRegisterInfos();
38 43
  44 + // 停止所有推流
  45 +// zlmrtpServerFactory.closeAllSendRtpStream();
  46 +
39 List<ParentPlatform> parentPlatforms = storager.queryEnableParentPlatformList(true); 47 List<ParentPlatform> parentPlatforms = storager.queryEnableParentPlatformList(true);
40 48
41 for (ParentPlatform parentPlatform : parentPlatforms) { 49 for (ParentPlatform parentPlatform : parentPlatforms) {
src/main/java/com/genersoft/iot/vmp/conf/security/WebSecurityConfig.java
@@ -111,7 +111,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter { @@ -111,7 +111,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
111 http.headers().contentTypeOptions().disable(); 111 http.headers().contentTypeOptions().disable();
112 http.authorizeRequests() 112 http.authorizeRequests()
113 // 放行接口 113 // 放行接口
114 - .antMatchers("/api/user/login","/index/hook/**").permitAll() 114 + .antMatchers("/#/**", "/api/user/login","/index/hook/**").permitAll()
115 // 除上面外的所有请求全部需要鉴权认证 115 // 除上面外的所有请求全部需要鉴权认证
116 .anyRequest().authenticated() 116 .anyRequest().authenticated()
117 // 异常处理(权限拒绝、登录失效等) 117 // 异常处理(权限拒绝、登录失效等)
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepaliveTimeoutListenerForPlatform.java
@@ -39,8 +39,8 @@ public class KeepaliveTimeoutListenerForPlatform extends KeyExpirationEventMessa @@ -39,8 +39,8 @@ public class KeepaliveTimeoutListenerForPlatform extends KeyExpirationEventMessa
39 // 获取失效的key 39 // 获取失效的key
40 String expiredKey = message.toString(); 40 String expiredKey = message.toString();
41 logger.info(expiredKey); 41 logger.info(expiredKey);
42 - if(!expiredKey.startsWith(VideoManagerConstants.PLATFORM_PREFIX)){  
43 - logger.info("收到redis过期监听,但开头不是"+VideoManagerConstants.PLATFORM_PREFIX+",忽略"); 42 + if(!expiredKey.startsWith(VideoManagerConstants.PLATFORM_KEEPLIVEKEY_PREFIX)){
  43 + logger.debug("收到redis过期监听,但开头不是"+VideoManagerConstants.PLATFORM_KEEPLIVEKEY_PREFIX+",忽略");
44 return; 44 return;
45 } 45 }
46 // 平台心跳到期,需要重发, 判断是否已经多次未收到心跳回复, 多次未收到,则重新发起注册, 注册尝试多次未得到回复,则认为平台离线 46 // 平台心跳到期,需要重发, 判断是否已经多次未收到心跳回复, 多次未收到,则重新发起注册, 注册尝试多次未得到回复,则认为平台离线
@@ -49,7 +49,6 @@ public class KeepaliveTimeoutListenerForPlatform extends KeyExpirationEventMessa @@ -49,7 +49,6 @@ public class KeepaliveTimeoutListenerForPlatform extends KeyExpirationEventMessa
49 49
50 publisher.platformKeepaliveExpireEventPublish(platformGBId); 50 publisher.platformKeepaliveExpireEventPublish(platformGBId);
51 }else if (expiredKey.startsWith(VideoManagerConstants.PLATFORM_REGISTER_PREFIX)) { 51 }else if (expiredKey.startsWith(VideoManagerConstants.PLATFORM_REGISTER_PREFIX)) {
52 - logger.info("11111111111111");  
53 String platformGBId = expiredKey.substring(VideoManagerConstants.PLATFORM_REGISTER_PREFIX.length(),expiredKey.length()); 52 String platformGBId = expiredKey.substring(VideoManagerConstants.PLATFORM_REGISTER_PREFIX.length(),expiredKey.length());
54 53
55 publisher.platformNotRegisterEventPublish(platformGBId); 54 publisher.platformNotRegisterEventPublish(platformGBId);
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepliveTimeoutListener.java
@@ -38,7 +38,7 @@ public class KeepliveTimeoutListener extends KeyExpirationEventMessageListener { @@ -38,7 +38,7 @@ public class KeepliveTimeoutListener extends KeyExpirationEventMessageListener {
38 // 获取失效的key 38 // 获取失效的key
39 String expiredKey = message.toString(); 39 String expiredKey = message.toString();
40 if(!expiredKey.startsWith(VideoManagerConstants.KEEPLIVEKEY_PREFIX)){ 40 if(!expiredKey.startsWith(VideoManagerConstants.KEEPLIVEKEY_PREFIX)){
41 - logger.info("收到redis过期监听,但开头不是"+VideoManagerConstants.KEEPLIVEKEY_PREFIX+",忽略"); 41 + logger.debug("收到redis过期监听,但开头不是"+VideoManagerConstants.KEEPLIVEKEY_PREFIX+",忽略");
42 return; 42 return;
43 } 43 }
44 44
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformKeepaliveExpire/PlatformKeepaliveExpireEventLister.java
@@ -63,6 +63,7 @@ public class PlatformKeepaliveExpireEventLister implements ApplicationListener&lt;P @@ -63,6 +63,7 @@ public class PlatformKeepaliveExpireEventLister implements ApplicationListener&lt;P
63 if (parentPlatformCatch.getKeepAliveReply() >= 3) { 63 if (parentPlatformCatch.getKeepAliveReply() >= 3) {
64 // 有3次未收到心跳回复, 设置平台状态为离线, 开始重新注册 64 // 有3次未收到心跳回复, 设置平台状态为离线, 开始重新注册
65 logger.warn("有3次未收到心跳回复,标记设置平台状态为离线, 并重新注册 平台国标ID:" + event.getPlatformGbID()); 65 logger.warn("有3次未收到心跳回复,标记设置平台状态为离线, 并重新注册 平台国标ID:" + event.getPlatformGbID());
  66 + storager.updateParentPlatformStatus(event.getPlatformGbID(), false);
66 publisher.platformNotRegisterEventPublish(event.getPlatformGbID()); 67 publisher.platformNotRegisterEventPublish(event.getPlatformGbID());
67 parentPlatformCatch.setKeepAliveReply(0); 68 parentPlatformCatch.setKeepAliveReply(0);
68 }else { 69 }else {
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformNotRegister/PlatformNotRegisterEventLister.java
1 package com.genersoft.iot.vmp.gb28181.event.platformNotRegister; 1 package com.genersoft.iot.vmp.gb28181.event.platformNotRegister;
2 2
3 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; 3 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
  4 +import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
4 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform; 5 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
  6 +import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
  7 +import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
5 import com.genersoft.iot.vmp.storager.IVideoManagerStorager; 8 import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
6 import org.slf4j.Logger; 9 import org.slf4j.Logger;
7 import org.slf4j.LoggerFactory; 10 import org.slf4j.LoggerFactory;
@@ -9,6 +12,10 @@ import org.springframework.beans.factory.annotation.Autowired; @@ -9,6 +12,10 @@ import org.springframework.beans.factory.annotation.Autowired;
9 import org.springframework.context.ApplicationListener; 12 import org.springframework.context.ApplicationListener;
10 import org.springframework.stereotype.Component; 13 import org.springframework.stereotype.Component;
11 14
  15 +import java.util.HashMap;
  16 +import java.util.List;
  17 +import java.util.Map;
  18 +
12 /** 19 /**
13 * @Description: 平台未注册事件,来源有二: 20 * @Description: 平台未注册事件,来源有二:
14 * 1、平台新添加 21 * 1、平台新添加
@@ -23,23 +30,53 @@ public class PlatformNotRegisterEventLister implements ApplicationListener&lt;Platf @@ -23,23 +30,53 @@ public class PlatformNotRegisterEventLister implements ApplicationListener&lt;Platf
23 30
24 @Autowired 31 @Autowired
25 private IVideoManagerStorager storager; 32 private IVideoManagerStorager storager;
  33 + @Autowired
  34 + private IRedisCatchStorage redisCatchStorage;
26 35
27 @Autowired 36 @Autowired
28 private SIPCommanderFroPlatform sipCommanderFroPlatform; 37 private SIPCommanderFroPlatform sipCommanderFroPlatform;
29 38
  39 + @Autowired
  40 + private ZLMRTPServerFactory zlmrtpServerFactory;
  41 +
30 // @Autowired 42 // @Autowired
31 // private RedisUtil redis; 43 // private RedisUtil redis;
32 44
33 @Override 45 @Override
34 public void onApplicationEvent(PlatformNotRegisterEvent event) { 46 public void onApplicationEvent(PlatformNotRegisterEvent event) {
35 47
36 - logger.debug("平台未注册事件触发,平台国标ID:" + event.getPlatformGbID()); 48 + logger.info("平台未注册事件触发,平台国标ID:" + event.getPlatformGbID());
37 49
38 ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformGbID()); 50 ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformGbID());
39 if (parentPlatform == null) { 51 if (parentPlatform == null) {
40 - logger.debug("平台未注册事件触发,但平台已经删除!!! 平台国标ID:" + event.getPlatformGbID()); 52 + logger.info("平台未注册事件触发,但平台已经删除!!! 平台国标ID:" + event.getPlatformGbID());
41 return; 53 return;
42 } 54 }
  55 + // 查询是否有推流, 如果有则都停止
  56 + List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(event.getPlatformGbID());
  57 + logger.info("停止[ {} ]的所有推流size", sendRtpItems.size());
  58 + if (sendRtpItems != null && sendRtpItems.size() > 0) {
  59 + logger.info("停止[ {} ]的所有推流", event.getPlatformGbID());
  60 + StringBuilder app = new StringBuilder();
  61 + StringBuilder stream = new StringBuilder();
  62 + for (int i = 0; i < sendRtpItems.size(); i++) {
  63 + if (app.length() != 0) {
  64 + app.append(",");
  65 + }
  66 + app.append(sendRtpItems.get(i).getApp());
  67 + if (stream.length() != 0) {
  68 + stream.append(",");
  69 + }
  70 + stream.append(sendRtpItems.get(i).getStreamId());
  71 + redisCatchStorage.deleteSendRTPServer(event.getPlatformGbID(), sendRtpItems.get(i).getChannelId());
  72 + }
  73 + Map<String, Object> param = new HashMap<>();
  74 + param.put("vhost","__defaultVhost__");
  75 + param.put("app", app.toString());
  76 + param.put("stream", stream.toString());
  77 + zlmrtpServerFactory.stopSendRtpStream(param);
  78 +
  79 + }
43 sipCommanderFroPlatform.register(parentPlatform); 80 sipCommanderFroPlatform.register(parentPlatform);
44 } 81 }
45 } 82 }
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/CheckForAllRecordsThread.java
@@ -43,7 +43,7 @@ public class CheckForAllRecordsThread extends Thread { @@ -43,7 +43,7 @@ public class CheckForAllRecordsThread extends Thread {
43 if (totalRecordList.size() < this.recordInfo.getSumNum()) { 43 if (totalRecordList.size() < this.recordInfo.getSumNum()) {
44 logger.info("已获取" + totalRecordList.size() + "项录像数据,共" + this.recordInfo.getSumNum() + "项"); 44 logger.info("已获取" + totalRecordList.size() + "项录像数据,共" + this.recordInfo.getSumNum() + "项");
45 } else { 45 } else {
46 - logger.info("录像数据已全部获取,共" + this.recordInfo.getSumNum() + "项"); 46 + logger.info("录像数据已全部获取,共 {} 项", this.recordInfo.getSumNum());
47 this.recordInfo.setRecordList(totalRecordList); 47 this.recordInfo.setRecordList(totalRecordList);
48 for (int i = 0; i < cacheKeys.size(); i++) { 48 for (int i = 0; i < cacheKeys.size(); i++) {
49 redis.del(cacheKeys.get(i).toString()); 49 redis.del(cacheKeys.get(i).toString());
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java
@@ -58,6 +58,7 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor { @@ -58,6 +58,7 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor {
58 param.put("vhost","__defaultVhost__"); 58 param.put("vhost","__defaultVhost__");
59 param.put("app",sendRtpItem.getApp()); 59 param.put("app",sendRtpItem.getApp());
60 param.put("stream",streamId); 60 param.put("stream",streamId);
  61 + param.put("ssrc",sendRtpItem.getSsrc());
61 logger.info("停止向上级推流:" + streamId); 62 logger.info("停止向上级推流:" + streamId);
62 zlmrtpServerFactory.stopSendRtpStream(param); 63 zlmrtpServerFactory.stopSendRtpStream(param);
63 redisCatchStorage.deleteSendRTPServer(platformGbId, channelId); 64 redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java
@@ -93,6 +93,11 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor { @@ -93,6 +93,11 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
93 GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId); 93 GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);
94 // 不是通道可能是直播流 94 // 不是通道可能是直播流
95 if (channel != null || gbStream != null ) { 95 if (channel != null || gbStream != null ) {
  96 + if (channel.getStatus() == 0) {
  97 + logger.info("通道离线,返回400");
  98 + responseAck(evt, Response.BAD_REQUEST, "channel [" + channel.getChannelId() + "] offline");
  99 + return;
  100 + }
96 responseAck(evt, Response.CALL_IS_BEING_FORWARDED); // 通道存在,发181,呼叫转接中 101 responseAck(evt, Response.CALL_IS_BEING_FORWARDED); // 通道存在,发181,呼叫转接中
97 }else { 102 }else {
98 logger.info("通道不存在,返回404"); 103 logger.info("通道不存在,返回404");
@@ -367,6 +372,12 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor { @@ -367,6 +372,12 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
367 getServerTransaction(evt).sendResponse(response); 372 getServerTransaction(evt).sendResponse(response);
368 } 373 }
369 374
  375 + private void responseAck(RequestEvent evt, int statusCode, String msg) throws SipException, InvalidArgumentException, ParseException {
  376 + Response response = getMessageFactory().createResponse(statusCode, evt.getRequest());
  377 + response.setReasonPhrase(msg);
  378 + getServerTransaction(evt).sendResponse(response);
  379 + }
  380 +
370 /** 381 /**
371 * 回复带sdp的200 382 * 回复带sdp的200
372 * @param evt 383 * @param evt
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
@@ -770,14 +770,17 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor { @@ -770,14 +770,17 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor {
770 try { 770 try {
771 Element rootElement = getRootElement(evt); 771 Element rootElement = getRootElement(evt);
772 String deviceId = XmlUtil.getText(rootElement, "DeviceID"); 772 String deviceId = XmlUtil.getText(rootElement, "DeviceID");
773 - // 检查设备是否存在, 不存在则不回复  
774 - if (storager.exists(deviceId)) { 773 + Device device = storager.queryVideoDevice(deviceId);
  774 + // 检查设备是否存在并在线, 不存在则不回复
  775 + if (device != null && device.getOnline() == 1) {
775 // 回复200 OK 776 // 回复200 OK
776 responseAck(evt); 777 responseAck(evt);
777 if (offLineDetector.isOnline(deviceId)) { 778 if (offLineDetector.isOnline(deviceId)) {
778 publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE); 779 publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
779 } else { 780 } else {
780 } 781 }
  782 + }else {
  783 + logger.warn("收到[ "+deviceId+" ]心跳信息, 但是设备" + (device == null? "不存在":"离线") + ", 心跳信息不予以回复");
781 } 784 }
782 } catch (ParseException | SipException | InvalidArgumentException | DocumentException e) { 785 } catch (ParseException | SipException | InvalidArgumentException | DocumentException e) {
783 e.printStackTrace(); 786 e.printStackTrace();
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/RegisterRequestProcessor.java
@@ -146,7 +146,7 @@ public class RegisterRequestProcessor extends SIPRequestAbstractProcessor { @@ -146,7 +146,7 @@ public class RegisterRequestProcessor extends SIPRequestAbstractProcessor {
146 // 注册成功 146 // 注册成功
147 // 保存到redis 147 // 保存到redis
148 // 下发catelog查询目录 148 // 下发catelog查询目录
149 - if (registerFlag == 1 && device != null) { 149 + if (registerFlag == 1 ) {
150 logger.info("注册成功! deviceId:" + device.getDeviceId()); 150 logger.info("注册成功! deviceId:" + device.getDeviceId());
151 // boolean exists = storager.exists(device.getDeviceId()); 151 // boolean exists = storager.exists(device.getDeviceId());
152 device.setRegisterTimeMillis(System.currentTimeMillis()); 152 device.setRegisterTimeMillis(System.currentTimeMillis());
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/RegisterResponseProcessor.java
@@ -80,12 +80,13 @@ public class RegisterResponseProcessor implements ISIPResponseProcessor { @@ -80,12 +80,13 @@ public class RegisterResponseProcessor implements ISIPResponseProcessor {
80 // 注册/注销成功 80 // 注册/注销成功
81 logger.info(String.format("%s %s成功", platformGBId, action)); 81 logger.info(String.format("%s %s成功", platformGBId, action));
82 redisCatchStorage.delPlatformRegisterInfo(callId); 82 redisCatchStorage.delPlatformRegisterInfo(callId);
83 - parentPlatform.setStatus(true); 83 + parentPlatform.setStatus("注册".equals(action));
84 // 取回Expires设置,避免注销过程中被置为0 84 // 取回Expires设置,避免注销过程中被置为0
85 ParentPlatform parentPlatformTmp = storager.queryParentPlatByServerGBId(platformGBId); 85 ParentPlatform parentPlatformTmp = storager.queryParentPlatByServerGBId(platformGBId);
86 String expires = parentPlatformTmp.getExpires(); 86 String expires = parentPlatformTmp.getExpires();
87 parentPlatform.setExpires(expires); 87 parentPlatform.setExpires(expires);
88 - storager.updateParentPlatform(parentPlatform); 88 + parentPlatform.setId(parentPlatformTmp.getId());
  89 + storager.updateParentPlatformStatus(platformGBId, "注册".equals(action));
89 90
90 redisCatchStorage.updatePlatformRegister(parentPlatform); 91 redisCatchStorage.updatePlatformRegister(parentPlatform);
91 92
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java
1 package com.genersoft.iot.vmp.media.zlm; 1 package com.genersoft.iot.vmp.media.zlm;
2 2
  3 +import com.alibaba.fastjson.JSON;
  4 +import com.alibaba.fastjson.JSONArray;
3 import com.alibaba.fastjson.JSONObject; 5 import com.alibaba.fastjson.JSONObject;
  6 +import com.genersoft.iot.vmp.conf.MediaServerConfig;
4 import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem; 7 import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
5 import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; 8 import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
6 import com.genersoft.iot.vmp.gb28181.bean.GbStream; 9 import com.genersoft.iot.vmp.gb28181.bean.GbStream;
@@ -119,4 +122,33 @@ public class ZLMMediaListManager { @@ -119,4 +122,33 @@ public class ZLMMediaListManager {
119 storager.mediaOutline(app, streamId); 122 storager.mediaOutline(app, streamId);
120 } 123 }
121 } 124 }
  125 +
  126 + public void clearAllSessions() {
  127 + logger.info("清空所有国标相关的session");
  128 + JSONObject allSessionJSON = zlmresTfulUtils.getAllSession();
  129 + MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
  130 + HashSet<String> allLocalPorts = new HashSet();
  131 + if (allSessionJSON.getInteger("code") == 0) {
  132 + JSONArray data = allSessionJSON.getJSONArray("data");
  133 + if (data.size() > 0) {
  134 + for (int i = 0; i < data.size(); i++) {
  135 + JSONObject sessionJOSN = data.getJSONObject(i);
  136 + Integer local_port = sessionJOSN.getInteger("local_port");
  137 + if (!local_port.equals(Integer.valueOf(mediaInfo.getHttpPort())) &&
  138 + !local_port.equals(Integer.valueOf(mediaInfo.getHttpSSLport())) &&
  139 + !local_port.equals(Integer.valueOf(mediaInfo.getRtmpPort())) &&
  140 + !local_port.equals(Integer.valueOf(mediaInfo.getRtspPort())) &&
  141 + !local_port.equals(Integer.valueOf(mediaInfo.getRtspSSlport())) &&
  142 + !local_port.equals(Integer.valueOf(mediaInfo.getHookOnFlowReport()))){
  143 + allLocalPorts.add(sessionJOSN.getInteger("local_port") + "");
  144 + }
  145 + }
  146 + }
  147 + }
  148 + if (allLocalPorts.size() > 0) {
  149 + List<String> result = new ArrayList<>(allLocalPorts);
  150 + String localPortSStr = String.join(",", result);
  151 + zlmresTfulUtils.kickSessions(localPortSStr);
  152 + }
  153 + }
122 } 154 }
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
@@ -194,4 +194,14 @@ public class ZLMRESTfulUtils { @@ -194,4 +194,14 @@ public class ZLMRESTfulUtils {
194 param.put("force", 1); 194 param.put("force", 1);
195 return sendPost("close_streams",param, null); 195 return sendPost("close_streams",param, null);
196 } 196 }
  197 +
  198 + public JSONObject getAllSession() {
  199 + return sendPost("getAllSession",null, null);
  200 + }
  201 +
  202 + public void kickSessions(String localPortSStr) {
  203 + Map<String, Object> param = new HashMap<>();
  204 + param.put("local_port", localPortSStr);
  205 + sendPost("kick_sessions",param, null);
  206 + }
197 } 207 }
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
@@ -18,13 +18,13 @@ public class ZLMRTPServerFactory { @@ -18,13 +18,13 @@ public class ZLMRTPServerFactory {
18 18
19 private Logger logger = LoggerFactory.getLogger("ZLMRTPServerFactory"); 19 private Logger logger = LoggerFactory.getLogger("ZLMRTPServerFactory");
20 20
21 - @Value("${media.rtp.udpPortRange}")  
22 - private String udpPortRange; 21 + @Value("${media.rtp.portRange}")
  22 + private String portRange;
23 23
24 @Autowired 24 @Autowired
25 private ZLMRESTfulUtils zlmresTfulUtils; 25 private ZLMRESTfulUtils zlmresTfulUtils;
26 26
27 - private int[] udpPortRangeArray = new int[2]; 27 + private int[] portRangeArray = new int[2];
28 28
29 private int currentPort = 0; 29 private int currentPort = 0;
30 30
@@ -52,7 +52,7 @@ public class ZLMRTPServerFactory { @@ -52,7 +52,7 @@ public class ZLMRTPServerFactory {
52 52
53 Map<String, Object> param = new HashMap<>(); 53 Map<String, Object> param = new HashMap<>();
54 int result = -1; 54 int result = -1;
55 - int newPort = getPortFromUdpPortRange(); 55 + int newPort = getPortFromportRange();
56 param.put("port", newPort); 56 param.put("port", newPort);
57 param.put("enable_tcp", 1); 57 param.put("enable_tcp", 1);
58 param.put("stream_id", streamId); 58 param.put("stream_id", streamId);
@@ -101,16 +101,16 @@ public class ZLMRTPServerFactory { @@ -101,16 +101,16 @@ public class ZLMRTPServerFactory {
101 return result; 101 return result;
102 } 102 }
103 103
104 - private int getPortFromUdpPortRange() { 104 + private int getPortFromportRange() {
105 if (currentPort == 0) { 105 if (currentPort == 0) {
106 - String[] udpPortRangeStrArray = udpPortRange.split(",");  
107 - udpPortRangeArray[0] = Integer.parseInt(udpPortRangeStrArray[0]);  
108 - udpPortRangeArray[1] = Integer.parseInt(udpPortRangeStrArray[1]); 106 + String[] portRangeStrArray = portRange.split(",");
  107 + portRangeArray[0] = Integer.parseInt(portRangeStrArray[0]);
  108 + portRangeArray[1] = Integer.parseInt(portRangeStrArray[1]);
109 } 109 }
110 110
111 - if (currentPort == 0 || currentPort++ > udpPortRangeArray[1]) {  
112 - currentPort = udpPortRangeArray[0];  
113 - return udpPortRangeArray[0]; 111 + if (currentPort == 0 || currentPort++ > portRangeArray[1]) {
  112 + currentPort = portRangeArray[0];
  113 + return portRangeArray[0];
114 } else { 114 } else {
115 if (currentPort % 2 == 1) { 115 if (currentPort % 2 == 1) {
116 currentPort++; 116 currentPort++;
@@ -244,4 +244,8 @@ public class ZLMRTPServerFactory { @@ -244,4 +244,8 @@ public class ZLMRTPServerFactory {
244 } 244 }
245 return result; 245 return result;
246 } 246 }
  247 +
  248 + public void closeAllSendRtpStream() {
  249 +
  250 + }
247 } 251 }
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java
@@ -164,6 +164,10 @@ public class ZLMRunner implements CommandLineRunner { @@ -164,6 +164,10 @@ public class ZLMRunner implements CommandLineRunner {
164 mediaServerConfig.setLocalIP(mediaIp); 164 mediaServerConfig.setLocalIP(mediaIp);
165 mediaServerConfig.setWanIp(StringUtils.isEmpty(mediaWanIp)? mediaIp: mediaWanIp); 165 mediaServerConfig.setWanIp(StringUtils.isEmpty(mediaWanIp)? mediaIp: mediaWanIp);
166 redisCatchStorage.updateMediaInfo(mediaServerConfig); 166 redisCatchStorage.updateMediaInfo(mediaServerConfig);
  167 +
  168 + // 清空所有session
  169 +// zlmMediaListManager.clearAllSessions();
  170 +
167 // 更新流列表 171 // 更新流列表
168 zlmMediaListManager.updateMediaList(); 172 zlmMediaListManager.updateMediaList();
169 // 恢复流代理 173 // 恢复流代理
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
@@ -6,6 +6,7 @@ import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; @@ -6,6 +6,7 @@ import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
6 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch; 6 import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
7 import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; 7 import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
8 8
  9 +import java.util.List;
9 import java.util.Map; 10 import java.util.Map;
10 11
11 public interface IRedisCatchStorage { 12 public interface IRedisCatchStorage {
@@ -91,6 +92,8 @@ public interface IRedisCatchStorage { @@ -91,6 +92,8 @@ public interface IRedisCatchStorage {
91 */ 92 */
92 SendRtpItem querySendRTPServer(String platformGbId, String channelId); 93 SendRtpItem querySendRTPServer(String platformGbId, String channelId);
93 94
  95 + List<SendRtpItem> querySendRTPServer(String platformGbId);
  96 +
94 /** 97 /**
95 * 删除RTP推送信息缓存 98 * 删除RTP推送信息缓存
96 * @param platformGbId 99 * @param platformGbId
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java
@@ -135,6 +135,13 @@ public interface IVideoManagerStorager { @@ -135,6 +135,13 @@ public interface IVideoManagerStorager {
135 */ 135 */
136 public boolean outline(String deviceId); 136 public boolean outline(String deviceId);
137 137
  138 + /**
  139 + * 更新所有设备离线
  140 + *
  141 + * @return true:更新成功 false:更新失败
  142 + */
  143 + public boolean outlineForAll();
  144 +
138 145
139 /** 146 /**
140 * 查询子设备 147 * 查询子设备
@@ -352,4 +359,10 @@ public interface IVideoManagerStorager { @@ -352,4 +359,10 @@ public interface IVideoManagerStorager {
352 * @param streamId 359 * @param streamId
353 */ 360 */
354 void mediaOutline(String app, String streamId); 361 void mediaOutline(String app, String streamId);
  362 +
  363 + /**
  364 + * 设置平台在线/离线
  365 + * @param online
  366 + */
  367 + void updateParentPlatformStatus(String platformGbID, boolean online);
355 } 368 }
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java
@@ -65,4 +65,7 @@ public interface DeviceMapper { @@ -65,4 +65,7 @@ public interface DeviceMapper {
65 65
66 @Delete("DELETE FROM device WHERE deviceId=#{deviceId}") 66 @Delete("DELETE FROM device WHERE deviceId=#{deviceId}")
67 int del(String deviceId); 67 int del(String deviceId);
  68 +
  69 + @Update("UPDATE device SET online=0")
  70 + int outlineForAll();
68 } 71 }
src/main/java/com/genersoft/iot/vmp/storager/dao/ParentPlatformMapper.java
@@ -66,5 +66,8 @@ public interface ParentPlatformMapper { @@ -66,5 +66,8 @@ public interface ParentPlatformMapper {
66 ParentPlatform getParentPlatById(int id); 66 ParentPlatform getParentPlatById(int id);
67 67
68 @Update("UPDATE parent_platform SET status=false" ) 68 @Update("UPDATE parent_platform SET status=false" )
69 - void outlineForAllParentPlatform(); 69 + int outlineForAllParentPlatform();
  70 +
  71 + @Update("UPDATE parent_platform SET status=#{online} WHERE serverGBId=#{platformGbID}" )
  72 + int updateParentPlatformStatus(String platformGbID, boolean online);
70 } 73 }
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
@@ -233,6 +233,20 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { @@ -233,6 +233,20 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
233 return (SendRtpItem)redis.get(key); 233 return (SendRtpItem)redis.get(key);
234 } 234 }
235 235
  236 + @Override
  237 + public List<SendRtpItem> querySendRTPServer(String platformGbId) {
  238 + String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + platformGbId + "_*";
  239 + List<Object> queryResult = redis.scan(key);
  240 + List<SendRtpItem> result= new ArrayList<>();
  241 +
  242 + for (int i = 0; i < queryResult.size(); i++) {
  243 + String keyItem = (String) queryResult.get(i);
  244 + result.add((SendRtpItem)redis.get(keyItem));
  245 + }
  246 +
  247 + return result;
  248 + }
  249 +
236 /** 250 /**
237 * 删除RTP推送信息缓存 251 * 删除RTP推送信息缓存
238 * @param platformGbId 252 * @param platformGbId
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java
@@ -257,6 +257,18 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager { @@ -257,6 +257,18 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
257 } 257 }
258 258
259 /** 259 /**
  260 + * 更新所有设备离线
  261 + *
  262 + * @return true:更新成功 false:更新失败
  263 + */
  264 + @Override
  265 + public synchronized boolean outlineForAll() {
  266 + logger.info("更新所有设备离线");
  267 + int result = deviceMapper.outlineForAll();
  268 + return result > 0;
  269 + }
  270 +
  271 + /**
260 * 清空通道 272 * 清空通道
261 * @param deviceId 273 * @param deviceId
262 */ 274 */
@@ -575,5 +587,8 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager { @@ -575,5 +587,8 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
575 gbStreamMapper.setStatus(app, streamId, false); 587 gbStreamMapper.setStatus(app, streamId, false);
576 } 588 }
577 589
578 - 590 + @Override
  591 + public void updateParentPlatformStatus(String platformGbID, boolean online) {
  592 + platformMapper.updateParentPlatformStatus(platformGbID, online);
  593 + }
579 } 594 }
src/main/resources/application-dev.yml
@@ -78,12 +78,12 @@ media: @@ -78,12 +78,12 @@ media:
78 autoApplyPlay: false 78 autoApplyPlay: false
79 # [可选] 部分设备需要扩展SDP,需要打开此设置 79 # [可选] 部分设备需要扩展SDP,需要打开此设置
80 seniorSdp: false 80 seniorSdp: false
81 - # 启用udp多端口模式 81 + # 启用多端口模式, 多端口模式使用端口区分每路流,兼容性更好。 单端口使用流的ssrc区分, 点播超时建议使用多端口测试
82 rtp: 82 rtp:
83 - # [可选] 是否启用udp多端口模式, 开启后会在udpPortRange范围内选择端口用于媒体流传输 83 + # [可选] 是否启用多端口模式, 开启后会在portRange范围内选择端口用于媒体流传输
84 enable: true 84 enable: true
85 - # [可选] 在此范围内选择端口用于媒体流传输, 不只是udp, 使用TCP被动传输模式时,也是从这个范围内选择端口  
86 - udpPortRange: 30000,30500 # 端口范围 85 + # [可选] 在此范围内选择端口用于媒体流传输,
  86 + portRange: 30000,30500 # 端口范围
87 87
88 # [可选] 日志配置, 一般不需要改 88 # [可选] 日志配置, 一般不需要改
89 logging: 89 logging: