Thirdparty
/
wvp-GB28181-pro
mirror of https://github.com/648540858/wvp-GB28181-pro
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

临时提交

结构优化
648540858 2023-12-18 23:44:37 +08:00
parent dc841a9dcf
commit 74ce2be55b
28 changed files with 210 additions and 211 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
sql/common.sql
View File

@ -128,6 +128,20 @@ alter table wvp_stream_push
alter table wvp_stream_proxy
add gb_id varchar(50) default NULL;
alter table wvp_stream_proxy
add longitude double precision;
alter table wvp_stream_proxy
add latitude double precision;
alter table wvp_stream_proxy
add status bool default false;

10
src/main/java/com/genersoft/iot/vmp/gb28181/GB28181ResourceServiceImpl.java
View File

@ -123,4 +123,14 @@ public class GB28181ResourceServiceImpl implements IResourceService {
public void streamOffline(String app, String streamId) {
// TODO
}
@Override
public void startPlayback(CommonGbChannel channel, Long startTime, Long stopTime, IResourcePlayCallback callback) {
}
@Override
public void startDownload(CommonGbChannel channel, Long startTime, Long stopTime, Integer downloadSpeed, IResourcePlayCallback playCallback) {
}
}

9
src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java
View File

@ -81,15 +81,6 @@ public class EventPublisher {
applicationEventPublisher.publishEvent(outEvent);
}
public void catalogEventPublishForStream(Integer platformId, List<GbStream> gbStreams, String type) {
CatalogEvent outEvent = new CatalogEvent(this);
outEvent.setGbStreams(gbStreams);
outEvent.setType(type);
outEvent.setPlatformId(platformId);
applicationEventPublisher.publishEvent(outEvent);
}
public void recordEndEventPush(RecordInfo recordInfo) {
RecordEndEvent outEvent = new RecordEndEvent(this);
outEvent.setRecordInfo(recordInfo);

6
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java
View File

@ -50,7 +50,7 @@ public interface ISIPCommanderForPlatform {
* @param size
* @return
*/
void catalogQuery(DeviceChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size) throws SipException, InvalidArgumentException, ParseException;
void catalogQuery(CommonGbChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size) throws SipException, InvalidArgumentException, ParseException;
void catalogQuery(List<CommonGbChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag) throws InvalidArgumentException, ParseException, SipException;
/**
@ -100,12 +100,12 @@ public interface ISIPCommanderForPlatform {
/**
* 回复recordInfo
* @param deviceChannel 通道信息
* @param channel 通道信息
* @param parentPlatform 平台信息
* @param fromTag fromTag
* @param recordInfo 录像信息
*/
void recordInfo(DeviceChannel deviceChannel, ParentPlatform parentPlatform, String fromTag, RecordInfo recordInfo) throws SipException, InvalidArgumentException, ParseException;
void recordInfo(CommonGbChannel channel, ParentPlatform parentPlatform, String fromTag, RecordInfo recordInfo) throws SipException, InvalidArgumentException, ParseException;
/**
* 录像播放推送完成时发送MediaStatus消息

8
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
View File

@ -165,12 +165,12 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
* @param parentPlatform 平台信息
*/
@Override
public void catalogQuery(DeviceChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size) throws SipException, InvalidArgumentException, ParseException {
public void catalogQuery(CommonGbChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size) throws SipException, InvalidArgumentException, ParseException {
if ( parentPlatform ==null) {
return ;
}
List<DeviceChannel> channels = new ArrayList<>();
List<CommonGbChannel> channels = new ArrayList<>();
if (channel != null) {
channels.add(channel);
}
@ -734,7 +734,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
return catalogXml.toString();
}
@Override
public void recordInfo(DeviceChannel deviceChannel, ParentPlatform parentPlatform, String fromTag, RecordInfo recordInfo) throws SipException, InvalidArgumentException, ParseException {
public void recordInfo(CommonGbChannel commonGbChannel, ParentPlatform parentPlatform, String fromTag, RecordInfo recordInfo) throws SipException, InvalidArgumentException, ParseException {
if ( parentPlatform ==null) {
return ;
}
@ -753,7 +753,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
if (recordInfo.getRecordList().size() > 0) {
for (RecordItem recordItem : recordInfo.getRecordList()) {
recordXml.append("<Item>\r\n");
if (deviceChannel != null) {
if (commonGbChannel != null) {
recordXml.append("<DeviceID>" + recordItem.getDeviceId() + "</DeviceID>\r\n")
.append("<Name>" + recordItem.getName() + "</Name>\r\n")
.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "</StartTime>\r\n")

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.CommonGbChannel;
import com.genersoft.iot.vmp.conf.CivilCodeFileConf;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
@ -469,8 +470,9 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
logger.warn("[ NotifyCatalog ] event not found ： {}", event );
}
CommonGbChannel commonGbChannel = CommonGbChannel.getInstance(null, channel);
// 转发变化信息
eventPublisher.catalogEventPublish(null, channel, event);
eventPublisher.catalogEventPublish(null, commonGbChannel, event);
}
}

20
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java
View File

@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.control.cmd;
import com.genersoft.iot.vmp.common.CommonGbChannel;
import com.genersoft.iot.vmp.common.enums.DeviceControlType;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DragZoomRequest;
@ -11,6 +12,8 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.control.ControlMessageHandler;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IPlatformChannelService;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.Element;
@ -47,7 +50,10 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent
private SIPCommander cmder;
@Autowired
private SIPCommanderFroPlatform cmderFroPlatform;
private IPlatformChannelService platformChannelService;
@Autowired
private IDeviceChannelService deviceChannelService;
@Qualifier("taskExecutor")
@Autowired
@ -109,8 +115,18 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent
DeviceControlType deviceControlType = DeviceControlType.typeOf(rootElement);
logger.info("[接受deviceControl命令] 命令: {}", deviceControlType);
if (!ObjectUtils.isEmpty(deviceControlType) && !parentPlatform.getServerGBId().equals(targetGBId)) {
CommonGbChannel commonGbChannel = platformChannelService.queryChannelByPlatformIdAndChannelDeviceId(parentPlatform.getId(), channelId);
if (commonGbChannel == null) {
try {
responseAck(request, Response.NOT_FOUND);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 错误信息: {}", e.getMessage());
}
return;
}
//判断是否存在该通道
Device deviceForPlatform = storager.queryVideoDeviceByPlatformIdAndChannelId(parentPlatform.getServerGBId(), channelId);
Device deviceForPlatform = deviceChannelService.getDeviceByChannelCommonGbId(commonGbChannel.getCommonGbId());
if (deviceForPlatform == null) {
try {
responseAck(request, Response.NOT_FOUND);

24
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/DeviceInfoQueryMessageHandler.java
View File

@ -1,11 +1,14 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.query.cmd;
import com.genersoft.iot.vmp.common.CommonGbChannel;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.query.QueryMessageHandler;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IPlatformChannelService;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.Element;
@ -35,8 +38,12 @@ public class DeviceInfoQueryMessageHandler extends SIPRequestProcessorParent imp
@Autowired
private SIPCommanderFroPlatform cmderFroPlatform;
@Autowired
private IVideoManagerStorage storager;
private IPlatformChannelService platformChannelService;
@Autowired
private IDeviceChannelService deviceChannelService;
@Override
public void afterPropertiesSet() throws Exception {
@ -52,9 +59,10 @@ public class DeviceInfoQueryMessageHandler extends SIPRequestProcessorParent imp
public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element rootElement) {
logger.info("[DeviceInfo查询]消息");
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
SIPRequest request = (SIPRequest) evt.getRequest();
try {
// 回复200 OK
responseAck((SIPRequest) evt.getRequest(), Response.OK);
responseAck(request, Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] DeviceInfo查询回复: {}", e.getMessage());
return;
@ -70,7 +78,17 @@ public class DeviceInfoQueryMessageHandler extends SIPRequestProcessorParent imp
Device device = null;
// 如果id指向平台的国标编号，那么就是查询平台的信息
if (!parentPlatform.getDeviceGBId().equals(channelId)) {
device = storager.queryDeviceInfoByPlatformIdAndChannelId(parentPlatform.getServerGBId(), channelId);
CommonGbChannel commonGbChannel = platformChannelService.queryChannelByPlatformIdAndChannelDeviceId(parentPlatform.getId(), channelId);
if (commonGbChannel == null) {
try {
responseAck(request, Response.NOT_FOUND);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 错误信息: {}", e.getMessage());
}
return;
}
device = deviceChannelService.getDeviceByChannelCommonGbId(commonGbChannel.getCommonGbId());
if (device ==null){
logger.error("[平台没有该通道的使用权限]:platformId"+parentPlatform.getServerGBId()+" deviceID:"+channelId);
return;

10
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/DeviceStatusQueryMessageHandler.java
View File

@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.query.cmd;
import com.genersoft.iot.vmp.common.CommonGbChannel;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
@ -9,6 +10,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.query.QueryMessageHandler;
import com.genersoft.iot.vmp.service.IPlatformChannelService;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.Element;
@ -43,7 +45,7 @@ public class DeviceStatusQueryMessageHandler extends SIPRequestProcessorParent i
private SIPCommanderFroPlatform cmderFroPlatform;
@Autowired
private SipConfig config;
private IPlatformChannelService platformChannelService;
@Autowired
private EventPublisher publisher;
@ -71,13 +73,13 @@ public class DeviceStatusQueryMessageHandler extends SIPRequestProcessorParent i
}
String sn = rootElement.element("SN").getText();
String channelId = getText(rootElement, "DeviceID");
DeviceChannel deviceChannel = storager.queryChannelInParentPlatform(parentPlatform.getServerGBId(), channelId);
if (deviceChannel ==null){
CommonGbChannel commonGbChannel = platformChannelService.queryChannelByPlatformIdAndChannelDeviceId(parentPlatform.getId(), channelId);
if (commonGbChannel ==null){
logger.error("[平台没有该通道的使用权限]:platformId"+parentPlatform.getServerGBId()+" deviceID:"+channelId);
return;
}
try {
cmderFroPlatform.deviceStatusResponse(parentPlatform,channelId, sn, fromHeader.getTag(),deviceChannel.isStatus());
cmderFroPlatform.deviceStatusResponse(parentPlatform,channelId, sn, fromHeader.getTag(),commonGbChannel.getCommonGbStatus());
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 DeviceStatus查询回复: {}", e.getMessage());
}

28
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/RecordInfoQueryMessageHandler.java
View File

@ -1,14 +1,20 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.query.cmd;
import com.genersoft.iot.vmp.common.CommonGbChannel;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.record.RecordEndEventListener;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.query.QueryMessageHandler;
import com.genersoft.iot.vmp.service.ICommonGbChannelService;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IPlatformChannelService;
import com.genersoft.iot.vmp.service.bean.CommonGbChannelType;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.dto.ChannelSourceInfo;
@ -40,7 +46,7 @@ public class RecordInfoQueryMessageHandler extends SIPRequestProcessorParent imp
private IVideoManagerStorage storager;
@Autowired
private SIPCommanderFroPlatform cmderFroPlatform;
private ISIPCommanderForPlatform cmderFroPlatform;
@Autowired
private SIPCommander commander;
@ -49,10 +55,10 @@ public class RecordInfoQueryMessageHandler extends SIPRequestProcessorParent imp
private RecordEndEventListener recordEndEventListener;
@Autowired
private SipConfig config;
private IDeviceChannelService deviceChannelService;
@Autowired
private EventPublisher publisher;
private IPlatformChannelService platformChannelService;
@Override
public void afterPropertiesSet() throws Exception {
@ -93,16 +99,15 @@ public class RecordInfoQueryMessageHandler extends SIPRequestProcessorParent imp
type = typeElement.getText();
}
// 确认是直播还是国标， 国标直接请求下级，直播请求录像管理服务
List<ChannelSourceInfo> channelSources = storager.getChannelSource(parentPlatform.getServerGBId(), channelId);
if (channelSources.get(0).getCount() > 0) { // 国标
CommonGbChannel commonGbChannel = platformChannelService.queryChannelByPlatformIdAndChannelDeviceId(parentPlatform.getId(), channelId);
if (commonGbChannel.getType().equals(CommonGbChannelType.GB28181)) { // 国标
// 向国标设备请求录像数据
Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(parentPlatform.getServerGBId(), channelId);
DeviceChannel deviceChannel = storager.queryChannelInParentPlatform(parentPlatform.getServerGBId(), channelId);
Device device = deviceChannelService.getDeviceByChannelCommonGbId(commonGbChannel.getCommonGbId());
// 接收录像数据
recordEndEventListener.addEndEventHandler(deviceChannel.getDeviceId(), channelId, (recordInfo)->{
recordEndEventListener.addEndEventHandler(device.getDeviceId(), channelId, (recordInfo)->{
try {
cmderFroPlatform.recordInfo(deviceChannel, parentPlatform, request.getFromTag(), recordInfo);
cmderFroPlatform.recordInfo(commonGbChannel, parentPlatform, request.getFromTag(), recordInfo);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 回复录像数据: {}", e.getMessage());
}
@ -128,7 +133,8 @@ public class RecordInfoQueryMessageHandler extends SIPRequestProcessorParent imp
logger.error("[命令发送失败] 录像查询: {}", e.getMessage());
}
}else if (channelSources.get(1).getCount() > 0) { // 直播流
}else if (commonGbChannel.getType().equals(CommonGbChannelType.PUSH)
|| commonGbChannel.getType().equals(CommonGbChannelType.PROXY)) { // 直播流
// TODO
try {
responseAck(request, Response.NOT_IMPLEMENTED); // 回复未实现

7
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/InviteResponseProcessor.java
View File

@ -14,6 +14,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.sdp.SdpException;
import javax.sdp.SdpParseException;
import javax.sdp.SessionDescription;
import javax.sip.InvalidArgumentException;
@ -90,7 +91,9 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
}
} catch (InvalidArgumentException | ParseException | SipException | SdpParseException e) {
logger.info("[点播回复ACK]，异常：", e );
}
}
} catch (SdpException e) {
logger.info("[点播收到200OK]，SDP解析异常：", e );
}
}
}

4
src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
View File

@ -86,4 +86,8 @@ public interface IDeviceChannelService {
*/
boolean updateChannelsForCatalog(Device device, List<DeviceChannel> deviceChannelList);
/**
* 根据通道ID获取设备
*/
Device getDeviceByChannelCommonGbId(int commonGbId);
}

2
src/main/java/com/genersoft/iot/vmp/service/IPlatformChannelService.java
View File

@ -35,4 +35,6 @@ public interface IPlatformChannelService {
* 查询通道
*/
CommonGbChannel queryChannelByPlatformIdAndChannelDeviceId(Integer platformId, String channelId);
List<CommonGbChannel> queryCommonGbChannellList(Integer id);
}

7
src/main/java/com/genersoft/iot/vmp/service/IStreamProxyService.java
View File

@ -5,9 +5,12 @@ import com.genersoft.iot.vmp.common.GeneralCallback;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxy;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.github.pagehelper.PageInfo;
import java.util.List;
public interface IStreamProxyService {
/**
@ -111,4 +114,8 @@ public interface IStreamProxyService {
*/
ResourceBaseInfo getOverview();
/**
* 更新redis发来的gps更新消息
*/
void updateStreamGPS(List<GPSMsgInfo> gpsMsgInfoList);
}

6
src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
View File

@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPush;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto;
@ -103,4 +104,9 @@ public interface IStreamPushService {
void batchUpdate(List<StreamPush> streamPushItemForUpdate);
void update(StreamPush transform);
/**
* 更新redis发来的gps更新消息
*/
void updateStreamGPS(List<GPSMsgInfo> gpsMsgInfoList);
}

5
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
View File

@ -711,4 +711,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
}
}
@Override
public Device getDeviceByChannelCommonGbId(int commonGbId) {
return channelMapper.getDeviceByChannelCommonGbId(commonGbId);
}
}

7
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformChannelServiceImpl.java
View File

@ -122,7 +122,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
@Override
public List<ParentPlatform> querySharePlatformListByChannelId(int commonGbId, List<String> platforms) {
return platformChannelMapper.querySharePlatformListByChannelId();
return platformChannelMapper.querySharePlatformListByChannelId(commonGbId, platforms);
}
@Override
@ -153,4 +153,9 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
public CommonGbChannel queryChannelByPlatformIdAndChannelDeviceId(Integer platformId, String channelId) {
return null;
}
@Override
public List<CommonGbChannel> queryCommonGbChannellList(Integer platformId) {
return platformChannelMapper.queryCommonGbChannellList(platformId);
}
}

15
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java
View File

@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.service.impl;
import com.genersoft.iot.vmp.common.CommonGbChannel;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
@ -12,6 +13,7 @@ import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlatformChannelService;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
@ -77,6 +79,9 @@ public class PlatformServiceImpl implements IPlatformService {
@Autowired
private UserSetting userSetting;
@Autowired
private IPlatformChannelService platformChannelService;
@Override
@ -393,14 +398,12 @@ public class PlatformServiceImpl implements IPlatformService {
SubscribeInfo subscribe = subscribeHolder.getMobilePositionSubscribe(platform.getId());
if (subscribe != null) {
// TODO 暂时只处理视频流的回复,后续增加对国标设备的支持
List<DeviceChannel> gbStreams = gbStreamMapper.queryGbStreamListInPlatform(platform.getServerGBId(), userSetting.isUsePushingAsStatus());
if (gbStreams.size() == 0) {
List<CommonGbChannel> channelList = platformChannelService.queryCommonGbChannellList(platform.getId());
if (channelList.isEmpty()) {
return;
}
for (DeviceChannel deviceChannel : gbStreams) {
String gbId = deviceChannel.getChannelId();
GPSMsgInfo gpsMsgInfo = redisCatchStorage.getGpsMsgInfo(gbId);
for (CommonGbChannel channel : channelList) {
GPSMsgInfo gpsMsgInfo = redisCatchStorage.getGpsMsgInfo(channel.getCommonGbDeviceID());
// 无最新位置不发送
if (gpsMsgInfo != null) {
// 经纬度都为0不发送

6
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
View File

@ -22,6 +22,7 @@ import com.genersoft.iot.vmp.service.ICommonGbChannelService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
@ -520,4 +521,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
return new ResourceBaseInfo(total, online);
}
@Override
public void updateStreamGPS(List<GPSMsgInfo> gpsMsgInfoList) {
streamProxyMapper.updateStreamGPS(gpsMsgInfoList);
}
}

6
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
View File

@ -15,6 +15,7 @@ import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.service.bean.CommonGbChannelType;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.Group;
import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
@ -472,4 +473,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
return new ResourceBaseInfo(total, online);
}
@Override
public void updateStreamGPS(List<GPSMsgInfo> gpsMsgInfoList) {
streamPushMapper.updateStreamGPS(gpsMsgInfoList);
}
}

18
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGpsMsgListener.java
View File

@ -1,6 +1,8 @@
package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@ -33,6 +35,12 @@ public class RedisGpsMsgListener implements MessageListener {
@Autowired
private IVideoManagerStorage storager;
@Autowired
private IStreamProxyService streamProxyService;
@Autowired
private IStreamPushService streamPushService;
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@ -66,10 +74,12 @@ public class RedisGpsMsgListener implements MessageListener {
*/
@Scheduled(fixedRate = 2 * 1000) //每2秒执行一次
public void execute(){
List<GPSMsgInfo> gpsMsgInfo = redisCatchStorage.getAllGpsMsgInfo();
if (gpsMsgInfo.size() > 0) {
storager.updateStreamGPS(gpsMsgInfo);
for (GPSMsgInfo msgInfo : gpsMsgInfo) {
List<GPSMsgInfo> gpsMsgInfoList = redisCatchStorage.getAllGpsMsgInfo();
if (gpsMsgInfoList.size() > 0) {
streamProxyService.updateStreamGPS(gpsMsgInfoList);
streamPushService.updateStreamGPS(gpsMsgInfoList);
for (GPSMsgInfo msgInfo : gpsMsgInfoList) {
msgInfo.setStored(true);
redisCatchStorage.updateGpsMsgInfo(msgInfo);
}

48
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorage.java
View File

@ -123,18 +123,7 @@ public interface IVideoManagerStorage {
*/
ParentPlatform queryParentPlatByServerGBId(String platformGbId);
DeviceChannel queryChannelInParentPlatform(String platformId, String channelId);
Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId);
/**
* 针对deviceinfo指令的查询接口
* @param platformId 平台id
* @param channelId 通道id
* @return 设备信息
*/
Device queryDeviceInfoByPlatformIdAndChannelId(String platformId, String channelId);
/**
* 添加Mobile Position设备移动位置
* @param mobilePosition
@ -181,33 +170,6 @@ public interface IVideoManagerStorage {
*/
PageInfo<StreamProxy> queryStreamProxyList(Integer page, Integer count);
/**
* 根据国标ID获取平台关联的直播流
* @param platformId
* @param channelId
* @return
*/
GbStream queryStreamInParentPlatform(String platformId, String channelId);
/**
* 获取平台关联的直播流
* @param platformId
* @return
*/
List<DeviceChannel> queryGbStreamListInPlatform(String platformId);
/**
* 移除单个推流
* @param app
* @param stream
*/
int removeMedia(String app, String stream);
/**
* 设置流离线
*/
int mediaOffline(String app, String streamId);
/**
* 根据媒体ID获取启用/不启用的代理列表
* @param id 媒体ID
@ -236,16 +198,6 @@ public interface IVideoManagerStorage {
*/
StreamProxy getStreamProxyByAppAndStream(String app, String streamId);
int updateStreamGPS(List<GPSMsgInfo> gpsMsgInfo);
List<ParentPlatform> queryPlatFormListForGBWithGBId(String channelId, List<String> platforms);
List<ParentPlatform> queryPlatFormListForStreamWithGBId(String app, String stream, List<String> platforms);
GbStream getGbStream(String app, String streamId);
List<ChannelSourceInfo> getChannelSource(String platformId, String gbId);
void updateChannelPosition(DeviceChannel deviceChannel);

13
src/main/java/com/genersoft/iot/vmp/storager/dao/CommonChannelPlatformMapper.java
View File

@ -2,12 +2,14 @@ package com.genersoft.iot.vmp.storager.dao;
import com.genersoft.iot.vmp.common.CommonGbChannel;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.service.bean.Group;
import com.genersoft.iot.vmp.service.bean.Region;
import com.genersoft.iot.vmp.vmanager.bean.UpdateCommonChannelToGroup;
import com.genersoft.iot.vmp.vmanager.bean.UpdateCommonChannelToRegion;
import org.apache.ibatis.annotations.*;
import org.springframework.stereotype.Repository;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.Map;
@ -42,4 +44,15 @@ public interface CommonChannelPlatformMapper {
"</script>")
int removeChannels(@Param("platformId") Integer platformId,
@Param("commonGbChannelIds") List<Integer> commonGbChannelIds);
@Select("<script>" +
"select p.* from wvp_platform p left join wvp_common_channel_platform cp on p.id = cp.platform_id where cp.common_gb_channel_id = #{commonGbId} and p.server_gb_id in " +
"<foreach collection='platforms' item='item' open='(' separator=',' close=')' >#{item}</foreach>" +
" </script>")
List<ParentPlatform> querySharePlatformListByChannelId(@Param("commonGbId") int commonGbId, @Param("platforms") List<String> platforms);
@Select("<script>" +
"select cc.* from wvp_common_channel cc left join wvp_common_channel_platform cp on cc.common_gb_id = cp.common_gb_channel_id where cp.platform_id = #{platformId}" +
" </script>")
List<CommonGbChannel> queryCommonGbChannellList(@Param("platformId") Integer platformId);
}

3
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
View File

@ -470,4 +470,7 @@ public interface DeviceChannelMapper {
"where device_id=#{deviceId}" +
" </script>"})
List<Integer> getCommonChannelIdList(@Param("deviceId") String deviceId);
@Select("select de.* from wvp_device de left join wvp_device_channel dc on de.device_id = dc.deviceId where dc.common_gb_channel_id=#{commonGbId}")
Device getDeviceByChannelCommonGbId(@Param("commonGbId") int commonGbId);
}

13
src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.storager.dao;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxy;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import org.apache.ibatis.annotations.*;
import org.springframework.stereotype.Repository;
@ -87,4 +88,16 @@ public interface StreamProxyMapper {
@Select("select count(1) from wvp_stream_proxy where status = true")
int getOnline();
@Update({"<script>" +
"<foreach collection='gpsMsgInfoList' item='item' separator=';'>" +
" UPDATE" +
" wvp_stream_proxy" +
" SET longitude = #{item.lng}, latitude= #{item.lat}" +
" WHERE gb_id=#{item.id}" +
"</foreach>" +
"</script>"})
void updateStreamGPS(@Param("gpsMsgInfoList") List<GPSMsgInfo> gpsMsgInfoList);
}

11
src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
View File

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.storager.dao;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPush;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
import org.apache.ibatis.annotations.*;
import org.springframework.stereotype.Repository;
@ -195,4 +196,14 @@ public interface StreamPushMapper {
@Select("select* from wvp_stream_push where id = #{id}")
StreamPush query(@Param("id") Integer id);
@Update({"<script>" +
"<foreach collection='gpsMsgInfoList' item='item' separator=';'>" +
" UPDATE" +
" wvp_stream_push" +
" SET longitude = #{item.lng}, latitude= #{item.lat}" +
" WHERE gb_id=#{item.id}" +
"</foreach>" +
"</script>"})
void updateStreamGPS(List<GPSMsgInfo> gpsMsgInfoList);
}

109
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
View File

@ -229,49 +229,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
return deviceMapper.queryDeviceWithAsMessageChannel();
}
@Override
public DeviceChannel queryChannelInParentPlatform(String platformId, String channelId) {
List<DeviceChannel> channels = platformChannelMapper.queryChannelInParentPlatform(platformId, channelId);
if (channels.size() > 1) {
// 出现长度大于0的时候肯定是国标通道的ID重复了
logger.warn("国标ID存在重复：{}", channelId);
}
if (channels.size() == 0) {
return null;
}else {
return channels.get(0);
}
}
@Override
public Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId) {
List<Device> devices = platformChannelMapper.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
if (devices.size() > 1) {
// 出现长度大于0的时候肯定是国标通道的ID重复了
logger.warn("国标ID存在重复：{}", channelId);
}
if (devices.size() == 0) {
return null;
}else {
return devices.get(0);
}
}
@Override
public Device queryDeviceInfoByPlatformIdAndChannelId(String platformId, String channelId) {
List<Device> devices = platformChannelMapper.queryDeviceInfoByPlatformIdAndChannelId(platformId, channelId);
if (devices.size() > 1) {
// 出现长度大于0的时候肯定是国标通道的ID重复了
logger.warn("国标ID存在重复：{}", channelId);
}
if (devices.size() == 0) {
return null;
}else {
return devices.get(0);
}
}
/**
* 查询最新移动位置
@ -306,27 +263,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
return new PageInfo<>(all);
}
/**
* 根据国标ID获取平台关联的直播流
* @param platformId
* @param gbId
* @return
*/
@Override
public GbStream queryStreamInParentPlatform(String platformId, String gbId) {
return gbStreamMapper.queryStreamInPlatform(platformId, gbId);
}
/**
* 获取平台关联的直播流
* @param platformId
* @return
*/
@Override
public List<DeviceChannel> queryGbStreamListInPlatform(String platformId) {
return gbStreamMapper.queryGbStreamListInPlatform(platformId, userSetting.isUsePushingAsStatus());
}
/**
* 按照是app和stream获取代理流
* @param app
@ -338,23 +274,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
return streamProxyMapper.selectOne(app, stream);
}
@Override
public int removeMedia(String app, String stream) {
return streamPushMapper.del(app, stream);
}
@Override
public int mediaOffline(String app, String stream) {
GbStream gbStream = gbStreamMapper.selectOne(app, stream);
int result;
if ("proxy".equals(gbStream.getStreamType())) {
result = streamProxyMapper.updateStatus(app, stream, false);
}else {
result = streamPushMapper.updatePushStatus(app, stream, false);
}
return result;
}
@Override
public List<StreamProxy> getStreamProxyListForEnableInMediaServer(String id, boolean enable) {
return streamProxyMapper.selectForEnableInMediaServer(id, enable);
@ -365,11 +284,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
return streamProxyMapper.selectOne(app, streamId);
}
@Override
public int updateStreamGPS(List<GPSMsgInfo> gpsMsgInfos) {
return gbStreamMapper.updateStreamGPS(gpsMsgInfos);
}
private DeviceChannel getDeviceChannelByCatalog(PlatformCatalog catalog) {
ParentPlatform platform = platformMapper.getParentPlatByServerGBId(catalog.getPlatformId());
@ -391,29 +305,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
return deviceChannel;
}
@Override
public List<ParentPlatform> queryPlatFormListForGBWithGBId(String channelId, List<String> platforms) {
return platformChannelMapper.queryPlatFormListForGBWithGBId(channelId, platforms);
}
@Override
public List<ParentPlatform> queryPlatFormListForStreamWithGBId(String app, String stream, List<String> platforms) {
if (platforms == null || platforms.size() == 0) {
return new ArrayList<>();
}
return platformGbStreamMapper.queryPlatFormListForGBWithGBId(app, stream, platforms);
}
@Override
public GbStream getGbStream(String app, String streamId) {
return gbStreamMapper.selectOne(app, streamId);
}
@Override
public List<ChannelSourceInfo> getChannelSource(String platformId, String gbId) {
return platformMapper.getChannelSource(platformId, gbId);
}
@Override
public void updateChannelPosition(DeviceChannel deviceChannel) {
if (deviceChannel.getChannelId().equals(deviceChannel.getDeviceId())) {

8
src/main/java/com/genersoft/iot/vmp/vmanager/channel/CommonChannelController.java
View File

@ -203,7 +203,7 @@ public class CommonChannelController {
result.setResult(wvpResult);
commonGbChannelService.stopPlay(channel, null);
});
commonGbChannelService.startPlay(channel, (callbackChannel, code, message, streamInfo) -> {
commonGbChannelService.startPlay(channel, (callbackChannel, mediaServerItem, code, message, streamInfo) -> {
if (code == ErrorCode.SUCCESS.getCode()) {
WVPResult<StreamContent> wvpResult = new WVPResult<>();
wvpResult.setCode(ErrorCode.SUCCESS.getCode());
@ -252,15 +252,15 @@ public class CommonChannelController {
result.onTimeout(()->{
logger.info("[停止播放通道] 超时 channelDeviceId：{} ", channelDeviceId);
// 释放rtpserver
WVPResult wvpResult = new WVPResult<>();
WVPResult<StreamContent> wvpResult = new WVPResult<>();
wvpResult.setCode(ErrorCode.ERROR100.getCode());
wvpResult.setMsg("停止播放通道超时");
result.setResult(wvpResult);
commonGbChannelService.stopPlay(channel, null);
});
commonGbChannelService.stopPlay(channel, (commonGbChannel, code, message, streamInfo) -> {
WVPResult wvpResult = new WVPResult();
commonGbChannelService.stopPlay(channel, (commonGbChannel, mediaServerItem, code, message, streamInfo) -> {
WVPResult<StreamContent> wvpResult = new WVPResult<>();
wvpResult.setCode(code);
wvpResult.setMsg(message);
result.setResult(wvpResult);