XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/conf/security/WebSecurityConfig.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
This commit is contained in:
648540858
2023-12-18 14:23:48 +08:00
parent 01ecb07a2f 76d09bcadb
commit 67d1ff0d26
32 changed files with 850 additions and 697 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/java/com/genersoft/iot/vmp/conf/security/JwtAuthenticationFilter.java
View File

@@ -78,6 +78,7 @@ public class JwtAuthenticationFilter extends OncePerRequestFilter {
// 构建UsernamePasswordAuthenticationToken,这里密码为null，是因为提供了正确的JWT,实现自动登录
User user = new User();
user.setId(jwtUser.getUserId());
user.setUsername(jwtUser.getUserName());
user.setPassword(jwtUser.getPassword());
Role role = new Role();

1
src/main/java/com/genersoft/iot/vmp/conf/security/JwtUtils.java
View File

@@ -144,6 +144,7 @@ public class JwtUtils implements InitializingBean {
jwtUser.setUserName(username);
jwtUser.setPassword(user.getPassword());
jwtUser.setRoleId(user.getRole().getId());
jwtUser.setUserId(user.getId());
return jwtUser;
} catch (InvalidJwtException e) {

9
src/main/java/com/genersoft/iot/vmp/conf/security/WebSecurityConfig.java
View File

@@ -1,12 +1,12 @@
package com.genersoft.iot.vmp.conf.security;
import com.genersoft.iot.vmp.conf.UserSetting;
import org.springframework.core.annotation.Order;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.security.authentication.AuthenticationManager;
import org.springframework.security.authentication.dao.DaoAuthenticationProvider;
import org.springframework.security.config.annotation.authentication.builders.AuthenticationManagerBuilder;
@@ -28,6 +28,7 @@ import java.util.Arrays;
/**
* 配置Spring Security
*
* @author lin
*/
@Configuration
@@ -75,6 +76,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
matchers.add("/js/**");
matchers.add("/api/device/query/snap/**");
matchers.add("/record_proxy/*/**");
matchers.add("/api/emit");
// 可以直接访问的静态数据
web.ignoring().antMatchers(matchers.toArray(new String[0]));
}
@@ -82,6 +84,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
/**
* 配置认证方式
*
* @param auth
* @throws Exception
*/
@@ -123,7 +126,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
}
CorsConfigurationSource configurationSource(){
CorsConfigurationSource configurationSource() {
// 配置跨域
CorsConfiguration corsConfiguration = new CorsConfiguration();
corsConfiguration.setAllowedHeaders(Arrays.asList("*"));
@@ -134,7 +137,7 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
corsConfiguration.setExposedHeaders(Arrays.asList(JwtUtils.getHeader()));
UrlBasedCorsConfigurationSource url = new UrlBasedCorsConfigurationSource();
url.registerCorsConfiguration("/**",corsConfiguration);
url.registerCorsConfiguration("/**", corsConfiguration);
return url;
}

9
src/main/java/com/genersoft/iot/vmp/conf/security/dto/JwtUser.java
View File

@@ -21,6 +21,7 @@ public class JwtUser {
EXCEPTION
}
private int userId;
private String userName;
private String password;
@@ -29,6 +30,14 @@ public class JwtUser {
private TokenStatus status;
public int getUserId() {
return userId;
}
public void setUserId(int userId) {
this.userId = userId;
}
public String getUserName() {
return userName;
}

176
src/main/java/com/genersoft/iot/vmp/gb28181/conf/StackLoggerImpl.java
View File

@@ -1,8 +1,8 @@
package com.genersoft.iot.vmp.gb28181.conf;
import gov.nist.core.StackLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.spi.LocationAwareLogger;
import org.springframework.stereotype.Component;
import java.util.Properties;
@@ -10,100 +10,132 @@ import java.util.Properties;
@Component
public class StackLoggerImpl implements StackLogger {
private final static Logger logger = LoggerFactory.getLogger(StackLoggerImpl.class);
/**
* 完全限定类名(Fully Qualified Class Name)，用于定位日志位置
*/
private static final String FQCN = StackLoggerImpl.class.getName();
@Override
public void logStackTrace() {
/**
* 获取栈中类信息(以便底层日志记录系统能够提取正确的位置信息(方法名、行号))
* @return LocationAwareLogger
*/
private static LocationAwareLogger getLocationAwareLogger() {
return (LocationAwareLogger) LoggerFactory.getLogger(new Throwable().getStackTrace()[4].getClassName());
}
}
@Override
public void logStackTrace(int traceLevel) {
System.out.println("traceLevel: " + traceLevel);
}
/**
* 封装打印日志的位置信息
* @param level 日志级别
* @param message 日志事件的消息
*/
private static void log(int level, String message) {
LocationAwareLogger locationAwareLogger = getLocationAwareLogger();
locationAwareLogger.log(null, FQCN, level, message, null, null);
}
@Override
public int getLineCount() {
return 0;
}
/**
* 封装打印日志的位置信息
* @param level 日志级别
* @param message 日志事件的消息
*/
private static void log(int level, String message, Throwable throwable) {
LocationAwareLogger locationAwareLogger = getLocationAwareLogger();
locationAwareLogger.log(null, FQCN, level, message, null, throwable);
}
@Override
public void logException(Throwable ex) {
@Override
public void logStackTrace() {
}
}
@Override
public void logDebug(String message) {
// logger.debug(message);
}
@Override
public void logStackTrace(int traceLevel) {
System.out.println("traceLevel: " + traceLevel);
}
@Override
public void logDebug(String message, Exception ex) {
// logger.debug(message);
}
@Override
public int getLineCount() {
return 0;
}
@Override
public void logTrace(String message) {
logger.trace(message);
}
@Override
public void logException(Throwable ex) {
@Override
public void logFatalError(String message) {
// logger.error(message);
}
}
@Override
public void logError(String message) {
// logger.error(message);
}
@Override
public void logDebug(String message) {
log(LocationAwareLogger.INFO_INT, message);
}
@Override
public boolean isLoggingEnabled() {
return true;
}
@Override
public void logDebug(String message, Exception ex) {
log(LocationAwareLogger.INFO_INT, message, ex);
}
@Override
public boolean isLoggingEnabled(int logLevel) {
return true;
}
@Override
public void logTrace(String message) {
log(LocationAwareLogger.INFO_INT, message);
}
@Override
public void logError(String message, Exception ex) {
// logger.error(message);
}
@Override
public void logFatalError(String message) {
log(LocationAwareLogger.INFO_INT, message);
}
@Override
public void logWarning(String message) {
logger.warn(message);
}
@Override
public void logError(String message) {
log(LocationAwareLogger.INFO_INT, message);
}
@Override
public void logInfo(String message) {
logger.info(message);
}
@Override
public boolean isLoggingEnabled() {
return true;
}
@Override
public void disableLogging() {
@Override
public boolean isLoggingEnabled(int logLevel) {
return true;
}
}
@Override
public void logError(String message, Exception ex) {
log(LocationAwareLogger.INFO_INT, message, ex);
}
@Override
public void enableLogging() {
@Override
public void logWarning(String message) {
log(LocationAwareLogger.INFO_INT, message);
}
}
@Override
public void logInfo(String message) {
log(LocationAwareLogger.INFO_INT, message);
}
@Override
public void setBuildTimeStamp(String buildTimeStamp) {
@Override
public void disableLogging() {
}
}
@Override
public void setStackProperties(Properties stackProperties) {
@Override
public void enableLogging() {
}
}
@Override
public String getLoggerName() {
return null;
}
@Override
public void setBuildTimeStamp(String buildTimeStamp) {
}
@Override
public void setStackProperties(Properties stackProperties) {
}
@Override
public String getLoggerName() {
return null;
}
}

77
src/main/java/com/genersoft/iot/vmp/gb28181/event/alarm/AlarmEventListener.java
View File

@@ -1,55 +1,68 @@
package com.genersoft.iot.vmp.gb28181.event.alarm;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Map;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import java.io.PrintWriter;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* @description: 报警事件监听
* @author: lawrencehj
* @data: 2021-01-20
* 报警事件监听器.
*
* @author lawrencehj
* @author <a href="mailto:xiaoQQya@126.com">xiaoQQya</a>
* @since 2021/01/20
*/
@Component
public class AlarmEventListener implements ApplicationListener<AlarmEvent> {
private final static Logger logger = LoggerFactory.getLogger(AlarmEventListener.class);
private static final Logger logger = LoggerFactory.getLogger(AlarmEventListener.class);
private static Map<String, SseEmitter> sseEmitters = new Hashtable<>();
private static final Map<String, PrintWriter> SSE_CACHE = new ConcurrentHashMap<>();
public void addSseEmitters(String browserId, SseEmitter sseEmitter) {
sseEmitters.put(browserId, sseEmitter);
public void addSseEmitter(String browserId, PrintWriter writer) {
SSE_CACHE.put(browserId, writer);
logger.info("SSE 在线数量: {}", SSE_CACHE.size());
}
public void removeSseEmitter(String browserId, PrintWriter writer) {
SSE_CACHE.remove(browserId, writer);
logger.info("SSE 在线数量: {}", SSE_CACHE.size());
}
@Override
public void onApplicationEvent(AlarmEvent event) {
public void onApplicationEvent(@NotNull AlarmEvent event) {
if (logger.isDebugEnabled()) {
logger.debug("设备报警事件触发，deviceId：" + event.getAlarmInfo().getDeviceId() + ", "
+ event.getAlarmInfo().getAlarmDescription());
logger.debug("设备报警事件触发, deviceId: {}, {}", event.getAlarmInfo().getDeviceId(), event.getAlarmInfo().getAlarmDescription());
}
String msg = "<strong>设备编码：</strong> <i>" + event.getAlarmInfo().getDeviceId() + "</i>"
+ "<br><strong>报警描述：</strong> <i>" + event.getAlarmInfo().getAlarmDescription() + "</i>"
+ "<br><strong>报警时间：</strong> <i>" + event.getAlarmInfo().getAlarmTime() + "</i>"
+ "<br><strong>报警位置：</strong> <i>" + event.getAlarmInfo().getLongitude() + "</i>"
+ ", <i>" + event.getAlarmInfo().getLatitude() + "</i>";
for (Iterator<Map.Entry<String, SseEmitter>> it = sseEmitters.entrySet().iterator(); it.hasNext();) {
Map.Entry<String, SseEmitter> emitter = it.next();
logger.info("推送到SSE连接，浏览器ID: " + emitter.getKey());
String msg = "<strong>设备编号：</strong> <i>" + event.getAlarmInfo().getDeviceId() + "</i>"
+ "<br><strong>通道编号：</strong> <i>" + event.getAlarmInfo().getChannelId() + "</i>"
+ "<br><strong>报警描述：</strong> <i>" + event.getAlarmInfo().getAlarmDescription() + "</i>"
+ "<br><strong>报警时间：</strong> <i>" + event.getAlarmInfo().getAlarmTime() + "</i>";
for (Iterator<Map.Entry<String, PrintWriter>> it = SSE_CACHE.entrySet().iterator(); it.hasNext(); ) {
Map.Entry<String, PrintWriter> response = it.next();
logger.info("推送到 SSE 连接, 浏览器 ID: {}", response.getKey());
try {
emitter.getValue().send(msg);
} catch (IOException | IllegalStateException e) {
if (logger.isDebugEnabled()) {
logger.debug("SSE连接已关闭");
PrintWriter writer = response.getValue();
if (writer.checkError()) {
it.remove();
continue;
}
// 移除已关闭的连接
String sseMsg = "event:message\n" +
"data:" + msg + "\n" +
"\n";
writer.write(sseMsg);
writer.flush();
} catch (Exception e) {
it.remove();
}
}

15
src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java
View File

@@ -144,8 +144,19 @@ public class VideoStreamSessionManager {
}
public void remove(String deviceId, String channelId, String stream) {
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream);
if (ssrcTransaction == null) {
List<SsrcTransaction> ssrcTransactionList = getSsrcTransactionForAll(deviceId, channelId, null, stream);
if (ssrcTransactionList == null || ssrcTransactionList.isEmpty()) {
return;
}
for (SsrcTransaction ssrcTransaction : ssrcTransactionList) {
redisTemplate.delete(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_"
+ deviceId + "_" + channelId + "_" + ssrcTransaction.getCallId() + "_" + ssrcTransaction.getStream());
}
}
public void removeByCallId(String deviceId, String channelId, String callId) {
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, callId, null);
if (ssrcTransaction == null ) {
return;
}
redisTemplate.delete(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_"

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java
View File

@@ -164,6 +164,7 @@ public class SIPRequestHeaderProvider {
Request request = null;
//请求行
SipURI requestLine = SipFactory.getInstance().createAddressFactory().createSipURI(channelId, device.getHostAddress());
// SipURI requestLine = SipFactory.getInstance().createAddressFactory().createSipURI(device.getDeviceId(), device.getHostAddress());
// via
ArrayList<ViaHeader> viaHeaders = new ArrayList<ViaHeader>();
ViaHeader viaHeader = SipFactory.getInstance().createHeaderFactory().createViaHeader(sipLayer.getLocalIp(device.getLocalIp()), sipConfig.getPort(), device.getTransport(), SipUtils.getNewViaTag());
@@ -174,6 +175,7 @@ public class SIPRequestHeaderProvider {
FromHeader fromHeader = SipFactory.getInstance().createHeaderFactory().createFromHeader(fromAddress, transactionInfo.getFromTag());
//to
SipURI toSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(channelId,device.getHostAddress());
// SipURI toSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(device.getDeviceId(),device.getHostAddress());
Address toAddress = SipFactory.getInstance().createAddressFactory().createAddress(toSipURI);
ToHeader toHeader = SipFactory.getInstance().createHeaderFactory().createToHeader(toAddress, transactionInfo.getToTag());

22
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
View File

@@ -40,6 +40,7 @@ import javax.sip.SipFactory;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Request;
import java.text.ParseException;
import java.util.List;
/**
* @description:设备能力接口，用于定义设备的控制、查询能力
@@ -373,7 +374,8 @@ public class SIPCommander implements ISIPCommander {
}), e -> {
ResponseEvent responseEvent = (ResponseEvent) e.event;
SIPResponse response = (SIPResponse) responseEvent.getResponse();
streamSession.put(device.getDeviceId(), channelId, "play", stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response,
String callId = response.getCallIdHeader().getCallId();
streamSession.put(device.getDeviceId(), channelId, callId, stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response,
InviteSessionType.PLAY);
okEvent.response(e);
});
@@ -611,17 +613,21 @@ public class SIPCommander implements ISIPCommander {
*/
@Override
public void streamByeCmd(Device device, String channelId, String stream, String callId, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException, SsrcTransactionNotFoundException {
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, callId, stream);
if (ssrcTransaction == null) {
List<SsrcTransaction> ssrcTransactionList = streamSession.getSsrcTransactionForAll(device.getDeviceId(), channelId, callId, stream);
if (ssrcTransactionList == null || ssrcTransactionList.isEmpty()) {
logger.info("[发送BYE] 未找到事务信息,设备： device: {}, channel: {}", device.getDeviceId(), channelId);
throw new SsrcTransactionNotFoundException(device.getDeviceId(), channelId, callId, stream);
}
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
streamSession.remove(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
for (SsrcTransaction ssrcTransaction : ssrcTransactionList) {
logger.info("[发送BYE] 设备： device: {}, channel: {}, callId: {}", device.getDeviceId(), channelId, ssrcTransaction.getCallId());
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
Request byteRequest = headerProvider.createByteRequest(device, channelId, ssrcTransaction.getSipTransactionInfo());
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), byteRequest, null, okEvent);
mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
streamSession.removeByCallId(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getCallId());
Request byteRequest = headerProvider.createByteRequest(device, channelId, ssrcTransaction.getSipTransactionInfo());
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), byteRequest, null, okEvent);
}
}
/**

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

@@ -179,7 +179,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
if (mediaServerItem != null) {
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransaction.getSsrc());
}
streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcTransaction.getStream());
streamSession.removeByCallId(device.getDeviceId(), channel.getChannelId(), ssrcTransaction.getCallId());
}
}
}

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

@@ -61,7 +61,7 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
return;
}
SIPRequest request = (SIPRequest) evt.getRequest();
logger.info("[收到心跳]， device: {}, callId: {}", device.getDeviceId(), request.getCallIdHeader().getCallId());
logger.info("[收到心跳] device: {}, callId: {}", device.getDeviceId(), request.getCallIdHeader().getCallId());
// 回复200 OK
try {
@@ -80,6 +80,11 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
device.setPort(remoteAddressInfo.getPort());
device.setHostAddress(remoteAddressInfo.getIp().concat(":").concat(String.valueOf(remoteAddressInfo.getPort())));
device.setIp(remoteAddressInfo.getIp());
// 设备地址变化会引起目录订阅任务失效，需要重新添加
if (device.getSubscribeCycleForCatalog() > 0) {
deviceService.removeCatalogSubscribe(device);
deviceService.addCatalogSubscribe(device);
}
}
if (device.getKeepaliveTime() == null) {
device.setKeepaliveIntervalTime(60);

31
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
View File

@@ -259,19 +259,22 @@ public class ZLMHttpHookListener {
} else {
result.setEnable_mp4(userSetting.isRecordPushLive());
}
// 替换流地址
if ("rtp".equals(param.getApp()) && !mediaInfo.isRtpEnable()) {
if (!mediaInfo.isRtpEnable()) {
String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));;
InviteInfo inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
// 国标流
if ("rtp".equals(param.getApp()) ) {
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
// 单端口模式下修改流 ID
if (!mediaInfo.isRtpEnable() && inviteInfo == null) {
String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));
inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
if (inviteInfo != null) {
result.setStream_replace(inviteInfo.getStream());
logger.info("[ZLM HOOK]推流鉴权 stream: {} 替换为 {}", param.getStream(), inviteInfo.getStream());
}
}
}
// 设置音频信息及录制信息
List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());
if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
@@ -292,10 +295,10 @@ public class ZLMHttpHookListener {
// 如果是录像下载就设置视频间隔十秒
if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {
// 获取录像的总时长，然后设置为这个视频的时长
InviteInfo inviteInfo = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, deviceId, channelId, param.getStream());
if (inviteInfo.getStreamInfo() != null ) {
String startTime = inviteInfo.getStreamInfo().getStartTime();
String endTime = inviteInfo.getStreamInfo().getEndTime();
InviteInfo inviteInfoForDownload = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, deviceId, channelId, param.getStream());
if (inviteInfoForDownload != null && inviteInfoForDownload.getStreamInfo() != null ) {
String startTime = inviteInfoForDownload.getStreamInfo().getStartTime();
String endTime = inviteInfoForDownload.getStreamInfo().getEndTime();
long difference = DateUtil.getDifference(startTime, endTime)/1000;
result.setMp4_max_second((int)difference);
result.setEnable_mp4(true);
@@ -529,11 +532,15 @@ public class ZLMHttpHookListener {
if (info != null) {
cmder.streamByeCmd(device, inviteInfo.getChannelId(),
inviteInfo.getStream(), null);
}else {
logger.info("[无人观看] 未找到设备的点播信息： {}， 流：{}", inviteInfo.getDeviceId(), param.getStream());
}
} catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) {
logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage());
}
}else {
logger.info("[无人观看] 未找到设备： {}，流：{}", inviteInfo.getDeviceId(), param.getStream());
}
inviteStreamService.removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(),
@@ -598,7 +605,7 @@ public class ZLMHttpHookListener {
String deviceId = s[0];
String channelId = s[1];
Device device = redisCatchStorage.getDevice(deviceId);
if (device == null) {
if (device == null || !device.isOnLine()) {
defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));
return defaultResult;
}

121
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/OnStreamChangedHookParam.java
View File

@@ -120,17 +120,17 @@ public class OnStreamChangedHookParam extends HookParam{
/**
* H264 = 0, H265 = 1, AAC = 2, G711A = 3, G711U = 4
*/
private int codecId;
private int codec_id;
/**
* 编码类型名称 CodecAAC CodecH264
*/
private String codecIdName;
private String codec_id_name;
/**
* Video = 0, Audio = 1
*/
private int codecType;
private int codec_type;
/**
* 轨道是否准备就绪
@@ -140,17 +140,17 @@ public class OnStreamChangedHookParam extends HookParam{
/**
* 音频采样位数
*/
private int sampleBit;
private int sample_bit;
/**
* 音频采样率
*/
private int sampleRate;
private int sample_rate;
/**
* 视频fps
*/
private int fps;
private float fps;
/**
* 视频高
@@ -162,6 +162,31 @@ public class OnStreamChangedHookParam extends HookParam{
*/
private int width;
/**
* 帧数
*/
private int frames;
/**
* 关键帧数
*/
private int key_frames;
/**
* GOP大小
*/
private int gop_size;
/**
* GOP间隔时长(ms)
*/
private int gop_interval_ms;
/**
* 丢帧率
*/
private float loss;
public int getChannels() {
return channels;
}
@@ -170,28 +195,28 @@ public class OnStreamChangedHookParam extends HookParam{
this.channels = channels;
}
public int getCodecId() {
return codecId;
public int getCodec_id() {
return codec_id;
}
public void setCodecId(int codecId) {
this.codecId = codecId;
public void setCodec_id(int codec_id) {
this.codec_id = codec_id;
}
public String getCodecIdName() {
return codecIdName;
public String getCodec_id_name() {
return codec_id_name;
}
public void setCodecIdName(String codecIdName) {
this.codecIdName = codecIdName;
public void setCodec_id_name(String codec_id_name) {
this.codec_id_name = codec_id_name;
}
public int getCodecType() {
return codecType;
public int getCodec_type() {
return codec_type;
}
public void setCodecType(int codecType) {
this.codecType = codecType;
public void setCodec_type(int codec_type) {
this.codec_type = codec_type;
}
public boolean isReady() {
@@ -202,27 +227,27 @@ public class OnStreamChangedHookParam extends HookParam{
this.ready = ready;
}
public int getSampleBit() {
return sampleBit;
public int getSample_bit() {
return sample_bit;
}
public void setSampleBit(int sampleBit) {
this.sampleBit = sampleBit;
public void setSample_bit(int sample_bit) {
this.sample_bit = sample_bit;
}
public int getSampleRate() {
return sampleRate;
public int getSample_rate() {
return sample_rate;
}
public void setSampleRate(int sampleRate) {
this.sampleRate = sampleRate;
public void setSample_rate(int sample_rate) {
this.sample_rate = sample_rate;
}
public int getFps() {
public float getFps() {
return fps;
}
public void setFps(int fps) {
public void setFps(float fps) {
this.fps = fps;
}
@@ -241,6 +266,46 @@ public class OnStreamChangedHookParam extends HookParam{
public void setWidth(int width) {
this.width = width;
}
public int getFrames() {
return frames;
}
public void setFrames(int frames) {
this.frames = frames;
}
public int getKey_frames() {
return key_frames;
}
public void setKey_frames(int key_frames) {
this.key_frames = key_frames;
}
public int getGop_size() {
return gop_size;
}
public void setGop_size(int gop_size) {
this.gop_size = gop_size;
}
public int getGop_interval_ms() {
return gop_interval_ms;
}
public void setGop_interval_ms(int gop_interval_ms) {
this.gop_interval_ms = gop_interval_ms;
}
public float getLoss() {
return loss;
}
public void setLoss(float loss) {
this.loss = loss;
}
}
public static class OriginSock{

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

@@ -243,6 +243,10 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
@Override
public void batchUpdateChannel(List<DeviceChannel> channels) {
String now = DateUtil.getNow();
for (DeviceChannel channel : channels) {
channel.setUpdateTime(now);
}
channelMapper.batchUpdate(channels);
for (DeviceChannel channel : channels) {
if (channel.getParentId() != null) {

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

@@ -217,7 +217,7 @@ public class DeviceServiceImpl implements IDeviceService {
for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
streamSession.remove(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
streamSession.removeByCallId(deviceId, ssrcTransaction.getChannelId(), ssrcTransaction.getCallId());
}
}
// 移除订阅

14
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
View File

@@ -163,14 +163,13 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (streamId == null) {
streamId = String.format("%08x", Long.parseLong(ssrc)).toUpperCase();
}
int ssrcCheckParam = 0;
if (ssrcCheck && tcpMode > 1) {
if (ssrcCheck && tcpMode > 0) {
// 目前zlm不支持 tcp模式更新ssrc，暂时关闭ssrc校验
logger.warn("[openRTPServer] TCP被动/TCP主动收流时，默认关闭ssrc检验");
logger.warn("[openRTPServer] 平台对接时下级可能自定义ssrc，但是tcp模式zlm收流目前无法更新ssrc，可能收流超时，此时请使用udp收流或者关闭ssrc校验");
}
int rtpServerPort;
if (mediaServerItem.isRtpEnable()) {
rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0) ? Long.parseLong(ssrc) : 0, port, reUsePort, tcpMode);
rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck ? Long.parseLong(ssrc) : 0, port, reUsePort, tcpMode);
} else {
rtpServerPort = mediaServerItem.getRtpProxyPort();
}
@@ -197,7 +196,10 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Override
public void closeRTPServer(String mediaServerId, String streamId) {
MediaServerItem mediaServerItem = this.getOne(mediaServerId);
closeRTPServer(mediaServerItem, streamId);
if (mediaServerItem.isRtpEnable()) {
closeRTPServer(mediaServerItem, streamId);
}
zlmresTfulUtils.closeStreams(mediaServerItem, "rtp", streamId);
}
@Override
@@ -558,7 +560,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
Map<String, Object> param = new HashMap<>();
param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
if (mediaServerItem.getRtspPort() != 0) {
param.put("ffmpeg.snap", "%s -rtsp_transport tcp -i %s -y -f mjpeg -t 0.001 %s");
param.put("ffmpeg.snap", "%s -rtsp_transport tcp -i %s -y -f mjpeg -frames:v 1 %s");
}
param.put("hook.enable","1");
param.put("hook.on_flow_report","");

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

@@ -233,6 +233,15 @@ public class PlayServiceImpl implements IPlayService {
HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
subscribe.removeSubscribe(hookSubscribe);
}
}else {
logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {},码流类型：{}，端口：{}, SSRC: {}",
device.getDeviceId(), channelId, device.isSwitchPrimarySubStream() ? "辅码流" : "主码流",
ssrcInfo.getPort(), ssrcInfo.getSsrc());
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
}
}, userSetting.getPlayTimeout());
@@ -263,6 +272,7 @@ public class PlayServiceImpl implements IPlayService {
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId,
timeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAY);
}, (event) -> {
logger.info("[点播失败] deviceId: {}, channelId:{}, {}: {}", device.getDeviceId(), channelId, event.statusCode, event.msg);
dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
// 释放ssrc
@@ -304,7 +314,13 @@ public class PlayServiceImpl implements IPlayService {
if (!device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
return;
}
String substring = contentString.substring(0, contentString.indexOf("y="));
String substring;
if (contentString.indexOf("y=") > 0) {
substring = contentString.substring(0, contentString.indexOf("y="));
}else {
substring = contentString;
}
try {
SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
int port = -1;
@@ -394,7 +410,7 @@ public class PlayServiceImpl implements IPlayService {
deviceChannel.setStreamId(streamInfo.getStream());
storager.startPlay(deviceId, channelId, streamInfo.getStream());
}
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAYBACK, deviceId, channelId);
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(InviteSessionType.PLAYBACK, ((OnStreamChangedHookParam) param).getStream());
if (inviteInfo != null) {
inviteInfo.setStatus(InviteSessionStatus.ok);
@@ -537,7 +553,6 @@ public class PlayServiceImpl implements IPlayService {
// 处理收到200ok后的TCP主动连接以及SSRC不一致的问题
InviteOKHandler(eventResult, ssrcInfo, mediaServerItem, device, channelId,
playBackTimeOutTaskKey, callback, inviteInfo, InviteSessionType.PLAYBACK);
}, errorEvent);
} catch (InvalidArgumentException | SipException | ParseException e) {
logger.error("[命令发送失败] 录像回放: {}", e.getMessage());
@@ -558,6 +573,10 @@ public class PlayServiceImpl implements IPlayService {
ResponseEvent responseEvent = (ResponseEvent) eventResult.event;
String contentString = new String(responseEvent.getResponse().getRawContent());
String ssrcInResponse = SipUtils.getSsrcFromSdp(contentString);
// 兼容回复的消息中缺少ssrc(y字段)的情况
if (ssrcInResponse == null) {
ssrcInResponse = ssrcInfo.getSsrc();
}
if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
// ssrc 一致
if (mediaServerItem.isRtpEnable()) {
@@ -636,6 +655,7 @@ public class PlayServiceImpl implements IPlayService {
@Override
public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, ErrorCallback<Object> callback) {
Device device = storager.queryVideoDevice(deviceId);

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

@@ -413,6 +413,8 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
logger.info("启用代理失败： {}/{}->{}({})", app, stream, jsonObject.getString("msg"),
streamProxy.getSrcUrl() == null? streamProxy.getUrl():streamProxy.getSrcUrl());
}
} else if (streamProxy != null && streamProxy.isEnable()) {
return true ;
}
return result;
}

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

@@ -117,6 +117,6 @@ public interface PlatformChannelMapper {
"where dc.channel_id = #{channelId} and pgc.platform_id=#{platformId}")
List<Device> queryDeviceInfoByPlatformIdAndChannelId(@Param("platformId") String platformId, @Param("channelId") String channelId);
@Select("SELECT pgc.platform_id from wvp_platform_gb_channel pgc left join wvp_device_channel dc on dc.id = pgc.device_channel_id WHERE dc.channel_id='${channelId}'")
List<String> queryParentPlatformByChannelId(String channelId);
@Select("SELECT pgc.platform_id from wvp_platform_gb_channel pgc left join wvp_device_channel dc on dc.id = pgc.device_channel_id WHERE dc.channel_id=#{channelId}")
List<String> queryParentPlatformByChannelId(@Param("channelId") String channelId);
}

37
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/SseController/SseController.java
View File

@@ -1,37 +0,0 @@
package com.genersoft.iot.vmp.vmanager.gb28181.SseController;
import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEventListener;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
/**
* @description: SSE推送
* @author: lawrencehj
* @data: 2021-01-20
*/
@Tag(name = "SSE推送")
@Controller
@RequestMapping("/api")
public class SseController {
@Autowired
AlarmEventListener alarmEventListener;
@GetMapping("/emit")
public SseEmitter emit(@RequestParam String browserId) {
final SseEmitter sseEmitter = new SseEmitter(0L);
try {
alarmEventListener.addSseEmitters(browserId, sseEmitter);
}catch (Exception e){
sseEmitter.completeWithError(e);
}
return sseEmitter;
}
}

55
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/sse/SseController.java Normal file
View File

@@ -0,0 +1,55 @@
package com.genersoft.iot.vmp.vmanager.gb28181.sse;
import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEventListener;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;
/**
* SSE 推送.
*
* @author lawrencehj
* @author <a href="mailto:xiaoQQya@126.com">xiaoQQya</a>
* @since 2021/01/20
*/
@Tag(name = "SSE 推送")
@RestController
@RequestMapping("/api")
public class SseController {
@Resource
private AlarmEventListener alarmEventListener;
/**
* SSE 推送.
*
* @param response 响应
* @param browserId 浏览器ID
* @throws IOException IOEXCEPTION
* @author <a href="mailto:xiaoQQya@126.com">xiaoQQya</a>
* @since 2023/11/06
*/
@GetMapping("/emit")
public void emit(HttpServletResponse response, @RequestParam String browserId) throws IOException, InterruptedException {
response.setContentType("text/event-stream");
response.setCharacterEncoding("utf-8");
PrintWriter writer = response.getWriter();
alarmEventListener.addSseEmitter(browserId, writer);
while (!writer.checkError()) {
Thread.sleep(1000);
writer.write(":keep alive\n\n");
writer.flush();
}
alarmEventListener.removeSseEmitter(browserId, writer);
}
}