guzijian / wvp-GB28181-pro · Commits

Commit 829ff200b8e9069d99d6d00fc14577e581d453f3

Authored by 648540858
2 parents 99506108 f4f3e60a

Merge branch 'wvp-28181-2.0' into wvp-28181-record 

# Conflicts:
#	sql/2.6.9更新.sql
Inline Side-by-side
Showing 9 changed files with 86 additions and 121 deletions
sql/2.6.9更新.sql 0 → 100644
  View file @829ff20
  1 +alter table wvp_device_channel
  2 + change stream_id stream_id varying(255)
  3 +
  4 +alter table wvp_platform
  5 + add auto_push_channel bool default false
  6 +
  7 +alter table wvp_stream_proxy
  8 + add stream_key character varying(255)
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/CatalogSubscribeTask.java
  View file @829ff20
... ... @@ -89,17 +89,17 @@ public class CatalogSubscribeTask implements ISubscribeTask {
89 89 ResponseEvent event = (ResponseEvent) eventResult.event;
90 90 if (event.getResponse().getRawContent() != null) {
91 91 // 成功
92   - logger.info("[取消目录订阅订阅]成功： {}", device.getDeviceId());
  92 + logger.info("[取消目录订阅]成功： {}", device.getDeviceId());
93 93 }else {
94 94 // 成功
95   - logger.info("[取消目录订阅订阅]成功： {}", device.getDeviceId());
  95 + logger.info("[取消目录订阅]成功： {}", device.getDeviceId());
96 96 }
97 97 },eventResult -> {
98 98 // 失败
99   - logger.warn("[取消目录订阅订阅]失败，信令发送失败： {}-{} ", device.getDeviceId(), eventResult.msg);
  99 + logger.warn("[取消目录订阅]失败，信令发送失败： {}-{} ", device.getDeviceId(), eventResult.msg);
100 100 });
101 101 } catch (InvalidArgumentException | SipException | ParseException e) {
102   - logger.error("[命令发送失败] 取消目录订阅订阅: {}", e.getMessage());
  102 + logger.error("[命令发送失败] 取消目录订阅: {}", e.getMessage());
103 103 }
104 104 }
105 105 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
  View file @829ff20
... ... @@ -132,7 +132,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
132 132  
133 133 if (CmdType.CATALOG.equals(cmd)) {
134 134 logger.info("接收到Catalog通知");
135   - processNotifyCatalogList(take.getEvt());
136 135 notifyRequestForCatalogProcessor.process(take.getEvt());
137 136 } else if (CmdType.ALARM.equals(cmd)) {
138 137 logger.info("接收到Alarm通知");
... ... @@ -371,114 +370,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
371 370 }
372 371 }
373 372  
374   - /***
375   - * 处理catalog设备目录列表Notify
376   - *
377   - * @param evt
378   - */
379   - private void processNotifyCatalogList(RequestEvent evt) {
380   - try {
381   - FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
382   - String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);
383   -
384   - Device device = redisCatchStorage.getDevice(deviceId);
385   - if (device == null || !device.isOnLine()) {
386   - logger.warn("[收到目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId():"" ));
387   - return;
388   - }
389   - Element rootElement = getRootElement(evt, device.getCharset());
390   - if (rootElement == null) {
391   - logger.warn("[ 收到目录订阅 ] content cannot be null, {}", evt.getRequest());
392   - return;
393   - }
394   - Element deviceListElement = rootElement.element("DeviceList");
395   - if (deviceListElement == null) {
396   - return;
397   - }
398   - Iterator<Element> deviceListIterator = deviceListElement.elementIterator();
399   - if (deviceListIterator != null) {
400   -
401   - // 遍历DeviceList
402   - while (deviceListIterator.hasNext()) {
403   - Element itemDevice = deviceListIterator.next();
404   - Element channelDeviceElement = itemDevice.element("DeviceID");
405   - if (channelDeviceElement == null) {
406   - continue;
407   - }
408   - Element eventElement = itemDevice.element("Event");
409   - String event;
410   - if (eventElement == null) {
411   - logger.warn("[收到目录订阅]：{}, 但是Event为空, 设为默认值 ADD", (device != null ? device.getDeviceId():"" ));
412   - event = CatalogEvent.ADD;
413   - }else {
414   - event = eventElement.getText().toUpperCase();
415   - }
416   - DeviceChannel channel = XmlUtil.channelContentHandler(itemDevice, device, event, civilCodeFileConf);
417   - if (channel == null) {
418   - logger.info("[收到目录订阅]：但是解析失败 {}", new String(evt.getRequest().getRawContent()));
419   - continue;
420   - }
421   - if (channel.getParentId() != null && channel.getParentId().equals(sipConfig.getId())) {
422   - channel.setParentId(null);
423   - }
424   - channel.setDeviceId(device.getDeviceId());
425   - logger.info("[收到目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
426   - switch (event) {
427   - case CatalogEvent.ON:
428   - // 上线
429   - logger.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
430   - storager.deviceChannelOnline(deviceId, channel.getChannelId());
431   - break;
432   - case CatalogEvent.OFF :
433   - // 离线
434   - logger.info("[收到通道离线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
435   - if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
436   - storager.deviceChannelOffline(deviceId, channel.getChannelId());
437   - }else {
438   - logger.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
439   - }
440   - break;
441   - case CatalogEvent.VLOST:
442   - // 视频丢失
443   - logger.info("[收到通道视频丢失通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
444   - if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
445   - storager.deviceChannelOffline(deviceId, channel.getChannelId());
446   - }else {
447   - logger.info("[收到通道视频丢失通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
448   - }
449   - break;
450   - case CatalogEvent.DEFECT:
451   - // 故障
452   - break;
453   - case CatalogEvent.ADD:
454   - // 增加
455   - logger.info("[收到增加通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
456   - deviceChannelService.updateChannel(deviceId, channel);
457   - break;
458   - case CatalogEvent.DEL:
459   - // 删除
460   - logger.info("[收到删除通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
461   - storager.delChannel(deviceId, channel.getChannelId());
462   - break;
463   - case CatalogEvent.UPDATE:
464   - // 更新
465   - logger.info("[收到更新通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
466   - deviceChannelService.updateChannel(deviceId, channel);
467   - break;
468   - default:
469   - logger.warn("[ NotifyCatalog ] event not found ： {}", event );
470   -
471   - }
472   - // 转发变化信息
473   - eventPublisher.catalogEventPublish(null, channel, event);
474   -
475   - }
476   - }
477   - } catch (DocumentException e) {
478   - logger.error("未处理的异常 ", e);
479   - }
480   - }
481   -
482 373 public void setCmder(SIPCommander cmder) {
483 374 }
484 375  
... ...
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
  View file @829ff20
... ... @@ -215,6 +215,21 @@ public class ZLMRESTfulUtils {
215 215 }
216 216 }
217 217  
  218 + public JSONObject isMediaOnline(MediaServerItem mediaServerItem, String app, String stream, String schema){
  219 + Map<String, Object> param = new HashMap<>();
  220 + if (app != null) {
  221 + param.put("app",app);
  222 + }
  223 + if (stream != null) {
  224 + param.put("stream",stream);
  225 + }
  226 + if (schema != null) {
  227 + param.put("schema",schema);
  228 + }
  229 + param.put("vhost","__defaultVhost__");
  230 + return sendPost(mediaServerItem, "isMediaOnline", param, null);
  231 + }
  232 +
218 233 public JSONObject getMediaList(MediaServerItem mediaServerItem, String app, String stream, String schema, RequestCallback callback){
219 234 Map<String, Object> param = new HashMap<>();
220 235 if (app != null) {
... ...
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java
  View file @829ff20
... ... @@ -9,6 +9,7 @@ import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
9 9 import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
10 10 import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForServerStarted;
11 11 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
  12 +import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
12 13 import com.genersoft.iot.vmp.service.IMediaServerService;
13 14 import org.slf4j.Logger;
14 15 import org.slf4j.LoggerFactory;
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
  View file @829ff20
... ... @@ -520,16 +520,18 @@ public class DeviceServiceImpl implements IDeviceService {
520 520  
521 521  
522 522 // 目录订阅相关的信息
523   - if (device.getSubscribeCycleForCatalog() > 0) {
524   - if (deviceInStore.getSubscribeCycleForCatalog() == 0 || deviceInStore.getSubscribeCycleForCatalog() != device.getSubscribeCycleForCatalog()) {
525   - deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
  523 + if (deviceInStore.getSubscribeCycleForCatalog() != device.getSubscribeCycleForCatalog()) {
  524 + if (device.getSubscribeCycleForCatalog() > 0) {
  525 + // 若已开启订阅，但订阅周期不同，则先取消
  526 + if (deviceInStore.getSubscribeCycleForCatalog() != 0) {
  527 + removeCatalogSubscribe(deviceInStore);
  528 + }
526 529 // 开启订阅
527   - addCatalogSubscribe(deviceInStore);
528   - }
529   - }else if (device.getSubscribeCycleForCatalog() == 0) {
530   - if (deviceInStore.getSubscribeCycleForCatalog() != 0) {
531 530 deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
  531 + addCatalogSubscribe(deviceInStore);
  532 + }else if (device.getSubscribeCycleForCatalog() == 0) {
532 533 // 取消订阅
  534 + deviceInStore.setSubscribeCycleForCatalog(device.getSubscribeCycleForCatalog());
533 535 removeCatalogSubscribe(deviceInStore);
534 536 }
535 537 }
... ... @@ -544,6 +546,8 @@ public class DeviceServiceImpl implements IDeviceService {
544 546 }
545 547 }else if (device.getSubscribeCycleForMobilePosition() == 0) {
546 548 if (deviceInStore.getSubscribeCycleForMobilePosition() != 0) {
  549 + deviceInStore.setMobilePositionSubmissionInterval(device.getMobilePositionSubmissionInterval());
  550 + deviceInStore.setSubscribeCycleForMobilePosition(device.getSubscribeCycleForMobilePosition());
547 551 // 取消订阅
548 552 removeMobilePositionSubscribe(deviceInStore);
549 553 }
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java
  View file @829ff20
... ... @@ -257,7 +257,7 @@ public class InviteStreamServiceImpl implements IInviteStreamService {
257 257 ":" + inviteInfo.getDeviceId() +
258 258 ":" + inviteInfo.getChannelId() +
259 259 ":" + inviteInfo.getStream() +
260   - ":" + inviteInfo.getSsrcInfo().getSsrc();
  260 + ":" + ssrc;
261 261 if (inviteInfoInDb.getSsrcInfo() != null) {
262 262 inviteInfoInDb.getSsrcInfo().setSsrc(ssrc);
263 263 }
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
  View file @829ff20
... ... @@ -35,15 +35,19 @@ import org.slf4j.Logger;
35 35 import org.slf4j.LoggerFactory;
36 36 import org.springframework.beans.factory.annotation.Autowired;
37 37 import org.springframework.jdbc.datasource.DataSourceTransactionManager;
  38 +import org.springframework.scheduling.annotation.Scheduled;
38 39 import org.springframework.stereotype.Service;
39 40 import org.springframework.transaction.TransactionDefinition;
40 41 import org.springframework.transaction.TransactionStatus;
  42 +import org.springframework.util.CollectionUtils;
41 43 import org.springframework.util.ObjectUtils;
42 44  
43 45 import java.util.HashMap;
44 46 import java.util.List;
45 47 import java.util.Map;
46 48 import java.util.UUID;
  49 +import java.util.function.Function;
  50 +import java.util.stream.Collectors;
47 51  
48 52 /**
49 53 * 视频代理业务
... ... @@ -560,4 +564,43 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
560 564  
561 565 return new ResourceBaseInfo(total, online);
562 566 }
  567 +
  568 +
  569 + @Scheduled(cron = "* 0/10 * * * ?")
  570 + public void asyncCheckStreamProxyStatus() {
  571 +
  572 + List<MediaServerItem> all = mediaServerService.getAllOnline();
  573 +
  574 + if (CollectionUtils.isEmpty(all)){
  575 + return;
  576 + }
  577 +
  578 + Map<String, MediaServerItem> serverItemMap = all.stream().collect(Collectors.toMap(MediaServerItem::getId, Function.identity(), (m1, m2) -> m1));
  579 +
  580 + List<StreamProxyItem> list = videoManagerStorager.getStreamProxyListForEnable(true);
  581 +
  582 + if (CollectionUtils.isEmpty(list)){
  583 + return;
  584 + }
  585 +
  586 + for (StreamProxyItem streamProxyItem : list) {
  587 +
  588 + MediaServerItem mediaServerItem = serverItemMap.get(streamProxyItem.getMediaServerId());
  589 +
  590 + // TODO 支持其他 schema
  591 + JSONObject mediaInfo = zlmresTfulUtils.isMediaOnline(mediaServerItem, streamProxyItem.getApp(), streamProxyItem.getStream(), "rtsp");
  592 +
  593 + if (mediaInfo == null){
  594 + streamProxyItem.setStatus(false);
  595 + } else {
  596 + if (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")) {
  597 + streamProxyItem.setStatus(true);
  598 + } else {
  599 + streamProxyItem.setStatus(false);
  600 + }
  601 + }
  602 +
  603 + updateStreamProxy(streamProxyItem);
  604 + }
  605 + }
563 606 }
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
  View file @829ff20
... ... @@ -506,6 +506,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
506 506 stream.setUpdateTime(DateUtil.getNow());
507 507 stream.setCreateTime(DateUtil.getNow());
508 508 stream.setServerId(userSetting.getServerId());
  509 + stream.setMediaServerId(mediaConfig.getId());
  510 + stream.setSelf(true);
  511 + stream.setPushIng(true);
509 512  
510 513 // 放在事务内执行
511 514 boolean result = false;
... ...