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

优化日志以及属性设置代码

pull/1642/head
648540858 2024-07-03 17:09:49 +08:00
parent 799932f614
commit adf040ec4b
34 changed files with 184 additions and 1042 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/com/genersoft/iot/vmp/VManageBootstrap.java
View File

@ -2,8 +2,7 @@ package com.genersoft.iot.vmp;
import com.genersoft.iot.vmp.utils.GitUtil;
import com.genersoft.iot.vmp.utils.SpringBeanFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
@ -26,19 +25,18 @@ import java.util.Collections;
@SpringBootApplication
@EnableScheduling
@EnableCaching
@Slf4j
public class VManageBootstrap extends SpringBootServletInitializer {
private final static Logger loggerger = LoggerFactory.getLogger(VManageBootstrap.class);
private static String[] args;
private static ConfigurableApplicationContext context;
public static void main(String[] args) {
VManageBootstrap.args = args;
VManageBootstrap.context = SpringApplication.run(VManageBootstrap.class, args);
GitUtil gitUtil1 = SpringBeanFactory.getBean("gitUtil");
logger.info("构建版本： {}", gitUtil1.getBuildVersion());
logger.info("构建时间： {}", gitUtil1.getBuildDate());
logger.info("GIT最后提交时间： {}", gitUtil1.getCommitTime());
log.info("构建版本： {}", gitUtil1.getBuildVersion());
log.info("构建时间： {}", gitUtil1.getBuildDate());
log.info("GIT最后提交时间： {}", gitUtil1.getCommitTime());
}
// 项目重启
public static void restart() {

20
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
View File

@ -1,9 +1,11 @@
package com.genersoft.iot.vmp.gb28181.event.subscribe.catalog;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@ -32,9 +34,6 @@ public class CatalogEventLister implements ApplicationListener<CatalogEvent> {
@Autowired
private ISIPCommanderForPlatform sipCommanderFroPlatform;
@Autowired
private IGbStreamService gbStreamService;
@Autowired
private SubscribeHolder subscribeHolder;
@ -145,16 +144,11 @@ public class CatalogEventLister implements ApplicationListener<CatalogEvent> {
continue;
}
log.info("[Catalog事件: {}]平台：{}，影响通道{}", event.getType(), platform.getServerGBId(), gbId);
List<CommonGBChannel> deviceChannelList = new ArrayList<>();
List<CommonGBChannel> channelList = new ArrayList<>();
CommonGBChannel deviceChannel = channelMap.get(gbId);
deviceChannelList.add(deviceChannel);
GbStream gbStream = storager.queryStreamInParentPlatform(platform.getServerGBId(), gbId);
if(gbStream != null){
CommonGBChannel deviceChannelByStream = gbStreamService.getDeviceChannelListByStreamWithStatus(gbStream, gbStream.getCatalogId(), platform);
deviceChannelList.add(deviceChannelByStream);
}
channelList.add(deviceChannel);
try {
sipCommanderFroPlatform.sendNotifyForCatalogAddOrUpdate(event.getType(), platform, deviceChannelList, subscribeInfo, null);
sipCommanderFroPlatform.sendNotifyForCatalogAddOrUpdate(event.getType(), platform, channelList, subscribeInfo, null);
} catch (InvalidArgumentException | ParseException | NoSuchFieldException |
SipException | IllegalAccessException e) {
log.error("[命令发送失败] 国标级联 Catalog通知: {}", e.getMessage());

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

@ -25,21 +25,21 @@ import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
import com.genersoft.iot.vmp.service.redisMsg.RedisPushStreamResponseListener;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
import com.genersoft.iot.vmp.utils.DateUtil;
import gov.nist.javax.sdp.TimeDescriptionImpl;
import gov.nist.javax.sdp.fields.TimeField;
@ -49,8 +49,6 @@ import gov.nist.javax.sip.message.SIPResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
@ -64,10 +62,10 @@ import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Vector;
import java.util.*;
/**
* SIP命令类型： INVITE请求
@ -491,7 +489,9 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
String startTimeStr = DateUtil.urlFormatter.format(start);
String endTimeStr = DateUtil.urlFormatter.format(end);
String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0,false,!channel.getHasAudio(), false, device.getStreamMode());
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null,
device.isSsrcCheck(), true, 0,false,!channel.getHasAudio(), false, tcpMode);
sendRtpItem.setStream(stream);
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
@ -521,7 +521,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
sendRtpItem.setPlayType(InviteStreamType.DOWNLOAD);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,!channel.getHasAudio(), false, device.getStreamMode());
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,!channel.getHasAudio(), false, tcpMode);
sendRtpItem.setStream(ssrcInfo.getStream());
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java
View File

@ -154,8 +154,6 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
deviceChannel.setLatitude(mobilePosition.getLatitude());
deviceChannel.setGpsTime(mobilePosition.getTime());
deviceChannel = deviceChannelService.updateGps(deviceChannel, sipMsgInfo.getDevice());
deviceChannelService.updateChannelGPS(device, deviceChannel, mobilePosition);
}
}

1
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MobilePositionNotifyMessageHandler.java
View File

@ -133,7 +133,6 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
deviceChannel.setLatitude(mobilePosition.getLatitude());
deviceChannel.setGpsTime(mobilePosition.getTime());
deviceChannel = deviceChannelService.updateGps(deviceChannel, sipMsgInfo.getDevice());
deviceChannelService.updateChannelGPS(device, deviceChannel, mobilePosition);
} catch (DocumentException e) {

5
src/main/java/com/genersoft/iot/vmp/jt1078/codec/decode/Jt808Decoder.java
View File

@ -11,8 +11,7 @@ import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.List;
@ -22,8 +21,8 @@ import java.util.List;
* @date 2023/4/27 18:10
* @email qingtaij@163.com
*/
@Slf4j
public class Jt808Decoder extends ByteToMessageDecoder {
private final static Logger logger = LoggerFactory.getLogger(Jt808Decoder.class);
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

5
src/main/java/com/genersoft/iot/vmp/jt1078/codec/encode/Jt808Encoder.java
View File

@ -7,16 +7,15 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
/**
* @author QingtaiJiang
* @date 2023/4/27 18:10
* @email qingtaij@163.com
*/
@Slf4j
public class Jt808Encoder extends MessageToByteEncoder<Rs> {
private final static Logger logger = LoggerFactory.getLogger(Jt808Encoder.class);
@Override
protected void encode(ChannelHandlerContext ctx, Rs msg, ByteBuf out) throws Exception {

5
src/main/java/com/genersoft/iot/vmp/jt1078/codec/encode/Jt808EncoderCmd.java
View File

@ -13,8 +13,7 @@ import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.util.ByteProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
import java.util.LinkedList;
@ -24,8 +23,8 @@ import java.util.LinkedList;
* @date 2023/4/27 18:25
* @email qingtaij@163.com
*/
@Slf4j
public class Jt808EncoderCmd extends MessageToByteEncoder<Cmd> {
private final static Logger logger = LoggerFactory.getLogger(Jt808EncoderCmd.class);
@Override
protected void encode(ChannelHandlerContext ctx, Cmd cmd, ByteBuf out) throws Exception {

6
src/main/java/com/genersoft/iot/vmp/jt1078/codec/netty/Jt808Handler.java
View File

@ -8,18 +8,16 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
/**
* @author QingtaiJiang
* @date 2023/4/27 18:14
* @email qingtaij@163.com
*/
@Slf4j
public class Jt808Handler extends ChannelInboundHandlerAdapter {
private final static Logger logger = LoggerFactory.getLogger(Jt808Handler.class);
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof Rs) {

5
src/main/java/com/genersoft/iot/vmp/jt1078/codec/netty/TcpServer.java
View File

@ -17,8 +17,7 @@ import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.TimeUnit;
@ -28,8 +27,8 @@ import java.util.concurrent.TimeUnit;
* @email qingtaij@163.com
*/
@Slf4j
public class TcpServer {
private final static Logger logger = LoggerFactory.getLogger(TcpServer.class);
private final Integer port;
private boolean isRunning = false;

6
src/main/java/com/genersoft/iot/vmp/jt1078/proc/factory/CodecFactory.java
View File

@ -3,8 +3,7 @@ package com.genersoft.iot.vmp.jt1078.proc.factory;
import com.genersoft.iot.vmp.jt1078.annotation.MsgId;
import com.genersoft.iot.vmp.jt1078.proc.request.Re;
import com.genersoft.iot.vmp.jt1078.util.ClassUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import java.util.HashMap;
import java.util.List;
@ -15,9 +14,8 @@ import java.util.Map;
* @date 2023/4/27 18:29
* @email qingtaij@163.com
*/
@Slf4j
public class CodecFactory {
private final static Logger logger = LoggerFactory.getLogger(CodecFactory.class);
private static Map<String, Class<?>> protocolHash;

5
src/main/java/com/genersoft/iot/vmp/jt1078/proc/request/Re.java
View File

@ -4,8 +4,7 @@ import com.genersoft.iot.vmp.jt1078.proc.Header;
import com.genersoft.iot.vmp.jt1078.proc.response.Rs;
import com.genersoft.iot.vmp.jt1078.session.Session;
import io.netty.buffer.ByteBuf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
/**
@ -13,8 +12,8 @@ import org.springframework.util.StringUtils;
* @date 2023/4/27 18:50
* @email qingtaij@163.com
*/
@Slf4j
public abstract class Re {
private final static Logger logger = LoggerFactory.getLogger(Re.class);
protected abstract Rs decode0(ByteBuf buf, Header header, Session session);

5
src/main/java/com/genersoft/iot/vmp/jt1078/session/Session.java
View File

@ -3,8 +3,7 @@ package com.genersoft.iot.vmp.jt1078.session;
import com.genersoft.iot.vmp.jt1078.proc.Header;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.atomic.AtomicInteger;
@ -13,8 +12,8 @@ import java.util.concurrent.atomic.AtomicInteger;
* @date 2023/4/27 18:54
* @email qingtaij@163.com
*/
@Slf4j
public class Session {
private final static Logger logger = LoggerFactory.getLogger(Session.class);
public static final AttributeKey<Session> KEY = AttributeKey.newInstance(Session.class.getName());

5
src/main/java/com/genersoft/iot/vmp/jt1078/session/SessionManager.java
View File

@ -2,8 +2,7 @@ package com.genersoft.iot.vmp.jt1078.session;
import com.genersoft.iot.vmp.jt1078.proc.entity.Cmd;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@ -16,9 +15,9 @@ import java.util.concurrent.TimeUnit;
* @date 2023/4/27 19:54
* @email qingtaij@163.com
*/
@Slf4j
public enum SessionManager {
INSTANCE;
private final static Logger logger = LoggerFactory.getLogger(SessionManager.class);
// 用与消息的缓存
private final Map<String, SynchronousQueue<String>> topicSubscribers = new ConcurrentHashMap<>();

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

@ -14,11 +14,6 @@ import java.util.List;
*/
public interface IDeviceChannelService {
/**
* 更新gps信息
*/
DeviceChannel updateGps(DeviceChannel deviceChannel, Device device);
/**
* 添加设备通道
*

10
src/main/java/com/genersoft/iot/vmp/service/IDeviceService.java
View File

@ -5,7 +5,6 @@ import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.SipTransactionInfo;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
import com.genersoft.iot.vmp.vmanager.bean.BaseTree;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import java.util.List;
@ -116,15 +115,6 @@ public interface IDeviceService {
*/
void updateDevice(Device device);
/**
* 树形查询接口
* @param deviceId 设备ID
* @param parentId 父ID
* @param onlyCatalog 只获取目录
* @return
*/
List<BaseTree<DeviceChannel>> queryVideoDeviceTree(String deviceId, String parentId, boolean onlyCatalog);
/**
* 查询树节点下的通道
* @param deviceId 设备ID

79
src/main/java/com/genersoft/iot/vmp/service/IGbStreamService.java
View File

@ -1,79 +0,0 @@
package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
import com.github.pagehelper.PageInfo;
import java.util.List;
import java.util.Map;
/**
* 级联国标平台关联流业务接口
*/
public interface IGbStreamService {
/**
* 分页获取所有
* @param page
* @param count
* @return
*/
PageInfo<GbStream> getAll(Integer page, Integer count, String platFormId, String catalogId,String query,String mediaServerId);
/**
* 移除
* @param app
* @param stream
*/
void del(String app, String stream);
/**
* 保存国标关联
* @param gbStreams
*/
boolean addPlatformInfo(List<GbStream> gbStreams, String platformId, String catalogId);
/**
* 移除国标关联
* @param gbStreams
* @param platformId
*/
boolean delPlatformInfo(String platformId, List<GbStream> gbStreams);
DeviceChannel getDeviceChannelListByStream(GbStream gbStream, String catalogId, ParentPlatform platform);
void sendCatalogMsg(CommonGBChannel gbStream, String type);
void sendCatalogMsgs(List<CommonGBChannel> gbStreams, String type);
/**
* 修改gbId或name
* @param streamPushItemForUpdate
* @return
*/
int updateGbIdOrName(List<StreamPush> streamPushItemForUpdate);
DeviceChannel getDeviceChannelListByStreamWithStatus(GbStream gbStream, String catalogId, ParentPlatform platform);
/**
* 查询所有未分配的通道
* @param platformId
* @return
*/
List<GbStream> getAllGBChannels(String platformId);
/**
* 移除所有关联的通道
* @param platformId
* @param catalogId
*/
void delAllPlatformInfo(String platformId, String catalogId);
List<GbStream> getGbChannelWithGbid(String gbId);
Map<String, GbStream> getAllGBId();
}

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

@ -10,13 +10,13 @@ import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.MobilePosition;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.storager.dao.DeviceMapper;
import com.genersoft.iot.vmp.storager.dao.DeviceMobilePositionMapper;
import com.genersoft.iot.vmp.storager.dao.PlatformChannelMapper;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
@ -24,7 +24,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
@ -51,6 +50,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
@Autowired
private DeviceChannelMapper channelMapper;
@Autowired
private PlatformChannelMapper platformChannelMapper;
@Autowired
private DeviceMapper deviceMapper;
@ -63,43 +65,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Override
public DeviceChannel updateGps(DeviceChannel deviceChannel, Device device) {
if (deviceChannel.getLongitude()*deviceChannel.getLatitude() > 0) {
if (device == null) {
device = deviceMapper.getDeviceByDeviceId(deviceChannel.getDeviceId());
}
if ("WGS84".equals(device.getGeoCoordSys())) {
deviceChannel.setLongitudeWgs84(deviceChannel.getLongitude());
deviceChannel.setLatitudeWgs84(deviceChannel.getLatitude());
Double[] position = Coordtransform.WGS84ToGCJ02(deviceChannel.getLongitude(), deviceChannel.getLatitude());
deviceChannel.setLongitudeGcj02(position[0]);
deviceChannel.setLatitudeGcj02(position[1]);
}else if ("GCJ02".equals(device.getGeoCoordSys())) {
deviceChannel.setLongitudeGcj02(deviceChannel.getLongitude());
deviceChannel.setLatitudeGcj02(deviceChannel.getLatitude());
Double[] position = Coordtransform.GCJ02ToWGS84(deviceChannel.getLongitude(), deviceChannel.getLatitude());
deviceChannel.setLongitudeWgs84(position[0]);
deviceChannel.setLatitudeWgs84(position[1]);
}else {
deviceChannel.setLongitudeGcj02(0.00);
deviceChannel.setLatitudeGcj02(0.00);
deviceChannel.setLongitudeWgs84(0.00);
deviceChannel.setLatitudeWgs84(0.00);
}
}else {
deviceChannel.setLongitudeGcj02(deviceChannel.getLongitude());
deviceChannel.setLatitudeGcj02(deviceChannel.getLatitude());
deviceChannel.setLongitudeWgs84(deviceChannel.getLongitude());
deviceChannel.setLatitudeWgs84(deviceChannel.getLatitude());
}
return deviceChannel;
}
@Override
public void updateChannel(String deviceId, DeviceChannel channel) {
String channelId = channel.getChannelId();
String channelId = channel.getDeviceId();
channel.setDeviceId(deviceId);
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
@ -108,7 +76,6 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
String now = DateUtil.getNow();
channel.setUpdateTime(now);
DeviceChannel deviceChannel = channelMapper.queryChannel(deviceId, channelId);
channel = updateGps(channel, null);
if (deviceChannel == null) {
channel.setCreateTime(now);
channelMapper.add(channel);
@ -129,30 +96,28 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
if (channelList.size() == 0) {
for (DeviceChannel channel : channels) {
channel.setDeviceId(deviceId);
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channel.getChannelId());
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channel.getDeviceId());
if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
channel.setStreamId(inviteInfo.getStreamInfo().getStream());
}
String now = DateUtil.getNow();
channel.setUpdateTime(now);
channel.setCreateTime(now);
channel = updateGps(channel, device);
addChannels.add(channel);
}
}else {
for (DeviceChannel deviceChannel : channelList) {
channelsInStore.put(deviceChannel.getChannelId(), deviceChannel);
channelsInStore.put(deviceChannel.getDeviceId(), deviceChannel);
}
for (DeviceChannel channel : channels) {
channel.setDeviceId(deviceId);
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channel.getChannelId());
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channel.getDeviceId());
if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
channel.setStreamId(inviteInfo.getStreamInfo().getStream());
}
String now = DateUtil.getNow();
channel.setUpdateTime(now);
channel = updateGps(channel, device);
if (channelsInStore.get(channel.getChannelId()) != null) {
if (channelsInStore.get(channel.getDeviceId()) != null) {
updateChannels.add(channel);
}else {
addChannels.add(channel);
@ -216,7 +181,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
String now = DateUtil.getNow();
deviceChannels.parallelStream().forEach(deviceChannel -> {
deviceChannel.setUpdateTime(now);
result.add(updateGps(deviceChannel, device));
result.add(deviceChannel);
});
int limitCount = 50;
if (result.size() > limitCount) {
@ -252,7 +217,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
@Override
public void online(DeviceChannel channel) {
channelMapper.online(channel.getDeviceId(), channel.getChannelId());
channelMapper.online(channel.getDeviceId(), channel.getDeviceId());
}
@Override
@ -263,12 +228,12 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
@Override
public void offline(DeviceChannel channel) {
channelMapper.offline(channel.getDeviceId(), channel.getChannelId());
channelMapper.offline(channel.getDeviceId(), channel.getDeviceId());
}
@Override
public void delete(DeviceChannel channel) {
channelMapper.del(channel.getDeviceId(), channel.getChannelId());
channelMapper.del(channel.getDeviceId(), channel.getDeviceId());
}
@Override
@ -315,7 +280,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
if (ObjectUtils.isEmpty(channel.getStreamIdentification())) {
logger.info("[重置通道码流类型] 设备: {}, 码流： {}", channel.getDeviceId(), channel.getStreamIdentification());
}else {
logger.info("[更新通道码流类型] 设备: {}, 通道：{}， 码流： {}", channel.getDeviceId(), channel.getChannelId(),
logger.info("[更新通道码流类型] 设备: {}, 通道：{}， 码流： {}", channel.getDeviceId(), channel.getDeviceId(),
channel.getStreamIdentification());
}
channelMapper.updateChannelStreamIdentification(channel);
@ -332,8 +297,8 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
deviceMobilePositionMapper.insertNewPosition(mobilePosition);
}
if (deviceChannel.getChannelId().equals(deviceChannel.getDeviceId())) {
deviceChannel.setChannelId(null);
if (deviceChannel.getDeviceId().equals(deviceChannel.getDeviceId())) {
deviceChannel.setDeviceId(null);
}
if (deviceChannel.getGpsTime() == null) {
deviceChannel.setGpsTime(DateUtil.getNow());
@ -345,7 +310,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
List<DeviceChannel> deviceChannels = new ArrayList<>();
if (deviceChannel.getChannelId() == null) {
if (deviceChannel.getDeviceId() == null) {
// 有的设备这里上报的deviceId与通道Id是一样，这种情况更新设备下的全部通道
List<DeviceChannel> deviceChannelsInDb = queryChaneListByDeviceId(device.getDeviceId());
deviceChannels.addAll(deviceChannelsInDb);
@ -361,7 +326,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
for (DeviceChannel channel : deviceChannels) {
// 向关联了该通道并且开启移动位置订阅的上级平台发送移动位置订阅消息
mobilePosition.setChannelId(channel.getChannelId());
mobilePosition.setChannelId(channel.getDeviceId());
try {
eventPublisher.mobilePositionEventPublish(mobilePosition);
}catch (Exception e) {
@ -435,6 +400,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
@Override
@Transactional
public boolean resetChannels(String deviceId, List<DeviceChannel> deviceChannelList) {
if (CollectionUtils.isEmpty(deviceChannelList)) {
return false;
@ -443,10 +409,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
Map<String,DeviceChannel> allChannelMap = new ConcurrentHashMap<>();
if (allChannels.size() > 0) {
for (DeviceChannel deviceChannel : allChannels) {
allChannelMap.put(deviceChannel.getChannelId(), deviceChannel);
allChannelMap.put(deviceChannel.getDeviceId(), deviceChannel);
}
}
TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
// 数据去重
List<DeviceChannel> channels = new ArrayList<>();
@ -459,19 +424,19 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
// 数据去重
Set<String> gbIdSet = new HashSet<>();
for (DeviceChannel deviceChannel : deviceChannelList) {
if (gbIdSet.contains(deviceChannel.getChannelId())) {
stringBuilder.append(deviceChannel.getChannelId()).append(",");
if (gbIdSet.contains(deviceChannel.getDeviceId())) {
stringBuilder.append(deviceChannel.getDeviceId()).append(",");
continue;
}
gbIdSet.add(deviceChannel.getChannelId());
if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).getHasAudio());
if (allChannelMap.get(deviceChannel.getChannelId()).isStatus() !=deviceChannel.isStatus()){
List<String> strings = platformChannelMapper.queryParentPlatformByChannelId(deviceChannel.getChannelId());
gbIdSet.add(deviceChannel.getDeviceId());
if (allChannelMap.containsKey(deviceChannel.getDeviceId())) {
deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getDeviceId()).getStreamId());
deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getDeviceId()).getHasAudio());
if (allChannelMap.get(deviceChannel.getDeviceId()).getStatus().equalsIgnoreCase(deviceChannel.getStatus())){
List<String> strings = platformChannelMapper.queryParentPlatformByChannelId(deviceChannel.getDeviceId());
if (!CollectionUtils.isEmpty(strings)){
strings.forEach(platformId->{
eventPublisher.catalogEventPublish(platformId, deviceChannel, deviceChannel.isStatus()? CatalogEvent.ON:CatalogEvent.OFF);
eventPublisher.catalogEventPublish(platformId, deviceChannel, deviceChannel.getStatus().equals("ON")? CatalogEvent.ON:CatalogEvent.OFF);
});
}
@ -483,7 +448,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
deviceChannel.setUpdateTime(DateUtil.getNow());
addChannels.add(deviceChannel);
}
allChannelMap.remove(deviceChannel.getChannelId());
allChannelMap.remove(deviceChannel.getDeviceId());
channels.add(deviceChannel);
if (!ObjectUtils.isEmpty(deviceChannel.getParentId())) {
if (subContMap.get(deviceChannel.getParentId()) == null) {
@ -497,8 +462,8 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
deleteChannels.addAll(allChannelMap.values());
if (!channels.isEmpty()) {
for (DeviceChannel channel : channels) {
if (subContMap.get(channel.getChannelId()) != null){
Integer count = subContMap.get(channel.getChannelId());
if (subContMap.get(channel.getDeviceId()) != null){
Integer count = subContMap.get(channel.getDeviceId());
if (count > 0) {
channel.setSubCount(count);
channel.setParental(1);
@ -514,61 +479,50 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
logger.info("通道重设，数据为空={}" , deviceChannelList);
return false;
}
try {
int limitCount = 50;
boolean result = false;
if (!result && !addChannels.isEmpty()) {
if (addChannels.size() > limitCount) {
for (int i = 0; i < addChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > addChannels.size()) {
toIndex = addChannels.size();
}
result = result || channelMapper.batchAdd(addChannels.subList(i, toIndex)) < 0;
int limitCount = 50;
boolean result = false;
if (!result && !addChannels.isEmpty()) {
if (addChannels.size() > limitCount) {
for (int i = 0; i < addChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > addChannels.size()) {
toIndex = addChannels.size();
}
}else {
result = result || channelMapper.batchAdd(addChannels) < 0;
result = result || channelMapper.batchAdd(addChannels.subList(i, toIndex)) < 0;
}
}else {
result = result || channelMapper.batchAdd(addChannels) < 0;
}
if (!result && !updateChannels.isEmpty()) {
if (updateChannels.size() > limitCount) {
for (int i = 0; i < updateChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > updateChannels.size()) {
toIndex = updateChannels.size();
}
result = result || channelMapper.batchUpdate(updateChannels.subList(i, toIndex)) < 0;
}
}else {
result = result || channelMapper.batchUpdate(updateChannels) < 0;
}
}
if (!result && !deleteChannels.isEmpty()) {
System.out.println("删除： " + deleteChannels.size());
if (deleteChannels.size() > limitCount) {
for (int i = 0; i < deleteChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > deleteChannels.size()) {
toIndex = deleteChannels.size();
}
result = result || channelMapper.batchDel(deleteChannels.subList(i, toIndex)) < 0;
}
}else {
result = result || channelMapper.batchDel(deleteChannels) < 0;
}
}
if (result) {
//事务回滚
dataSourceTransactionManager.rollback(transactionStatus);
}
dataSourceTransactionManager.commit(transactionStatus); //手动提交
return true;
}catch (Exception e) {
logger.error("未处理的异常 ", e);
dataSourceTransactionManager.rollback(transactionStatus);
return false;
}
if (!result && !updateChannels.isEmpty()) {
if (updateChannels.size() > limitCount) {
for (int i = 0; i < updateChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > updateChannels.size()) {
toIndex = updateChannels.size();
}
result = result || channelMapper.batchUpdate(updateChannels.subList(i, toIndex)) < 0;
}
}else {
result = result || channelMapper.batchUpdate(updateChannels) < 0;
}
}
if (!result && !deleteChannels.isEmpty()) {
System.out.println("删除： " + deleteChannels.size());
if (deleteChannels.size() > limitCount) {
for (int i = 0; i < deleteChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > deleteChannels.size()) {
toIndex = deleteChannels.size();
}
result = result || channelMapper.batchDel(deleteChannels.subList(i, toIndex)) < 0;
}
}else {
result = result || channelMapper.batchDel(deleteChannels) < 0;
}
}
return true;
}
}

64
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
View File

@ -15,16 +15,15 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.storager.dao.DeviceMapper;
import com.genersoft.iot.vmp.storager.dao.PlatformChannelMapper;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.BaseTree;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -39,7 +38,8 @@ import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.text.ParseException;
import java.time.Instant;
import java.util.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
@ -408,29 +408,14 @@ public class DeviceServiceImpl implements IDeviceService {
*/
private void updateDeviceChannelGeoCoordSys(Device device) {
List<DeviceChannel> deviceChannels = deviceChannelMapper.getAllChannelWithCoordinate(device.getDeviceId());
if (deviceChannels.size() > 0) {
if (!deviceChannels.isEmpty()) {
List<DeviceChannel> deviceChannelsForStore = new ArrayList<>();
for (DeviceChannel deviceChannel : deviceChannels) {
deviceChannelsForStore.add(deviceChannelService.updateGps(deviceChannel, device));
}
deviceChannelsForStore.addAll(deviceChannels);
deviceChannelService.updateChannels(device.getDeviceId(), deviceChannelsForStore);
}
}
@Override
public List<BaseTree<DeviceChannel>> queryVideoDeviceTree(String deviceId, String parentId, boolean onlyCatalog) {
Device device = deviceMapper.getDeviceByDeviceId(deviceId);
if (device == null) {
return null;
}
if (ObjectUtils.isEmpty(parentId) ) {
parentId = deviceId;
}
List<DeviceChannel> rootNodes = deviceChannelMapper.getSubChannelsByDeviceId(deviceId, parentId, onlyCatalog);
return transportChannelsToTree(rootNodes, "");
}
@Override
public List<DeviceChannel> queryVideoDeviceInTreeNode(String deviceId, String parentId) {
Device device = deviceMapper.getDeviceByDeviceId(deviceId);
@ -444,45 +429,6 @@ public class DeviceServiceImpl implements IDeviceService {
}
}
private List<BaseTree<DeviceChannel>> transportChannelsToTree(List<DeviceChannel> channels, String parentId) {
if (channels == null) {
return null;
}
List<BaseTree<DeviceChannel>> treeNotes = new ArrayList<>();
if (channels.size() == 0) {
return treeNotes;
}
for (DeviceChannel channel : channels) {
BaseTree<DeviceChannel> node = new BaseTree<>();
node.setId(channel.getChannelId());
node.setDeviceId(channel.getDeviceId());
node.setName(channel.getName());
node.setPid(parentId);
node.setBasicData(channel);
node.setParent(false);
if (channel.getChannelId().length() <= 8) {
node.setParent(true);
}else {
if (channel.getChannelId().length() != 20) {
node.setParent(channel.getParental() == 1);
}else {
try {
int type = Integer.parseInt(channel.getChannelId().substring(10, 13));
if (type == 215 || type == 216 || type == 200) {
node.setParent(true);
}
}catch (NumberFormatException e) {
node.setParent(false);
}
}
}
treeNotes.add(node);
}
Collections.sort(treeNotes);
return treeNotes;
}
@Override
public boolean isExist(String deviceId) {
return deviceMapper.getDeviceByDeviceId(deviceId) != null;

276
src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java
View File

@ -1,276 +0,0 @@
package com.genersoft.iot.vmp.service.impl;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
import com.genersoft.iot.vmp.storager.dao.PlatformCatalogMapper;
import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.ObjectUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@Service
@DS("master")
public class GbStreamServiceImpl implements IGbStreamService {
private final static Logger logger = LoggerFactory.getLogger(GbStreamServiceImpl.class);
@Autowired
DataSourceTransactionManager dataSourceTransactionManager;
@Autowired
TransactionDefinition transactionDefinition;
@Autowired
private GbStreamMapper gbStreamMapper;
@Autowired
private PlatformGbStreamMapper platformGbStreamMapper;
@Autowired
private SubscribeHolder subscribeHolder;
@Autowired
private ParentPlatformMapper platformMapper;
@Autowired
private PlatformCatalogMapper catalogMapper;
@Autowired
private EventPublisher eventPublisher;
@Override
public PageInfo<GbStream> getAll(Integer page, Integer count, String platFormId, String catalogId, String query, String mediaServerId) {
PageHelper.startPage(page, count);
List<GbStream> all = gbStreamMapper.selectAll(platFormId, catalogId, query, mediaServerId);
return new PageInfo<>(all);
}
@Override
public void del(String app, String stream) {
gbStreamMapper.del(app, stream);
}
@Override
public boolean addPlatformInfo(List<GbStream> gbStreams, String platformId, String catalogId) {
// 放在事务内执行
boolean result = false;
TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
ParentPlatform parentPlatform = platformMapper.getParentPlatByServerGBId(platformId);
if (catalogId == null) {
catalogId = parentPlatform.getCatalogId();
}
try {
List<DeviceChannel> deviceChannelList = new ArrayList<>();
for (int i = 0; i < gbStreams.size(); i++) {
GbStream gbStream = gbStreams.get(i);
gbStream.setCatalogId(catalogId);
gbStream.setPlatformId(platformId);
// TODO 修改为批量提交
platformGbStreamMapper.add(gbStream);
logger.info("[关联通道]直播流通道 平台：{}, 共需关联通道数:{}, 已关联：{}", platformId, gbStreams.size(), i + 1);
DeviceChannel deviceChannelListByStream = getDeviceChannelListByStreamWithStatus(gbStream, catalogId, parentPlatform);
deviceChannelList.add(deviceChannelListByStream);
}
dataSourceTransactionManager.commit(transactionStatus); //手动提交
if (subscribeHolder.getCatalogSubscribe(platformId) != null) {
eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.ADD);
}
result = true;
}catch (Exception e) {
logger.error("批量保存流与平台的关系时错误", e);
dataSourceTransactionManager.rollback(transactionStatus);
}
return result;
}
@Override
public DeviceChannel getDeviceChannelListByStream(GbStream gbStream, String catalogId, ParentPlatform platform) {
DeviceChannel deviceChannel = new DeviceChannel();
deviceChannel.setChannelId(gbStream.getGbId());
deviceChannel.setName(gbStream.getName());
deviceChannel.setLongitude(gbStream.getLongitude());
deviceChannel.setLatitude(gbStream.getLatitude());
deviceChannel.setDeviceId(platform.getDeviceGBId());
deviceChannel.setManufacture("wvp-pro");
deviceChannel.setStatus(gbStream.isStatus());
deviceChannel.setRegisterWay(1);
PlatformCatalog catalog = catalogMapper.selectByPlatFormAndCatalogId(platform.getServerGBId(), catalogId);
if (catalog != null) {
deviceChannel.setCivilCode(catalog.getCivilCode());
deviceChannel.setParentId(catalog.getParentId());
deviceChannel.setBusinessGroupId(catalog.getBusinessGroupId());
}else {
deviceChannel.setCivilCode(platform.getAdministrativeDivision());
deviceChannel.setParentId(platform.getDeviceGBId());
}
deviceChannel.setModel("live");
deviceChannel.setOwner("wvp-pro");
deviceChannel.setParental(0);
deviceChannel.setSecrecy("0");
return deviceChannel;
}
@Override
public boolean delPlatformInfo(String platformId, List<GbStream> gbStreams) {
// 放在事务内执行
boolean result = false;
TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
try {
List<DeviceChannel> deviceChannelList = new ArrayList<>();
platformGbStreamMapper.delByAppAndStreamsByPlatformId(gbStreams, platformId);
for (GbStream gbStream : gbStreams) {
DeviceChannel deviceChannel = new DeviceChannel();
deviceChannel.setChannelId(gbStream.getGbId());
deviceChannelList.add(deviceChannel);
}
eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.DEL);
dataSourceTransactionManager.commit(transactionStatus); //手动提交
result = true;
}catch (Exception e) {
logger.error("批量移除流与平台的关系时错误", e);
dataSourceTransactionManager.rollback(transactionStatus);
}
return result;
}
@Override
public void sendCatalogMsg(CommonGBChannel channel, String type) {
if (channel == null || type == null) {
logger.warn("[发送目录订阅]类型：流信息或类型为NULL");
return;
}
List<CommonGBChannel> gbStreams = new ArrayList<>();
if (channel.getGbDeviceId() != null) {
gbStreams.add(channel);
}else {
GbStream gbStreamIndb = gbStreamMapper.selectOne(channel.getApp(), channel.getStream());
if (gbStreamIndb != null && gbStreamIndb.getGbId() != null){
gbStreams.add(gbStreamIndb);
}
}
sendCatalogMsgs(gbStreams, type);
}
@Override
public void sendCatalogMsgs(List<GbStream> gbStreams, String type) {
if (gbStreams.size() > 0) {
for (GbStream gs : gbStreams) {
if (ObjectUtils.isEmpty(gs.getGbId())){
continue;
}
List<ParentPlatform> parentPlatforms = platformGbStreamMapper.selectByAppAndStream(gs.getApp(), gs.getStream());
if (parentPlatforms.size() > 0) {
for (ParentPlatform parentPlatform : parentPlatforms) {
if (parentPlatform != null) {
eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), gs, type);
}
}
}
}
}
}
@Override
public int updateGbIdOrName(List<StreamPush> streamPushItemForUpdate) {
return gbStreamMapper.updateGbIdOrName(streamPushItemForUpdate);
}
@Override
public DeviceChannel getDeviceChannelListByStreamWithStatus(GbStream gbStream, String catalogId, ParentPlatform platform) {
DeviceChannel deviceChannel = new DeviceChannel();
deviceChannel.setChannelId(gbStream.getGbId());
deviceChannel.setName(gbStream.getName());
deviceChannel.setLongitude(gbStream.getLongitude());
deviceChannel.setLatitude(gbStream.getLatitude());
deviceChannel.setDeviceId(platform.getDeviceGBId());
deviceChannel.setManufacture("wvp-pro");
// todo 目前是每一条查询一次，需要优化
Boolean status = null;
if ("proxy".equals(gbStream.getStreamType())) {
status = gbStreamMapper.selectStatusForProxy(gbStream.getApp(), gbStream.getStream());
}else {
status = gbStreamMapper.selectStatusForPush(gbStream.getApp(), gbStream.getStream());
}
deviceChannel.setStatus(status != null && status);
deviceChannel.setRegisterWay(1);
PlatformCatalog catalog = catalogMapper.selectByPlatFormAndCatalogId(platform.getServerGBId(), catalogId);
if (catalog != null) {
deviceChannel.setCivilCode(catalog.getCivilCode());
deviceChannel.setParentId(catalog.getParentId());
deviceChannel.setBusinessGroupId(catalog.getBusinessGroupId());
}else {
deviceChannel.setCivilCode(platform.getAdministrativeDivision());
deviceChannel.setParentId(platform.getDeviceGBId());
}
deviceChannel.setModel("live");
deviceChannel.setOwner("wvp-pro");
deviceChannel.setParental(0);
deviceChannel.setSecrecy("0");
return deviceChannel;
}
@Override
public List<GbStream> getAllGBChannels(String platformId) {
return gbStreamMapper.selectAll(platformId, null, null, null);
}
@Override
public void delAllPlatformInfo(String platformId, String catalogId) {
if (platformId == null) {
return ;
}
ParentPlatform platform = platformMapper.getParentPlatByServerGBId(platformId);
if (platform == null) {
return ;
}
if (ObjectUtils.isEmpty(catalogId)) {
catalogId = null;
}
List<GbStream> gbStreams = platformGbStreamMapper.queryChannelInParentPlatformAndCatalog(platformId, catalogId);
List<DeviceChannel> deviceChannelList = new ArrayList<>();
for (GbStream gbStream : gbStreams) {
DeviceChannel deviceChannel = new DeviceChannel();
deviceChannel.setChannelId(gbStream.getGbId());
deviceChannelList.add(deviceChannel);
}
eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.DEL);
platformGbStreamMapper.delByPlatformAndCatalogId(platformId, catalogId);
}
@Override
public List<GbStream> getGbChannelWithGbid(String gbId) {
return gbStreamMapper.selectByGBId(gbId);
}
@Override
public Map<String, GbStream> getAllGBId() {
return gbStreamMapper.getAllGBId();
}
}

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

@ -114,7 +114,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
}
SubscribeInfo catalogSubscribe = subscribeHolder.getCatalogSubscribe(platformId);
if (catalogSubscribe != null) {
List<DeviceChannel> deviceChannelList = getDeviceChannelListByChannelReduceList(channelReducesToAdd, catalogId, platform);
List<CommonGBChannel> deviceChannelList = getDeviceChannelListByChannelReduceList(channelReducesToAdd, catalogId, platform);
if (deviceChannelList != null) {
eventPublisher.catalogEventPublish(platformId, deviceChannelList, CatalogEvent.ADD);
}
@ -167,7 +167,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
catalogId = null;
}
List<DeviceChannel> deviceChannels = platformChannelMapper.queryAllChannelInCatalog(platformId, catalogId);
List<CommonGBChannel> deviceChannels = platformChannelMapper.queryAllChannelInCatalog(platformId, catalogId);
eventPublisher.catalogEventPublish(platformId, deviceChannels, CatalogEvent.DEL);
return platformChannelMapper.delChannelForGBByCatalogId(platformId, catalogId);

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

@ -471,7 +471,7 @@ public class PlatformServiceImpl implements IPlatformService {
return;
}
for (DeviceChannel deviceChannel : gbStreams) {
String gbId = deviceChannel.getChannelId();
String gbId = deviceChannel.getDeviceId();
GPSMsgInfo gpsMsgInfo = redisCatchStorage.getGpsMsgInfo(gbId);
// 无最新位置不发送
if (gpsMsgInfo != null) {
@ -789,19 +789,19 @@ public class PlatformServiceImpl implements IPlatformService {
try {
if (sendBye) {
commanderForPlatform.streamByeCmd(platform, channel.getChannelId(), stream, null, null);
commanderForPlatform.streamByeCmd(platform, channel.getDeviceId(), stream, null, null);
}
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.warn("[消息发送失败] 停止语音对讲， 平台：{}，通道：{}", platform.getId(), channel.getChannelId() );
logger.warn("[消息发送失败] 停止语音对讲， 平台：{}，通道：{}", platform.getId(), channel.getDeviceId() );
} finally {
mediaServerService.closeRTPServer(mediaServerItem, stream);
InviteInfo inviteInfo = inviteStreamService.getInviteInfo(null, platform.getServerGBId(), channel.getChannelId(), stream);
InviteInfo inviteInfo = inviteStreamService.getInviteInfo(null, platform.getServerGBId(), channel.getDeviceId(), stream);
if (inviteInfo != null) {
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), inviteInfo.getSsrcInfo().getSsrc());
inviteStreamService.removeInviteInfo(inviteInfo);
}
streamSession.remove(platform.getServerGBId(), channel.getChannelId(), stream);
streamSession.remove(platform.getServerGBId(), channel.getDeviceId(), stream);
}
}
}

64
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@ -249,8 +249,9 @@ public class PlayServiceImpl implements IPlayService {
startTime, endTime
);
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(event.getMediaServer(), event.getStream(), null,
device.isSsrcCheck(), true, 0, false, !deviceChannel.getHasAudio(), false, device.getStreamModeForParam());
device.isSsrcCheck(), true, 0, false, !deviceChannel.getHasAudio(), false, tcpMode);
playBack(event.getMediaServer(), ssrcInfo, deviceId, channelId, startTime, endTime, null);
}
}
@ -312,7 +313,8 @@ public class PlayServiceImpl implements IPlayService {
}
}
String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, !channel.getHasAudio(), false, device.getStreamModeForParam());
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, !channel.getHasAudio(), false, tcpMode);
if (ssrcInfo == null) {
callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null,
@ -465,23 +467,23 @@ public class PlayServiceImpl implements IPlayService {
return;
}
logger.info("[点播开始] deviceId: {}, channelId: {},码流类型：{}, 收流端口： {}, 码流：{}, 收流模式：{}, SSRC: {}, SSRC校验：{}",
device.getDeviceId(), channel.getChannelId(), channel.getStreamIdentification(), ssrcInfo.getPort(), ssrcInfo.getStream(),
device.getDeviceId(), channel.getDeviceId(), channel.getStreamIdentification(), ssrcInfo.getPort(), ssrcInfo.getStream(),
device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
//端口获取失败的ssrcInfo 没有必要发送点播指令
if (ssrcInfo.getPort() <= 0) {
logger.info("[点播端口分配异常]，deviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channel.getChannelId(), ssrcInfo);
logger.info("[点播端口分配异常]，deviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channel.getDeviceId(), ssrcInfo);
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channel.getDeviceId(), ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "点播端口分配异常", null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId(), null,
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId(), null,
InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "点播端口分配异常", null);
return;
}
// 初始化redis中的invite消息状态
InviteInfo inviteInfo = InviteInfo.getInviteInfo(device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream(), ssrcInfo,
InviteInfo inviteInfo = InviteInfo.getInviteInfo(device.getDeviceId(), channel.getDeviceId(), ssrcInfo.getStream(), ssrcInfo,
mediaServerItem.getSdpIp(), ssrcInfo.getPort(), device.getStreamMode(), InviteSessionType.PLAY,
InviteSessionStatus.ready);
inviteStreamService.updateInviteInfo(inviteInfo);
@ -489,38 +491,38 @@ public class PlayServiceImpl implements IPlayService {
String timeOutTaskKey = UUID.randomUUID().toString();
dynamicTask.startDelay(timeOutTaskKey, () -> {
// 执行超时任务时查询是否已经成功，成功了则不执行超时任务，防止超时任务取消失败的情况
InviteInfo inviteInfoForTimeOut = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId());
InviteInfo inviteInfoForTimeOut = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId());
if (inviteInfoForTimeOut == null || inviteInfoForTimeOut.getStreamInfo() == null) {
logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {},码流：{}，端口：{}, SSRC: {}",
device.getDeviceId(), channel.getChannelId(), channel.getStreamIdentification(),
device.getDeviceId(), channel.getDeviceId(), channel.getStreamIdentification(),
ssrcInfo.getPort(), ssrcInfo.getSsrc());
callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId(), null,
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId(), null,
InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null);
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId());
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId());
try {
cmder.streamByeCmd(device, channel.getChannelId(), ssrcInfo.getStream(), null);
cmder.streamByeCmd(device, channel.getDeviceId(), ssrcInfo.getStream(), null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
logger.error("[点播超时]， 发送BYE失败 {}", e.getMessage());
} finally {
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channel.getDeviceId(), ssrcInfo.getStream());
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 取消订阅消息监听
subscribe.removeSubscribe(Hook.getInstance(HookType.on_media_arrival, "rtp", ssrcInfo.getStream(), mediaServerItem.getId()));
}
}else {
logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {},码流：{}，端口：{}, SSRC: {}",
device.getDeviceId(), channel.getChannelId(), channel.getStreamIdentification(),
device.getDeviceId(), channel.getDeviceId(), channel.getStreamIdentification(),
ssrcInfo.getPort(), ssrcInfo.getSsrc());
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channel.getDeviceId(), ssrcInfo.getStream());
}
}, userSetting.getPlayTimeout());
@ -529,42 +531,42 @@ public class PlayServiceImpl implements IPlayService {
logger.info("收到订阅消息： " + hookData);
dynamicTask.stop(timeOutTaskKey);
// hook响应
StreamInfo streamInfo = onPublishHandlerForPlay(hookData.getMediaServer(), hookData.getMediaInfo(), device.getDeviceId(), channel.getChannelId());
StreamInfo streamInfo = onPublishHandlerForPlay(hookData.getMediaServer(), hookData.getMediaInfo(), device.getDeviceId(), channel.getDeviceId());
if (streamInfo == null){
callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId(), null,
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId(), null,
InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(),
InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null);
return;
}
callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId(), null,
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId(), null,
InviteErrorCode.SUCCESS.getCode(),
InviteErrorCode.SUCCESS.getMsg(),
streamInfo);
logger.info("[点播成功] deviceId: {}, channelId:{}, 码流类型：{}", device.getDeviceId(), channel.getChannelId(),
logger.info("[点播成功] deviceId: {}, channelId:{}, 码流类型：{}", device.getDeviceId(), channel.getDeviceId(),
channel.getStreamIdentification());
snapOnPlay(hookData.getMediaServer(), device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
snapOnPlay(hookData.getMediaServer(), device.getDeviceId(), channel.getDeviceId(), ssrcInfo.getStream());
}, (eventResult) -> {
// 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channel.getChannelId(),
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channel.getDeviceId(),
timeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAY);
}, (event) -> {
logger.info("[点播失败] deviceId: {}, channelId:{}, {}: {}", device.getDeviceId(), channel.getChannelId(), event.statusCode, event.msg);
logger.info("[点播失败] deviceId: {}, channelId:{}, {}: {}", device.getDeviceId(), channel.getDeviceId(), event.statusCode, event.msg);
dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channel.getDeviceId(), ssrcInfo.getStream());
callback.run(event.statusCode, event.msg, null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId(), null,
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId(), null,
InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(),
String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg), null);
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId());
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId());
});
} catch (InvalidArgumentException | SipException | ParseException e) {
@ -574,15 +576,15 @@ public class PlayServiceImpl implements IPlayService {
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channel.getDeviceId(), ssrcInfo.getStream());
callback.run(InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(),
InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getMsg(), null);
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId(), null,
inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId(), null,
InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(),
InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getMsg(), null);
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getChannelId());
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channel.getDeviceId());
}
}
@ -757,7 +759,8 @@ public class PlayServiceImpl implements IPlayService {
.replace(":", "")
.replace(" ", "");
String stream = deviceId + "_" + channelId + "_" + startTimeStr + "_" + endTimeTimeStr;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, !channel.getHasAudio(), false, device.getStreamModeForParam());
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, !channel.getHasAudio(), false, tcpMode);
playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, callback);
}
@ -953,8 +956,9 @@ public class PlayServiceImpl implements IPlayService {
null);
return;
}
int tcpMode = device.getStreamMode().equals("TCP-ACTIVE")? 2: (device.getStreamMode().equals("TCP-PASSIVE")? 1:0);
// 录像下载不使用固定流地址，固定流地址会导致如果开始时间与结束时间一致时文件错误的叠加在一起
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,!channel.getHasAudio(), false, device.getStreamModeForParam());
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,!channel.getHasAudio(), false, tcpMode);
download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, callback);
}

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

@ -3,7 +3,6 @@ package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
import com.genersoft.iot.vmp.utils.DateUtil;
@ -35,9 +34,6 @@ public class RedisPushStreamStatusListMsgListener implements MessageListener {
@Resource
private IStreamPushService streamPushService;
@Resource
private IGbStreamService gbStreamService;
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@ -103,7 +99,7 @@ public class RedisPushStreamStatusListMsgListener implements MessageListener {
if(!streamPushItemForUpdate.isEmpty()){
log.info("修改{}条",streamPushItemForUpdate.size());
log.info(JSONObject.toJSONString(streamPushItemForUpdate));
gbStreamService.updateGbIdOrName(streamPushItemForUpdate);
streamPushService.batchUpdate(streamPushItemForUpdate);
}
}catch (Exception e) {
log.warn("[REDIS消息-推流设备列表更新] 发现未处理的异常, \r\n{}", new String(message.getBody()));

7
src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java
View File

@ -1,9 +1,6 @@
package com.genersoft.iot.vmp.storager.dao;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
import org.apache.ibatis.annotations.Delete;
import org.apache.ibatis.annotations.Insert;
@ -62,7 +59,7 @@ public interface PlatformChannelMapper {
"SELECT dc.* from wvp_platform_gb_channel pgc left join wvp_device_channel dc on dc.id = pgc.device_channel_id WHERE pgc.platform_id=#{platformId} " +
" <if test='catalogId != null' > and pgc.catalog_id=#{catalogId}</if>" +
"</script>")
List<DeviceChannel> queryAllChannelInCatalog(@Param("platformId") String platformId, @Param("catalogId") String catalogId);
List<CommonGBChannel> queryAllChannelInCatalog(@Param("platformId") String platformId, @Param("catalogId") String catalogId);
@Select(" select dc.channel_id as id, dc.name as name, pgc.platform_id as platform_id, pgc.catalog_id as parent_id, 0 as children_count, 1 as type " +
" from wvp_device_channel dc left join wvp_platform_gb_channel pgc on dc.id = pgc.device_channel_id " +

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

@ -7,9 +7,7 @@ import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
@ -19,7 +17,6 @@ import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOfflineEvent;
import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOnlineEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
@ -29,10 +26,8 @@ import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
@ -74,9 +69,6 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
@Autowired
private PlatformGbStreamMapper platformGbStreamMapper;
@Autowired
private IGbStreamService gbStreamService;
@Autowired
private IMediaServerService mediaServerService;
@ -412,17 +404,23 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
}
@Override
@Transactional
public int updateStatusByAppAndStream(String app, String stream, boolean status) {
// 状态变化时推送到国标上级
StreamProxy streamProxyItem = streamProxyMapper.selectOne(app, stream);
if (streamProxyItem == null) {
return 0;
}
int result = streamProxyMapper.updateStatus(app, stream, status);
if (!ObjectUtils.isEmpty(streamProxyItem.getGbId())) {
gbStreamService.sendCatalogMsg(streamProxyItem, status?CatalogEvent.ON:CatalogEvent.OFF);
streamProxyItem.setGbStatus(status);
if (streamProxyItem.getGbId() > 0) {
if (status) {
gbChannelService.online(streamProxyItem.getCommonGBChannel());
}else {
gbChannelService.offline(streamProxyItem.getCommonGBChannel());
}
}
return result;
return 1;
}
@Override

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

@ -90,4 +90,6 @@ public interface IStreamPushService {
void deleteByAppAndStream(String app, String stream);
void updatePushStatus(Integer streamPushId, boolean pushIng);
void batchUpdate(List<StreamPush> streamPushItemForUpdate);
}

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

@ -549,4 +549,8 @@ public class StreamPushServiceImpl implements IStreamPushService {
return new ArrayList<>(result.values());
}
@Override
public void batchUpdate(List<StreamPush> streamPushItemForUpdate) {
}
}

87
src/main/java/com/genersoft/iot/vmp/vmanager/bean/BaseTree.java
View File

@ -1,87 +0,0 @@
package com.genersoft.iot.vmp.vmanager.bean;
import org.jetbrains.annotations.NotNull;
import java.text.Collator;
import java.util.Comparator;
/**
* @author lin
*/
public class BaseTree<T> implements Comparable<BaseTree>{
private String id;
private String deviceId;
private String pid;
private String name;
private boolean parent;
private T basicData;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getDeviceId() {
return deviceId;
}
public void setDeviceId(String deviceId) {
this.deviceId = deviceId;
}
public String getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public T getBasicData() {
return basicData;
}
public void setBasicData(T basicData) {
this.basicData = basicData;
}
public boolean isParent() {
return parent;
}
public void setParent(boolean parent) {
this.parent = parent;
}
@Override
public int compareTo(@NotNull BaseTree treeNode) {
if (this.parent || treeNode.isParent()) {
if (!this.parent && !treeNode.isParent()) {
Comparator cmp = Collator.getInstance(java.util.Locale.CHINA);
return cmp.compare(treeNode.getName(), this.getName());
}else {
if (this.isParent()) {
return 1;
}else {
return -1;
}
}
}else{
Comparator cmp = Collator.getInstance(java.util.Locale.CHINA);
return cmp.compare(treeNode.getName(), this.getName());
}
}
}

98
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java
View File

@ -18,7 +18,6 @@ import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.BaseTree;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.github.pagehelper.PageInfo;
@ -47,7 +46,10 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.text.ParseException;
import java.util.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@Tag(name = "国标设备查询", description = "国标设备查询")
@SuppressWarnings("rawtypes")
@ -494,96 +496,4 @@ public class DeviceQuery {
resp.setStatus(HttpServletResponse.SC_NOT_FOUND);
}
}
/**
* 查询国标树
* @param deviceId 设备ID
* @param parentId 父ID
* @param page 当前页
* @param count 每页条数
* @return 国标设备
*/
@Operation(summary = "查询国标树")
@Parameter(name = "deviceId", description = "设备国标编号", required = true)
@Parameter(name = "parentId", description = "父级国标编号")
@Parameter(name = "onlyCatalog", description = "只获取目录")
@Parameter(name = "page", description = "当前页", required = true)
@Parameter(name = "count", description = "每页条数", required = true)
@GetMapping("/tree/{deviceId}")
public ResponseEntity<PageInfo> getTree(@PathVariable String deviceId,
@RequestParam(required = false) String parentId,
@RequestParam(required = false) Boolean onlyCatalog,
int page, int count){
if (page <= 0) {
page = 1;
}
if (onlyCatalog == null) {
onlyCatalog = false;
}
List<BaseTree<DeviceChannel>> treeData = deviceService.queryVideoDeviceTree(deviceId, parentId, onlyCatalog);
if (treeData == null || (page - 1) * count > treeData.size()) {
PageInfo<BaseTree<DeviceChannel>> pageInfo = new PageInfo<>();
pageInfo.setPageNum(page);
pageInfo.setTotal(treeData == null? 0 : treeData.size());
pageInfo.setSize(0);
pageInfo.setList(new ArrayList<>());
return new ResponseEntity<>(pageInfo,HttpStatus.OK);
}
int toIndex = Math.min(page * count, treeData.size());
// 处理分页
List<BaseTree<DeviceChannel>> trees = treeData.subList((page - 1) * count, toIndex);
PageInfo<BaseTree<DeviceChannel>> pageInfo = new PageInfo<>();
pageInfo.setPageNum(page);
pageInfo.setTotal(treeData.size());
pageInfo.setSize(trees.size());
pageInfo.setList(trees);
return new ResponseEntity<>(pageInfo,HttpStatus.OK);
}
/**
* 查询国标树下的通道
* @param deviceId 设备ID
* @param parentId 父ID
* @param page 当前页
* @param count 每页条数
* @return 国标设备
*/
@Operation(summary = "查询国标树下的通道")
@Parameter(name = "deviceId", description = "设备国标编号", required = true)
@Parameter(name = "parentId", description = "父级国标编号")
@Parameter(name = "page", description = "当前页", required = true)
@Parameter(name = "count", description = "每页条数", required = true)
@GetMapping("/tree/channel/{deviceId}")
public ResponseEntity<PageInfo> getChannelInTreeNode(@PathVariable String deviceId, @RequestParam(required = false) String parentId, int page, int count){
if (page <= 0) {
page = 1;
}
List<DeviceChannel> treeData = deviceService.queryVideoDeviceInTreeNode(deviceId, parentId);
if (treeData == null || (page - 1) * count > treeData.size()) {
PageInfo<BaseTree<DeviceChannel>> pageInfo = new PageInfo<>();
pageInfo.setPageNum(page);
pageInfo.setTotal(treeData == null? 0 : treeData.size());
pageInfo.setSize(0);
pageInfo.setList(new ArrayList<>());
return new ResponseEntity<>(pageInfo,HttpStatus.OK);
}
int toIndex = Math.min(page * count, treeData.size());
// 处理分页
List<DeviceChannel> trees = treeData.subList((page - 1) * count, toIndex);
PageInfo<DeviceChannel> pageInfo = new PageInfo<>();
pageInfo.setPageNum(page);
pageInfo.setTotal(treeData.size());
pageInfo.setSize(trees.size());
pageInfo.setList(trees);
return new ResponseEntity<>(pageInfo,HttpStatus.OK);
}
}

134
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/gbStream/GbStreamController.java
View File

@ -1,134 +0,0 @@
package com.genersoft.iot.vmp.vmanager.gb28181.gbStream;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.gb28181.gbStream.bean.GbStreamParam;
import com.github.pagehelper.PageInfo;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.security.SecurityRequirement;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.*;
import java.util.List;
@Tag(name = "视频流关联到级联平台")
@RestController
@RequestMapping("/api/gbStream")
public class GbStreamController {
private final static Logger logger = LoggerFactory.getLogger(GbStreamController.class);
@Autowired
private IGbStreamService gbStreamService;
@Autowired
private IStreamPushService service;
@Autowired
private IPlatformService platformService;
/**
* 查询国标通道
* @param page 当前页
* @param count 每页条数
* @param platformId 平台ID
* @return
*/
@Operation(summary = "查询国标通道", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "page", description = "当前页", required = true)
@Parameter(name = "count", description = "每页条数", required = true)
@Parameter(name = "platformId", description = "平台ID", required = true)
@Parameter(name = "catalogId", description = "目录ID")
@Parameter(name = "query", description = "查询内容")
@Parameter(name = "mediaServerId", description = "流媒体ID")
@GetMapping(value = "/list")
@ResponseBody
public PageInfo<GbStream> list(@RequestParam(required = true)Integer page,
@RequestParam(required = true)Integer count,
@RequestParam(required = true)String platformId,
@RequestParam(required = false)String catalogId,
@RequestParam(required = false)String query,
@RequestParam(required = false)String mediaServerId){
if (ObjectUtils.isEmpty(catalogId)) {
catalogId = null;
}
if (ObjectUtils.isEmpty(query)) {
query = null;
}
if (ObjectUtils.isEmpty(mediaServerId)) {
mediaServerId = null;
}
// catalogId 为null 查询未在平台下分配的数据
// catalogId 不为null 查询平台下这个，目录下的通道
return gbStreamService.getAll(page, count, platformId, catalogId, query, mediaServerId);
}
/**
* 移除国标关联
* @param gbStreamParam
* @return
*/
@Operation(summary = "移除国标关联", security = @SecurityRequirement(name = JwtUtils.HEADER))
@DeleteMapping(value = "/del")
@ResponseBody
public void del(@RequestBody GbStreamParam gbStreamParam){
if (gbStreamParam.getGbStreams() == null || gbStreamParam.getGbStreams().isEmpty()) {
if (gbStreamParam.isAll()) {
gbStreamService.delAllPlatformInfo(gbStreamParam.getPlatformId(), gbStreamParam.getCatalogId());
}
}else {
gbStreamService.delPlatformInfo(gbStreamParam.getPlatformId(), gbStreamParam.getGbStreams());
}
}
/**
* 保存国标关联
* @param gbStreamParam
* @return
*/
@Operation(summary = "保存国标关联", security = @SecurityRequirement(name = JwtUtils.HEADER))
@PostMapping(value = "/add")
@ResponseBody
public void add(@RequestBody GbStreamParam gbStreamParam){
if (gbStreamParam.getGbStreams() == null || gbStreamParam.getGbStreams().isEmpty()) {
if (gbStreamParam.isAll()) {
List<GbStream> allGBChannels = gbStreamService.getAllGBChannels(gbStreamParam.getPlatformId());
gbStreamService.addPlatformInfo(allGBChannels, gbStreamParam.getPlatformId(), gbStreamParam.getCatalogId());
}
}else {
gbStreamService.addPlatformInfo(gbStreamParam.getGbStreams(), gbStreamParam.getPlatformId(), gbStreamParam.getCatalogId());
}
}
/**
* 保存国标关联
* @param gbId
* @return
*/
@Operation(summary = "保存国标关联", security = @SecurityRequirement(name = JwtUtils.HEADER))
@GetMapping(value = "/addWithGbid")
@ResponseBody
public void add(String gbId, String platformGbId, @RequestParam(required = false) String catalogGbId){
List<GbStream> gbStreams = gbStreamService.getGbChannelWithGbid(gbId);
if (gbStreams.isEmpty()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "gbId的信息未找到");
}
gbStreamService.addPlatformInfo(gbStreams, platformGbId, catalogGbId);
}
}

54
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/gbStream/bean/GbStreamParam.java
View File

@ -1,54 +0,0 @@
package com.genersoft.iot.vmp.vmanager.gb28181.gbStream.bean;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import io.swagger.v3.oas.annotations.media.Schema;
import java.util.List;
@Schema(description = "国标关联参数")
public class GbStreamParam {
@Schema(description = "平台ID")
private String platformId;
@Schema(description = "目录ID")
private String catalogId;
@Schema(description = "关联所有通道")
private boolean all;
@Schema(description = "流国标信息列表")
private List<GbStream> gbStreams;
public String getPlatformId() {
return platformId;
}
public String getCatalogId() {
return catalogId;
}
public void setCatalogId(String catalogId) {
this.catalogId = catalogId;
}
public void setPlatformId(String platformId) {
this.platformId = platformId;
}
public List<GbStream> getGbStreams() {
return gbStreams;
}
public void setGbStreams(List<GbStream> gbStreams) {
this.gbStreams = gbStreams;
}
public boolean isAll() {
return all;
}
public void setAll(boolean all) {
this.all = all;
}
}

3
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/platform/PlatformController.java
View File

@ -77,9 +77,6 @@ public class PlatformController {
@Autowired
private IDeviceChannelService deviceChannelService;
@Autowired
private IGbStreamService gbStreamService;
/**
* 获取国标服务的配置
*

2
src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java
View File

@ -108,7 +108,7 @@ public class ApiStreamController {
resultJSON.put("error","channel[ " + code + " ]未找到");
result.setResult(resultJSON);
return result;
}else if (!deviceChannel.isStatus()) {
}else if (!deviceChannel.getStatus().equalsIgnoreCase("ON")) {
JSONObject resultJSON = new JSONObject();
resultJSON.put("error","channel[ " + code + " ]offline");
result.setResult(resultJSON);