|  |  |  | 
|---|
|  |  |  | import com.genersoft.iot.vmp.service.IPlatformService; | 
|---|
|  |  |  | import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; | 
|---|
|  |  |  | import com.genersoft.iot.vmp.storager.IRedisCatchStorage; | 
|---|
|  |  |  | import com.genersoft.iot.vmp.storager.dao.GbStreamMapper; | 
|---|
|  |  |  | import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper; | 
|---|
|  |  |  | import com.genersoft.iot.vmp.storager.dao.*; | 
|---|
|  |  |  | import com.genersoft.iot.vmp.utils.DateUtil; | 
|---|
|  |  |  | import com.github.pagehelper.PageHelper; | 
|---|
|  |  |  | import com.github.pagehelper.PageInfo; | 
|---|
|  |  |  | import org.slf4j.Logger; | 
|---|
|  |  |  | 
|---|
|  |  |  |  | 
|---|
|  |  |  | import javax.sip.InvalidArgumentException; | 
|---|
|  |  |  | import javax.sip.SipException; | 
|---|
|  |  |  | import javax.sip.TimeoutEvent; | 
|---|
|  |  |  | import java.text.ParseException; | 
|---|
|  |  |  | import java.util.HashMap; | 
|---|
|  |  |  | import java.util.List; | 
|---|
|  |  |  | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Autowired | 
|---|
|  |  |  | private ParentPlatformMapper platformMapper; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Autowired | 
|---|
|  |  |  | private PlatformCatalogMapper catalogMapper; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Autowired | 
|---|
|  |  |  | private PlatformChannelMapper platformChannelMapper; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Autowired | 
|---|
|  |  |  | private PlatformGbStreamMapper platformGbStreamMapper; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Autowired | 
|---|
|  |  |  | private IRedisCatchStorage redisCatchStorage; | 
|---|
|  |  |  | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Override | 
|---|
|  |  |  | public void online(ParentPlatform parentPlatform) { | 
|---|
|  |  |  | logger.info("[国标级联]:{}, 平台上线/更新注册", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | public boolean update(ParentPlatform parentPlatform) { | 
|---|
|  |  |  | logger.info("[国标级联]更新平台 {}", parentPlatform.getDeviceGBId()); | 
|---|
|  |  |  | parentPlatform.setCharacterSet(parentPlatform.getCharacterSet().toUpperCase()); | 
|---|
|  |  |  | ParentPlatform parentPlatformOld = platformMapper.getParentPlatById(parentPlatform.getId()); | 
|---|
|  |  |  | ParentPlatformCatch parentPlatformCatchOld = redisCatchStorage.queryPlatformCatchInfo(parentPlatformOld.getServerGBId()); | 
|---|
|  |  |  | parentPlatform.setUpdateTime(DateUtil.getNow()); | 
|---|
|  |  |  | if (!parentPlatformOld.getTreeType().equals(parentPlatform.getTreeType())) { | 
|---|
|  |  |  | // 目录结构发生变化,清空之前的关联关系 | 
|---|
|  |  |  | logger.info("保存平台{}时发现目录结构变化,清空关联关系", parentPlatform.getDeviceGBId()); | 
|---|
|  |  |  | catalogMapper.delByPlatformId(parentPlatformOld.getServerGBId()); | 
|---|
|  |  |  | platformChannelMapper.delByPlatformId(parentPlatformOld.getServerGBId()); | 
|---|
|  |  |  | platformGbStreamMapper.delByPlatformId(parentPlatformOld.getServerGBId()); | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  |  | 
|---|
|  |  |  | // 停止心跳定时 | 
|---|
|  |  |  | final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatformOld.getServerGBId(); | 
|---|
|  |  |  | dynamicTask.stop(keepaliveTaskKey); | 
|---|
|  |  |  | // 停止注册定时 | 
|---|
|  |  |  | final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatformOld.getServerGBId(); | 
|---|
|  |  |  | dynamicTask.stop(registerTaskKey); | 
|---|
|  |  |  | // 注销旧的 | 
|---|
|  |  |  | try { | 
|---|
|  |  |  | if (parentPlatformOld.isStatus()) { | 
|---|
|  |  |  | logger.info("保存平台{}时发现救平台在线,发送注销命令", parentPlatformOld.getServerGBId()); | 
|---|
|  |  |  | commanderForPlatform.unregister(parentPlatformOld, parentPlatformCatchOld.getSipTransactionInfo(), null, eventResult -> { | 
|---|
|  |  |  | logger.info("[国标级联] 注销成功, 平台:{}", parentPlatformOld.getServerGBId()); | 
|---|
|  |  |  | }); | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  | } catch (InvalidArgumentException | ParseException | SipException e) { | 
|---|
|  |  |  | logger.error("[命令发送失败] 国标级联 注销: {}", e.getMessage()); | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  | // 更新数据库 | 
|---|
|  |  |  | if (parentPlatform.getCatalogGroup() == 0) { | 
|---|
|  |  |  | parentPlatform.setCatalogGroup(1); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | if (parentPlatform.getAdministrativeDivision() == null) { | 
|---|
|  |  |  | parentPlatform.setAdministrativeDivision(parentPlatform.getAdministrativeDivision()); | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  | platformMapper.updateParentPlatform(parentPlatform); | 
|---|
|  |  |  | // 更新redis | 
|---|
|  |  |  | redisCatchStorage.delPlatformCatchInfo(parentPlatformOld.getServerGBId()); | 
|---|
|  |  |  | ParentPlatformCatch parentPlatformCatch = new ParentPlatformCatch(); | 
|---|
|  |  |  | parentPlatformCatch.setParentPlatform(parentPlatform); | 
|---|
|  |  |  | parentPlatformCatch.setId(parentPlatform.getServerGBId()); | 
|---|
|  |  |  | redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch); | 
|---|
|  |  |  | // 注册 | 
|---|
|  |  |  | if (parentPlatform.isEnable()) { | 
|---|
|  |  |  | // 保存时启用就发送注册 | 
|---|
|  |  |  | // 注册成功时由程序直接调用了online方法 | 
|---|
|  |  |  | try { | 
|---|
|  |  |  | logger.info("[国标级联] 平台注册 {}", parentPlatform.getDeviceGBId()); | 
|---|
|  |  |  | commanderForPlatform.register(parentPlatform, eventResult -> { | 
|---|
|  |  |  | logger.info("[国标级联] {},添加向上级注册失败,请确定上级平台可用时重新保存", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | }, null); | 
|---|
|  |  |  | } catch (InvalidArgumentException | ParseException | SipException e) { | 
|---|
|  |  |  | logger.error("[命令发送失败] 国标级联: {}", e.getMessage()); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | } | 
|---|
|  |  |  | // 重新开启定时注册, 使用续订消息 | 
|---|
|  |  |  | // 重新开始心跳保活 | 
|---|
|  |  |  |  | 
|---|
|  |  |  |  | 
|---|
|  |  |  | return false; | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Override | 
|---|
|  |  |  | public void online(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo) { | 
|---|
|  |  |  | logger.info("[国标级联]:{}, 平台上线", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), true); | 
|---|
|  |  |  | ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId()); | 
|---|
|  |  |  | if (parentPlatformCatch != null) { | 
|---|
|  |  |  | parentPlatformCatch.getParentPlatform().setStatus(true); | 
|---|
|  |  |  | redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch); | 
|---|
|  |  |  | }else { | 
|---|
|  |  |  | if (parentPlatformCatch == null) { | 
|---|
|  |  |  | parentPlatformCatch = new ParentPlatformCatch(); | 
|---|
|  |  |  | parentPlatformCatch.setParentPlatform(parentPlatform); | 
|---|
|  |  |  | parentPlatformCatch.setId(parentPlatform.getServerGBId()); | 
|---|
|  |  |  | parentPlatform.setStatus(true); | 
|---|
|  |  |  | parentPlatformCatch.setParentPlatform(parentPlatform); | 
|---|
|  |  |  | redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch); | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  | parentPlatformCatch.getParentPlatform().setStatus(true); | 
|---|
|  |  |  | parentPlatformCatch.setSipTransactionInfo(sipTransactionInfo); | 
|---|
|  |  |  | redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch); | 
|---|
|  |  |  |  | 
|---|
|  |  |  | final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId(); | 
|---|
|  |  |  | if (dynamicTask.contains(registerTaskKey)) { | 
|---|
|  |  |  | dynamicTask.stop(registerTaskKey); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | // 添加注册任务 | 
|---|
|  |  |  | dynamicTask.startDelay(registerTaskKey, | 
|---|
|  |  |  | if (!dynamicTask.isAlive(registerTaskKey)) { | 
|---|
|  |  |  | logger.info("[国标级联]:{}, 添加定时注册任务", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | // 添加注册任务 | 
|---|
|  |  |  | dynamicTask.startCron(registerTaskKey, | 
|---|
|  |  |  | // 注册失败(注册成功时由程序直接调用了online方法) | 
|---|
|  |  |  | ()-> { | 
|---|
|  |  |  | try { | 
|---|
|  |  |  | commanderForPlatform.register(parentPlatform, eventResult -> offline(parentPlatform),null); | 
|---|
|  |  |  | } catch (InvalidArgumentException | ParseException | SipException e) { | 
|---|
|  |  |  | logger.error("[命令发送失败] 国标级联定时注册: {}", e.getMessage()); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | }, | 
|---|
|  |  |  | (parentPlatform.getExpires() - 10) *1000); | 
|---|
|  |  |  | ()-> registerTask(parentPlatform, sipTransactionInfo), | 
|---|
|  |  |  | parentPlatform.getExpires() * 1000); | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  |  | 
|---|
|  |  |  | final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId(); | 
|---|
|  |  |  | if (!dynamicTask.contains(keepaliveTaskKey)) { | 
|---|
|  |  |  | logger.info("[国标级联]:{}, 添加定时心跳任务", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | // 添加心跳任务 | 
|---|
|  |  |  | dynamicTask.startCron(keepaliveTaskKey, | 
|---|
|  |  |  | ()-> { | 
|---|
|  |  |  | 
|---|
|  |  |  | // 此时是第三次心跳超时, 平台离线 | 
|---|
|  |  |  | if (platformCatch.getKeepAliveReply()  == 2) { | 
|---|
|  |  |  | // 设置平台离线,并重新注册 | 
|---|
|  |  |  | offline(parentPlatform); | 
|---|
|  |  |  | logger.info("[国标级联] {},三次心跳超时后再次发起注册", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | try { | 
|---|
|  |  |  | commanderForPlatform.register(parentPlatform, eventResult1 -> { | 
|---|
|  |  |  | logger.info("[国标级联] {},三次心跳超时后再次发起注册仍然失败,开始定时发起注册,间隔为1分钟", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | // 添加注册任务 | 
|---|
|  |  |  | dynamicTask.startCron(registerTaskKey, | 
|---|
|  |  |  | // 注册失败(注册成功时由程序直接调用了online方法) | 
|---|
|  |  |  | ()->logger.info("[国标级联] {},平台离线后持续发起注册,失败", parentPlatform.getServerGBId()), | 
|---|
|  |  |  | 60*1000); | 
|---|
|  |  |  | offline(parentPlatform, false); | 
|---|
|  |  |  | }, null); | 
|---|
|  |  |  | } catch (InvalidArgumentException | ParseException | SipException e) { | 
|---|
|  |  |  | logger.error("[命令发送失败] 国标级联 注册: {}", e.getMessage()); | 
|---|
|  |  |  | 
|---|
|  |  |  | // 心跳成功 | 
|---|
|  |  |  | // 清空之前的心跳超时计数 | 
|---|
|  |  |  | ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId()); | 
|---|
|  |  |  | if (platformCatch.getKeepAliveReply() > 0) { | 
|---|
|  |  |  | if (platformCatch != null && platformCatch.getKeepAliveReply() > 0) { | 
|---|
|  |  |  | platformCatch.setKeepAliveReply(0); | 
|---|
|  |  |  | redisCatchStorage.updatePlatformCatchInfo(platformCatch); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | 
|---|
|  |  |  | logger.error("[命令发送失败] 国标级联 发送心跳: {}", e.getMessage()); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | }, | 
|---|
|  |  |  | (parentPlatform.getKeepTimeout() - 10)*1000); | 
|---|
|  |  |  | (parentPlatform.getKeepTimeout())*1000); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  | private void registerTask(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo){ | 
|---|
|  |  |  | try { | 
|---|
|  |  |  | // 设置超时重发, 后续从底层支持消息重发 | 
|---|
|  |  |  | String key = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId() + "_timeout"; | 
|---|
|  |  |  | if (dynamicTask.isAlive(key)) { | 
|---|
|  |  |  | return; | 
|---|
|  |  |  | } | 
|---|
|  |  |  | dynamicTask.startDelay(key, ()->{ | 
|---|
|  |  |  | registerTask(parentPlatform, sipTransactionInfo); | 
|---|
|  |  |  | }, 1000); | 
|---|
|  |  |  | logger.info("[国标级联] 平台:{}注册即将到期,开始续订", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | commanderForPlatform.register(parentPlatform, sipTransactionInfo,  eventResult -> { | 
|---|
|  |  |  | dynamicTask.stop(key); | 
|---|
|  |  |  | offline(parentPlatform, false); | 
|---|
|  |  |  | },eventResult -> { | 
|---|
|  |  |  | dynamicTask.stop(key); | 
|---|
|  |  |  | }); | 
|---|
|  |  |  | } catch (InvalidArgumentException | ParseException | SipException e) { | 
|---|
|  |  |  | logger.error("[命令发送失败] 国标级联定时注册: {}", e.getMessage()); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | } | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Override | 
|---|
|  |  |  | public void offline(ParentPlatform parentPlatform) { | 
|---|
|  |  |  | public void offline(ParentPlatform parentPlatform, boolean stopRegister) { | 
|---|
|  |  |  | logger.info("[平台离线]:{}", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId()); | 
|---|
|  |  |  | parentPlatformCatch.setKeepAliveReply(0); | 
|---|
|  |  |  | 
|---|
|  |  |  | // 停止所有推流 | 
|---|
|  |  |  | logger.info("[平台离线] {}, 停止所有推流", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | stopAllPush(parentPlatform.getServerGBId()); | 
|---|
|  |  |  | // 清除注册定时 | 
|---|
|  |  |  | logger.info("[平台离线] {}, 停止定时注册任务", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId(); | 
|---|
|  |  |  | if (dynamicTask.contains(registerTaskKey)) { | 
|---|
|  |  |  | dynamicTask.stop(registerTaskKey); | 
|---|
|  |  |  | if (stopRegister) { | 
|---|
|  |  |  | // 清除注册定时 | 
|---|
|  |  |  | logger.info("[平台离线] {}, 停止定时注册任务", parentPlatform.getServerGBId()); | 
|---|
|  |  |  | final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId(); | 
|---|
|  |  |  | if (dynamicTask.contains(registerTaskKey)) { | 
|---|
|  |  |  | dynamicTask.stop(registerTaskKey); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | } | 
|---|
|  |  |  | // 清除心跳定时 | 
|---|
|  |  |  | logger.info("[平台离线] {}, 停止定时发送心跳任务", parentPlatform.getServerGBId()); | 
|---|