DeviceServiceImpl.java 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638
  1. package com.genersoft.iot.vmp.service.impl;
  2. import com.baomidou.dynamic.datasource.annotation.DS;
  3. import com.genersoft.iot.vmp.common.CommonCallback;
  4. import com.genersoft.iot.vmp.common.VideoManagerConstants;
  5. import com.genersoft.iot.vmp.conf.DynamicTask;
  6. import com.genersoft.iot.vmp.conf.UserSetting;
  7. import com.genersoft.iot.vmp.gb28181.bean.*;
  8. import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
  9. import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
  10. import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
  11. import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask;
  12. import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask;
  13. import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
  14. import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
  15. import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
  16. import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
  17. import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
  18. import com.genersoft.iot.vmp.service.IDeviceChannelService;
  19. import com.genersoft.iot.vmp.service.IDeviceService;
  20. import com.genersoft.iot.vmp.service.IInviteStreamService;
  21. import com.genersoft.iot.vmp.service.IMediaServerService;
  22. import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
  23. import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
  24. import com.genersoft.iot.vmp.storager.dao.DeviceMapper;
  25. import com.genersoft.iot.vmp.storager.dao.PlatformChannelMapper;
  26. import com.genersoft.iot.vmp.utils.DateUtil;
  27. import com.genersoft.iot.vmp.vmanager.bean.BaseTree;
  28. import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
  29. import org.slf4j.Logger;
  30. import org.slf4j.LoggerFactory;
  31. import org.springframework.beans.factory.annotation.Autowired;
  32. import org.springframework.jdbc.datasource.DataSourceTransactionManager;
  33. import org.springframework.stereotype.Service;
  34. import org.springframework.transaction.TransactionDefinition;
  35. import org.springframework.transaction.TransactionStatus;
  36. import org.springframework.util.ObjectUtils;
  37. import javax.sip.InvalidArgumentException;
  38. import javax.sip.SipException;
  39. import java.text.ParseException;
  40. import java.time.Instant;
  41. import java.util.*;
  42. import java.util.concurrent.TimeUnit;
  43. /**
  44. * 设备业务(目录订阅)
  45. */
  46. @Service
  47. @DS("master")
  48. public class DeviceServiceImpl implements IDeviceService {
  49. private final static Logger logger = LoggerFactory.getLogger(DeviceServiceImpl.class);
  50. @Autowired
  51. private SIPCommander cmder;
  52. @Autowired
  53. private DynamicTask dynamicTask;
  54. @Autowired
  55. private ISIPCommander sipCommander;
  56. @Autowired
  57. private CatalogResponseMessageHandler catalogResponseMessageHandler;
  58. @Autowired
  59. private IRedisCatchStorage redisCatchStorage;
  60. @Autowired
  61. private IInviteStreamService inviteStreamService;
  62. @Autowired
  63. private DeviceMapper deviceMapper;
  64. @Autowired
  65. private PlatformChannelMapper platformChannelMapper;
  66. @Autowired
  67. private IDeviceChannelService deviceChannelService;
  68. @Autowired
  69. private DeviceChannelMapper deviceChannelMapper;
  70. @Autowired
  71. DataSourceTransactionManager dataSourceTransactionManager;
  72. @Autowired
  73. TransactionDefinition transactionDefinition;
  74. @Autowired
  75. private UserSetting userSetting;
  76. @Autowired
  77. private ISIPCommander commander;
  78. @Autowired
  79. private VideoStreamSessionManager streamSession;
  80. @Autowired
  81. private IMediaServerService mediaServerService;
  82. @Autowired
  83. private AudioBroadcastManager audioBroadcastManager;
  84. @Autowired
  85. private ZLMRESTfulUtils zlmresTfulUtils;
  86. @Override
  87. public void online(Device device, SipTransactionInfo sipTransactionInfo) {
  88. logger.info("[设备上线] deviceId:{}->{}:{}", device.getDeviceId(), device.getIp(), device.getPort());
  89. Device deviceInRedis = redisCatchStorage.getDevice(device.getDeviceId());
  90. Device deviceInDb = deviceMapper.getDeviceByDeviceId(device.getDeviceId());
  91. String now = DateUtil.getNow();
  92. if (deviceInRedis != null && deviceInDb == null) {
  93. // redis 存在脏数据
  94. inviteStreamService.clearInviteInfo(device.getDeviceId());
  95. }
  96. device.setUpdateTime(now);
  97. device.setKeepaliveTime(now);
  98. if (device.getKeepaliveIntervalTime() == 0) {
  99. // 默认心跳间隔60
  100. device.setKeepaliveIntervalTime(60);
  101. }
  102. if (sipTransactionInfo != null) {
  103. device.setSipTransactionInfo(sipTransactionInfo);
  104. }else {
  105. if (deviceInRedis != null) {
  106. device.setSipTransactionInfo(deviceInRedis.getSipTransactionInfo());
  107. }
  108. }
  109. // 第一次上线 或则设备之前是离线状态--进行通道同步和设备信息查询
  110. if (deviceInDb == null) {
  111. device.setOnLine(true);
  112. device.setCreateTime(now);
  113. device.setUpdateTime(now);
  114. logger.info("[设备上线,首次注册]: {},查询设备信息以及通道信息", device.getDeviceId());
  115. deviceMapper.add(device);
  116. redisCatchStorage.updateDevice(device);
  117. try {
  118. commander.deviceInfoQuery(device);
  119. } catch (InvalidArgumentException | SipException | ParseException e) {
  120. logger.error("[命令发送失败] 查询设备信息: {}", e.getMessage());
  121. }
  122. sync(device);
  123. }else {
  124. if(!device.isOnLine()){
  125. device.setOnLine(true);
  126. device.setCreateTime(now);
  127. deviceMapper.update(device);
  128. redisCatchStorage.updateDevice(device);
  129. if (userSetting.getSyncChannelOnDeviceOnline()) {
  130. logger.info("[设备上线,离线状态下重新注册]: {},查询设备信息以及通道信息", device.getDeviceId());
  131. try {
  132. commander.deviceInfoQuery(device);
  133. } catch (InvalidArgumentException | SipException | ParseException e) {
  134. logger.error("[命令发送失败] 查询设备信息: {}", e.getMessage());
  135. }
  136. sync(device);
  137. // TODO 如果设备下的通道级联到了其他平台,那么需要发送事件或者notify给上级平台
  138. }
  139. // 上线添加订阅
  140. if (device.getSubscribeCycleForCatalog() > 0) {
  141. // 查询在线设备那些开启了订阅,为设备开启定时的目录订阅
  142. addCatalogSubscribe(device);
  143. }
  144. if (device.getSubscribeCycleForMobilePosition() > 0) {
  145. addMobilePositionSubscribe(device);
  146. }
  147. if (userSetting.getDeviceStatusNotify()) {
  148. // 发送redis消息
  149. redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, true);
  150. }
  151. }else {
  152. if (deviceChannelMapper.queryAllChannels(device.getDeviceId()).size() == 0) {
  153. logger.info("[设备上线]: {},通道数为0,查询通道信息", device.getDeviceId());
  154. sync(device);
  155. }
  156. deviceMapper.update(device);
  157. redisCatchStorage.updateDevice(device);
  158. }
  159. }
  160. // 刷新过期任务
  161. String registerExpireTaskKey = VideoManagerConstants.REGISTER_EXPIRE_TASK_KEY_PREFIX + device.getDeviceId();
  162. // 如果第一次注册那么必须在60 * 3时间内收到一个心跳,否则设备离线
  163. dynamicTask.startDelay(registerExpireTaskKey, ()-> offline(device.getDeviceId(), "首次注册后未能收到心跳"), device.getKeepaliveIntervalTime() * 1000 * 3);
  164. //
  165. // try {
  166. // cmder.alarmSubscribe(device, 600, "0", "4", "0", "2023-7-27T00:00:00", "2023-7-28T00:00:00");
  167. // } catch (InvalidArgumentException e) {
  168. // throw new RuntimeException(e);
  169. // } catch (SipException e) {
  170. // throw new RuntimeException(e);
  171. // } catch (ParseException e) {
  172. // throw new RuntimeException(e);
  173. // }
  174. }
  175. @Override
  176. public void offline(String deviceId, String reason) {
  177. logger.warn("[设备离线],{}, device:{}", reason, deviceId);
  178. Device device = deviceMapper.getDeviceByDeviceId(deviceId);
  179. if (device == null) {
  180. return;
  181. }
  182. String registerExpireTaskKey = VideoManagerConstants.REGISTER_EXPIRE_TASK_KEY_PREFIX + deviceId;
  183. dynamicTask.stop(registerExpireTaskKey);
  184. if (device.isOnLine()) {
  185. if (userSetting.getDeviceStatusNotify()) {
  186. // 发送redis消息
  187. redisCatchStorage.sendDeviceOrChannelStatus(device.getDeviceId(), null, false);
  188. }
  189. }
  190. device.setOnLine(false);
  191. redisCatchStorage.updateDevice(device);
  192. deviceMapper.update(device);
  193. //进行通道离线
  194. // deviceChannelMapper.offlineByDeviceId(deviceId);
  195. // 离线释放所有ssrc
  196. List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(deviceId, null, null, null);
  197. if (ssrcTransactions != null && ssrcTransactions.size() > 0) {
  198. for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
  199. mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
  200. mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
  201. streamSession.removeByCallId(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getCallId());
  202. }
  203. }
  204. // 移除订阅
  205. removeCatalogSubscribe(device, null);
  206. removeMobilePositionSubscribe(device, null);
  207. List<AudioBroadcastCatch> audioBroadcastCatches = audioBroadcastManager.get(deviceId);
  208. if (audioBroadcastCatches.size() > 0) {
  209. for (AudioBroadcastCatch audioBroadcastCatch : audioBroadcastCatches) {
  210. SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null);
  211. if (sendRtpItem != null) {
  212. redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null);
  213. MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
  214. Map<String, Object> param = new HashMap<>();
  215. param.put("vhost", "__defaultVhost__");
  216. param.put("app", sendRtpItem.getApp());
  217. param.put("stream", sendRtpItem.getStream());
  218. zlmresTfulUtils.stopSendRtp(mediaInfo, param);
  219. }
  220. audioBroadcastManager.del(deviceId, audioBroadcastCatch.getChannelId());
  221. }
  222. }
  223. }
  224. @Override
  225. public boolean addCatalogSubscribe(Device device) {
  226. if (device == null || device.getSubscribeCycleForCatalog() < 0) {
  227. return false;
  228. }
  229. logger.info("[添加目录订阅] 设备{}", device.getDeviceId());
  230. // 添加目录订阅
  231. CatalogSubscribeTask catalogSubscribeTask = new CatalogSubscribeTask(device, sipCommander, dynamicTask);
  232. // 刷新订阅
  233. int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForCatalog(),30);
  234. // 设置最小值为30
  235. dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, (subscribeCycleForCatalog -1) * 1000);
  236. return true;
  237. }
  238. @Override
  239. public boolean removeCatalogSubscribe(Device device, CommonCallback<Boolean> callback) {
  240. if (device == null || device.getSubscribeCycleForCatalog() < 0) {
  241. return false;
  242. }
  243. logger.info("[移除目录订阅]: {}", device.getDeviceId());
  244. String taskKey = device.getDeviceId() + "catalog";
  245. if (device.isOnLine()) {
  246. Runnable runnable = dynamicTask.get(taskKey);
  247. if (runnable instanceof ISubscribeTask) {
  248. ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
  249. subscribeTask.stop(callback);
  250. }
  251. }
  252. dynamicTask.stop(taskKey);
  253. return true;
  254. }
  255. @Override
  256. public boolean addMobilePositionSubscribe(Device device) {
  257. if (device == null || device.getSubscribeCycleForMobilePosition() < 0) {
  258. return false;
  259. }
  260. logger.info("[添加移动位置订阅] 设备{}", device.getDeviceId());
  261. // 添加目录订阅
  262. MobilePositionSubscribeTask mobilePositionSubscribeTask = new MobilePositionSubscribeTask(device, sipCommander, dynamicTask);
  263. // 设置最小值为30
  264. int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForMobilePosition(),30);
  265. // 刷新订阅
  266. dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, subscribeCycleForCatalog * 1000);
  267. return true;
  268. }
  269. @Override
  270. public boolean removeMobilePositionSubscribe(Device device, CommonCallback<Boolean> callback) {
  271. if (device == null || device.getSubscribeCycleForCatalog() < 0) {
  272. return false;
  273. }
  274. logger.info("[移除移动位置订阅]: {}", device.getDeviceId());
  275. String taskKey = device.getDeviceId() + "mobile_position";
  276. if (device.isOnLine()) {
  277. Runnable runnable = dynamicTask.get(taskKey);
  278. if (runnable instanceof ISubscribeTask) {
  279. ISubscribeTask subscribeTask = (ISubscribeTask) runnable;
  280. subscribeTask.stop(callback);
  281. }
  282. }
  283. dynamicTask.stop(taskKey);
  284. return true;
  285. }
  286. @Override
  287. public SyncStatus getChannelSyncStatus(String deviceId) {
  288. return catalogResponseMessageHandler.getChannelSyncProgress(deviceId);
  289. }
  290. @Override
  291. public Boolean isSyncRunning(String deviceId) {
  292. return catalogResponseMessageHandler.isSyncRunning(deviceId);
  293. }
  294. @Override
  295. public void sync(Device device) {
  296. if (catalogResponseMessageHandler.isSyncRunning(device.getDeviceId())) {
  297. logger.info("开启同步时发现同步已经存在");
  298. return;
  299. }
  300. int sn = (int)((Math.random()*9+1)*100000);
  301. catalogResponseMessageHandler.setChannelSyncReady(device, sn);
  302. try {
  303. sipCommander.catalogQuery(device, sn, event -> {
  304. String errorMsg = String.format("同步通道失败,错误码: %s, %s", event.statusCode, event.msg);
  305. catalogResponseMessageHandler.setChannelSyncEnd(device.getDeviceId(), errorMsg);
  306. });
  307. } catch (SipException | InvalidArgumentException | ParseException e) {
  308. logger.error("[同步通道], 信令发送失败:{}", e.getMessage() );
  309. String errorMsg = String.format("同步通道失败,信令发送失败: %s", e.getMessage());
  310. catalogResponseMessageHandler.setChannelSyncEnd(device.getDeviceId(), errorMsg);
  311. }
  312. }
  313. @Override
  314. public Device getDevice(String deviceId) {
  315. Device device = redisCatchStorage.getDevice(deviceId);
  316. if (device == null) {
  317. device = deviceMapper.getDeviceByDeviceId(deviceId);
  318. if (device != null) {
  319. redisCatchStorage.updateDevice(device);
  320. }
  321. }
  322. return device;
  323. }
  324. @Override
  325. public List<Device> getAllOnlineDevice() {
  326. return deviceMapper.getOnlineDevices();
  327. }
  328. @Override
  329. public boolean expire(Device device) {
  330. Instant registerTimeDate = Instant.from(DateUtil.formatter.parse(device.getRegisterTime()));
  331. Instant expireInstant = registerTimeDate.plusMillis(TimeUnit.SECONDS.toMillis(device.getExpires()));
  332. return expireInstant.isBefore(Instant.now());
  333. }
  334. @Override
  335. public void checkDeviceStatus(Device device) {
  336. if (device == null || !device.isOnLine()) {
  337. return;
  338. }
  339. try {
  340. sipCommander.deviceStatusQuery(device, null);
  341. } catch (InvalidArgumentException | SipException | ParseException e) {
  342. logger.error("[命令发送失败] 设备状态查询: {}", e.getMessage());
  343. }
  344. }
  345. @Override
  346. public Device getDeviceByHostAndPort(String host, int port) {
  347. return deviceMapper.getDeviceByHostAndPort(host, port);
  348. }
  349. @Override
  350. public void updateDevice(Device device) {
  351. String now = DateUtil.getNow();
  352. device.setUpdateTime(now);
  353. device.setCharset(device.getCharset().toUpperCase());
  354. device.setUpdateTime(DateUtil.getNow());
  355. if (deviceMapper.update(device) > 0) {
  356. redisCatchStorage.updateDevice(device);
  357. }
  358. }
  359. /**
  360. * 更新通道坐标系
  361. */
  362. private void updateDeviceChannelGeoCoordSys(Device device) {
  363. List<DeviceChannel> deviceChannels = deviceChannelMapper.getAllChannelWithCoordinate(device.getDeviceId());
  364. if (deviceChannels.size() > 0) {
  365. List<DeviceChannel> deviceChannelsForStore = new ArrayList<>();
  366. for (DeviceChannel deviceChannel : deviceChannels) {
  367. deviceChannelsForStore.add(deviceChannelService.updateGps(deviceChannel, device));
  368. }
  369. deviceChannelService.updateChannels(device.getDeviceId(), deviceChannelsForStore);
  370. }
  371. }
  372. @Override
  373. public List<BaseTree<DeviceChannel>> queryVideoDeviceTree(String deviceId, String parentId, boolean onlyCatalog) {
  374. Device device = deviceMapper.getDeviceByDeviceId(deviceId);
  375. if (device == null) {
  376. return null;
  377. }
  378. if (ObjectUtils.isEmpty(parentId) ) {
  379. parentId = deviceId;
  380. }
  381. List<DeviceChannel> rootNodes = deviceChannelMapper.getSubChannelsByDeviceId(deviceId, parentId, onlyCatalog);
  382. return transportChannelsToTree(rootNodes, "");
  383. }
  384. @Override
  385. public List<DeviceChannel> queryVideoDeviceInTreeNode(String deviceId, String parentId) {
  386. Device device = deviceMapper.getDeviceByDeviceId(deviceId);
  387. if (device == null) {
  388. return null;
  389. }
  390. if (ObjectUtils.isEmpty(parentId) || parentId.equals(deviceId)) {
  391. return deviceChannelMapper.getSubChannelsByDeviceId(deviceId, null, false);
  392. }else {
  393. return deviceChannelMapper.getSubChannelsByDeviceId(deviceId, parentId, false);
  394. }
  395. }
  396. private List<BaseTree<DeviceChannel>> transportChannelsToTree(List<DeviceChannel> channels, String parentId) {
  397. if (channels == null) {
  398. return null;
  399. }
  400. List<BaseTree<DeviceChannel>> treeNotes = new ArrayList<>();
  401. if (channels.size() == 0) {
  402. return treeNotes;
  403. }
  404. for (DeviceChannel channel : channels) {
  405. BaseTree<DeviceChannel> node = new BaseTree<>();
  406. node.setId(channel.getChannelId());
  407. node.setDeviceId(channel.getDeviceId());
  408. node.setName(channel.getName());
  409. node.setPid(parentId);
  410. node.setBasicData(channel);
  411. node.setParent(false);
  412. if (channel.getChannelId().length() <= 8) {
  413. node.setParent(true);
  414. }else {
  415. if (channel.getChannelId().length() != 20) {
  416. node.setParent(channel.getParental() == 1);
  417. }else {
  418. try {
  419. int type = Integer.parseInt(channel.getChannelId().substring(10, 13));
  420. if (type == 215 || type == 216 || type == 200) {
  421. node.setParent(true);
  422. }
  423. }catch (NumberFormatException e) {
  424. node.setParent(false);
  425. }
  426. }
  427. }
  428. treeNotes.add(node);
  429. }
  430. Collections.sort(treeNotes);
  431. return treeNotes;
  432. }
  433. @Override
  434. public boolean isExist(String deviceId) {
  435. return deviceMapper.getDeviceByDeviceId(deviceId) != null;
  436. }
  437. @Override
  438. public void addDevice(Device device) {
  439. device.setOnLine(false);
  440. device.setCreateTime(DateUtil.getNow());
  441. device.setUpdateTime(DateUtil.getNow());
  442. deviceMapper.addCustomDevice(device);
  443. }
  444. @Override
  445. public void updateCustomDevice(Device device) {
  446. Device deviceInStore = deviceMapper.getDeviceByDeviceId(device.getDeviceId());
  447. if (deviceInStore == null) {
  448. logger.warn("更新设备时未找到设备信息");
  449. return;
  450. }
  451. if (!ObjectUtils.isEmpty(device.getName())) {
  452. deviceInStore.setName(device.getName());
  453. }
  454. if (!ObjectUtils.isEmpty(device.getCharset())) {
  455. deviceInStore.setCharset(device.getCharset());
  456. }
  457. if (!ObjectUtils.isEmpty(device.getMediaServerId())) {
  458. deviceInStore.setMediaServerId(device.getMediaServerId());
  459. }
  460. if (!ObjectUtils.isEmpty(device.getCharset())) {
  461. deviceInStore.setCharset(device.getCharset());
  462. }
  463. if (!ObjectUtils.isEmpty(device.getSdpIp())) {
  464. deviceInStore.setSdpIp(device.getSdpIp());
  465. }
  466. if (!ObjectUtils.isEmpty(device.getPassword())) {
  467. deviceInStore.setPassword(device.getPassword());
  468. }
  469. if (!ObjectUtils.isEmpty(device.getStreamMode())) {
  470. deviceInStore.setStreamMode(device.getStreamMode());
  471. }
  472. // 目录订阅相关的信息
  473. if (deviceInStore.getSubscribeCycleForCatalog() != device.getSubscribeCycleForCatalog()) {
  474. if (device.getSubscribeCycleForCatalog() > 0) {
  475. // 若已开启订阅,但订阅周期不同,则先取消
  476. if (deviceInStore.getSubscribeCycleForCatalog() != 0) {
  477. removeCatalogSubscribe(deviceInStore, result->{
  478. // 开启订阅
  479. deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
  480. addCatalogSubscribe(deviceInStore);
  481. // 因为是异步执行,需要在这里更新下数据
  482. deviceMapper.updateCustom(deviceInStore);
  483. redisCatchStorage.updateDevice(deviceInStore);
  484. });
  485. }else {
  486. // 开启订阅
  487. deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
  488. addCatalogSubscribe(deviceInStore);
  489. }
  490. }else if (device.getSubscribeCycleForCatalog() == 0) {
  491. // 取消订阅
  492. deviceInStore.setSubscribeCycleForCatalog(0);
  493. removeCatalogSubscribe(deviceInStore, null);
  494. }
  495. }
  496. // 移动位置订阅相关的信息
  497. if (deviceInStore.getSubscribeCycleForMobilePosition() != device.getSubscribeCycleForMobilePosition()) {
  498. if (device.getSubscribeCycleForMobilePosition() > 0) {
  499. // 若已开启订阅,但订阅周期不同,则先取消
  500. if (deviceInStore.getSubscribeCycleForMobilePosition() != 0) {
  501. removeMobilePositionSubscribe(deviceInStore, result->{
  502. // 开启订阅
  503. deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
  504. addMobilePositionSubscribe(deviceInStore);
  505. // 因为是异步执行,需要在这里更新下数据
  506. deviceMapper.updateCustom(deviceInStore);
  507. redisCatchStorage.updateDevice(deviceInStore);
  508. });
  509. }else {
  510. // 开启订阅
  511. deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
  512. addMobilePositionSubscribe(deviceInStore);
  513. }
  514. }else if (device.getSubscribeCycleForMobilePosition() == 0) {
  515. // 取消订阅
  516. deviceInStore.setSubscribeCycleForCatalog(0);
  517. removeCatalogSubscribe(deviceInStore, null);
  518. }
  519. }
  520. if (deviceInStore.getGeoCoordSys() != null) {
  521. // 坐标系变化,需要重新计算GCJ02坐标和WGS84坐标
  522. if (!deviceInStore.getGeoCoordSys().equals(device.getGeoCoordSys())) {
  523. deviceInStore.setGeoCoordSys(device.getGeoCoordSys());
  524. updateDeviceChannelGeoCoordSys(deviceInStore);
  525. }
  526. }else {
  527. deviceInStore.setGeoCoordSys("WGS84");
  528. }
  529. if (device.getCharset() == null) {
  530. deviceInStore.setCharset("GB2312");
  531. }
  532. //SSRC校验
  533. deviceInStore.setSsrcCheck(device.isSsrcCheck());
  534. //作为消息通道
  535. deviceInStore.setAsMessageChannel(device.isAsMessageChannel());
  536. deviceMapper.updateCustom(deviceInStore);
  537. redisCatchStorage.updateDevice(deviceInStore);
  538. }
  539. @Override
  540. public boolean delete(String deviceId) {
  541. TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
  542. boolean result = false;
  543. try {
  544. platformChannelMapper.delChannelForDeviceId(deviceId);
  545. deviceChannelMapper.cleanChannelsByDeviceId(deviceId);
  546. if ( deviceMapper.del(deviceId) < 0 ) {
  547. //事务回滚
  548. dataSourceTransactionManager.rollback(transactionStatus);
  549. }
  550. result = true;
  551. dataSourceTransactionManager.commit(transactionStatus); //手动提交
  552. }catch (Exception e) {
  553. dataSourceTransactionManager.rollback(transactionStatus);
  554. }
  555. if (result) {
  556. redisCatchStorage.removeDevice(deviceId);
  557. }
  558. return result;
  559. }
  560. @Override
  561. public ResourceBaseInfo getOverview() {
  562. List<Device> onlineDevices = deviceMapper.getOnlineDevices();
  563. List<Device> all = deviceMapper.getAll();
  564. return new ResourceBaseInfo(all.size(), onlineDevices.size());
  565. }
  566. @Override
  567. public List<Device> getAll() {
  568. return deviceMapper.getAll();
  569. }
  570. }