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

Merge remote-tracking branch 'origin/master' into 重构/1078

Browse Source 
# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/gb28181/dao/CommonGBChannelMapper.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/dao/PlatformChannelMapper.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/dao/provider/ChannelProvider.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceChannelServiceImpl.java
#	数据库/2.7.3/初始化-mysql-2.7.3.sql
#	数据库/2.7.3/初始化-postgresql-kingbase-2.7.3.sql
This commit is contained in:
lin
2025-03-18 15:30:35 +08:00
parent 51f126e571 4af1e3859a
commit 0329c8f5d1
231 changed files with 15701 additions and 4618 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/com/genersoft/iot/vmp/conf/SipConfig.java
View File

@@ -6,6 +6,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
@ConfigurationProperties(prefix = "sip", ignoreInvalidFields = true)
@Order(0)
@@ -16,6 +18,8 @@ public class SipConfig {
private String showIp;
private List<String> monitorIps;
private Integer port;
private String domain;
@@ -30,5 +34,5 @@ public class SipConfig {
private boolean alarm = false;
private long timeout = 15;
private long timeout = 150;
}

5
src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java
View File

@@ -31,10 +31,13 @@ public class SipPlatformRunner implements CommandLineRunner {
@Autowired
private ISIPCommanderForPlatform sipCommanderForPlatform;
@Autowired
private UserSetting userSetting;
@Override
public void run(String... args) throws Exception {
// 获取所有启用的平台
List<Platform> parentPlatforms = platformService.queryEnablePlatformList();
List<Platform> parentPlatforms = platformService.queryEnablePlatformList(userSetting.getServerId());
for (Platform platform : parentPlatforms) {

16
src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java
View File

@@ -37,6 +37,11 @@ public class UserSetting {
*/
private Integer playTimeout = 10000;
/**
* 获取设备录像数据超时时间,单位：毫秒
*/
private Integer recordInfoTimeout = 15000;
/**
* 上级点播等待超时时间,单位：毫秒
*/
@@ -175,4 +180,15 @@ public class UserSetting {
*/
private long loginTimeout = 30;
/**
* jwk文件路径，若不指定则使用resources目录下的jwk.json
*/
private String jwkFile = "classpath:jwk.json";
/**
* wvp集群模式下如果注册向上级的wvp奔溃，则自动选择一个其他wvp继续注册到上级
*/
private boolean autoRegisterPlatform = false;
}

11
src/main/java/com/genersoft/iot/vmp/conf/WVPTimerTask.java
View File

@@ -1,12 +1,14 @@
package com.genersoft.iot.vmp.conf;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.ServerInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
@Component
public class WVPTimerTask {
@@ -19,11 +21,8 @@ public class WVPTimerTask {
@Autowired
private SipConfig sipConfig;
@Scheduled(fixedDelay = 2 * 1000) //每3秒执行一次
@Scheduled(fixedDelay = 2, timeUnit = TimeUnit.SECONDS) //每3秒执行一次
public void execute(){
JSONObject jsonObject = new JSONObject();
jsonObject.put("ip", sipConfig.getShowIp());
jsonObject.put("port", serverPort);
redisCatchStorage.updateWVPInfo(jsonObject, 3);
redisCatchStorage.updateWVPInfo(ServerInfo.create(sipConfig.getShowIp(), serverPort), 3);
}
}

91
src/main/java/com/genersoft/iot/vmp/conf/redis/RedisRpcConfig.java
View File

@@ -3,10 +3,12 @@ package com.genersoft.iot.vmp.conf.redis;
import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcClassHandler;
import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcMessage;
import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcRequest;
import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcResponse;
import com.genersoft.iot.vmp.service.redisMsg.control.RedisRpcController;
import com.genersoft.iot.vmp.service.redisMsg.dto.RpcController;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -16,8 +18,10 @@ import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import javax.sip.message.Response;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
@@ -36,9 +40,6 @@ public class RedisRpcConfig implements MessageListener {
@Autowired
private UserSetting userSetting;
@Autowired
private RedisRpcController redisRpcController;
@Autowired
private RedisTemplate<Object, Object> redisTemplate;
@@ -48,6 +49,40 @@ public class RedisRpcConfig implements MessageListener {
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
private final static Map<String, RedisRpcClassHandler> protocolHash = new HashMap<>();
public void addHandler(String path, RedisRpcClassHandler handler) {
protocolHash.put(path, handler);
}
// @Override
// public void run(String... args) throws Exception {
// List<Class<?>> classList = ClassUtil.getClassList("com.genersoft.iot.vmp.service.redisMsg.control", RedisRpcController.class);
// for (Class<?> handlerClass : classList) {
// String controllerPath = handlerClass.getAnnotation(RedisRpcController.class).value();
// Object bean = ClassUtil.getBean(controllerPath, handlerClass);
// // 扫描其下的方法
// Method[] methods = handlerClass.getDeclaredMethods();
// for (Method method : methods) {
// RedisRpcMapping annotation = method.getAnnotation(RedisRpcMapping.class);
// if (annotation != null) {
// String methodPath = annotation.value();
// if (methodPath != null) {
// protocolHash.put(controllerPath + "/" + methodPath, new RedisRpcClassHandler(bean, method));
// }
// }
//
// }
//
// }
// for (String s : protocolHash.keySet()) {
// System.out.println(s);
// }
// if (log.isDebugEnabled()) {
// log.debug("消息ID缓存表 protocolHash:{}", protocolHash);
// }
// }
@Override
public void onMessage(Message message, byte[] pattern) {
boolean isEmpty = taskQueue.isEmpty();
@@ -63,10 +98,10 @@ public class RedisRpcConfig implements MessageListener {
} else if (redisRpcMessage.getResponse() != null){
handlerResponse(redisRpcMessage.getResponse());
} else {
log.error("[redis rpc 解析失败] {}", JSON.toJSONString(redisRpcMessage));
log.error("[redis-rpc]解析失败 {}", JSON.toJSONString(redisRpcMessage));
}
} catch (Exception e) {
log.error("[redis rpc 解析异常] ", e);
log.error("[redis-rpc]解析异常 {}",new String(msg.getBody()), e);
}
}
});
@@ -87,17 +122,23 @@ public class RedisRpcConfig implements MessageListener {
return;
}
log.info("[redis-rpc] << {}", request);
Method method = getMethod(request.getUri());
RedisRpcClassHandler redisRpcClassHandler = protocolHash.get(request.getUri());
if (redisRpcClassHandler == null) {
log.error("[redis-rpc] 路径: {}不存在", request.getUri());
return;
}
RpcController controller = redisRpcClassHandler.getController();
Method method = redisRpcClassHandler.getMethod();
// 没有携带目标ID的可以理解为哪个wvp有结果就哪个回复，携带目标ID，但是如果是不存在的uri则直接回复404
if (userSetting.getServerId().equals(request.getToId())) {
if (method == null) {
// 回复404结果
RedisRpcResponse response = request.getResponse();
response.setStatusCode(404);
response.setStatusCode(ErrorCode.ERROR404.getCode());
sendResponse(response);
return;
}
RedisRpcResponse response = (RedisRpcResponse)method.invoke(redisRpcController, request);
RedisRpcResponse response = (RedisRpcResponse)method.invoke(controller, request);
if(response != null) {
sendResponse(response);
}
@@ -105,26 +146,14 @@ public class RedisRpcConfig implements MessageListener {
if (method == null) {
return;
}
RedisRpcResponse response = (RedisRpcResponse)method.invoke(redisRpcController, request);
RedisRpcResponse response = (RedisRpcResponse)method.invoke(controller, request);
if (response != null) {
sendResponse(response);
}
}
}catch (InvocationTargetException | IllegalAccessException e) {
log.error("[redis rpc ] 处理请求失败 ", e);
log.error("[redis-rpc ] 处理请求失败 ", e);
}
}
private Method getMethod(String name) {
// 启动后扫描所有的路径注解
Method[] methods = redisRpcController.getClass().getMethods();
for (Method method : methods) {
if (method.getName().equals(name)) {
return method;
}
}
return null;
}
private void sendResponse(RedisRpcResponse response){
@@ -142,23 +171,28 @@ public class RedisRpcConfig implements MessageListener {
redisTemplate.convertAndSend(REDIS_REQUEST_CHANNEL_KEY, message);
}
private final Map<Long, SynchronousQueue<RedisRpcResponse>> topicSubscribers = new ConcurrentHashMap<>();
private final Map<Long, CommonCallback<RedisRpcResponse>> callbacks = new ConcurrentHashMap<>();
public RedisRpcResponse request(RedisRpcRequest request, int timeOut) {
public RedisRpcResponse request(RedisRpcRequest request, long timeOut) {
return request(request, timeOut, TimeUnit.SECONDS);
}
public RedisRpcResponse request(RedisRpcRequest request, long timeOut, TimeUnit timeUnit) {
request.setSn((long) random.nextInt(1000) + 1);
SynchronousQueue<RedisRpcResponse> subscribe = subscribe(request.getSn());
try {
sendRequest(request);
return subscribe.poll(timeOut, TimeUnit.SECONDS);
return subscribe.poll(timeOut, timeUnit);
} catch (InterruptedException e) {
log.warn("[redis rpc timeout] uri: {}, sn: {}", request.getUri(), request.getSn(), e);
RedisRpcResponse redisRpcResponse = new RedisRpcResponse();
redisRpcResponse.setStatusCode(ErrorCode.ERROR486.getCode());
return redisRpcResponse;
} finally {
this.unsubscribe(request.getSn());
}
return null;
}
public void request(RedisRpcRequest request, CommonCallback<RedisRpcResponse> callback) {
@@ -209,6 +243,9 @@ public class RedisRpcConfig implements MessageListener {
return callbacks.size();
}
// @Scheduled(fixedRate = 1000) //每1秒执行一次
// public void execute(){
// logger.info("callbacks的长度: " + callbacks.size());

18
src/main/java/com/genersoft/iot/vmp/conf/redis/bean/RedisRpcClassHandler.java Normal file
View File

@@ -0,0 +1,18 @@
package com.genersoft.iot.vmp.conf.redis.bean;
import com.genersoft.iot.vmp.service.redisMsg.dto.RpcController;
import lombok.Data;
import java.lang.reflect.Method;
@Data
public class RedisRpcClassHandler {
private RpcController controller;
private Method method;
public RedisRpcClassHandler(RpcController controller, Method method) {
this.controller = controller;
this.method = method;
}
}

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

@@ -21,12 +21,16 @@ import org.jose4j.jwt.consumer.JwtConsumer;
import org.jose4j.jwt.consumer.JwtConsumerBuilder;
import org.jose4j.lang.JoseException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.File;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.List;
@@ -92,8 +96,46 @@ public class JwtUtils implements InitializingBean {
*/
private RsaJsonWebKey generateRsaJsonWebKey() throws JoseException {
RsaJsonWebKey rsaJsonWebKey = null;
try (BufferedReader reader = new BufferedReader(new InputStreamReader(getClass().getClassLoader().getResourceAsStream("/jwk.json"), StandardCharsets.UTF_8))) {
String jwkJson = reader.readLine();
try {
String jwkFile = userSetting.getJwkFile();
InputStream inputStream = null;
if (jwkFile.startsWith("classpath:")){
String filePath = jwkFile.substring("classpath:".length());
ClassPathResource civilCodeFile = new ClassPathResource(filePath);
if (civilCodeFile.exists()) {
inputStream = civilCodeFile.getInputStream();
}
}else {
File civilCodeFile = new File(userSetting.getCivilCodeFile());
if (civilCodeFile.exists()) {
inputStream = Files.newInputStream(civilCodeFile.toPath());
}
}
if (inputStream == null ) {
log.warn("[API AUTH] 读取jwk.json失败，文件不存在，将使用新生成的随机RSA密钥对");
// 生成一个RSA密钥对，该密钥对将用于JWT的签名和验证，包装在JWK中
rsaJsonWebKey = RsaJwkGenerator.generateJwk(2048);
// 给JWK一个密钥ID
rsaJsonWebKey.setKeyId(keyId);
return rsaJsonWebKey;
}
BufferedReader inputStreamReader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
int index = -1;
String line;
StringBuilder content = new StringBuilder();
while ((line = inputStreamReader.readLine()) != null) {
content.append(line);
index ++;
if (index == 0) {
continue;
}
}
inputStreamReader.close();
inputStream.close();
String jwkJson = content.toString();
JsonWebKeySet jsonWebKeySet = new JsonWebKeySet(jwkJson);
List<JsonWebKey> jsonWebKeys = jsonWebKeySet.getJsonWebKeys();
if (!jsonWebKeys.isEmpty()) {
@@ -102,14 +144,15 @@ public class JwtUtils implements InitializingBean {
rsaJsonWebKey = (RsaJsonWebKey) jsonWebKey;
}
}
} catch (Exception e) {
// ignored
}
} catch (Exception ignore) {}
if (rsaJsonWebKey == null) {
log.warn("[API AUTH] 读取jwk.json失败，获取内容失败，将使用新生成的随机RSA密钥对");
// 生成一个RSA密钥对，该密钥对将用于JWT的签名和验证，包装在JWK中
rsaJsonWebKey = RsaJwkGenerator.generateJwk(2048);
// 给JWK一个密钥ID
rsaJsonWebKey.setKeyId(keyId);
}else {
log.info("[API AUTH] 读取jwk.json成功");
}
return rsaJsonWebKey;
}

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

@@ -148,8 +148,10 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
corsConfiguration.setAllowCredentials(true);
corsConfiguration.setAllowedOrigins(userSetting.getAllowedOrigins());
}else {
corsConfiguration.setAllowCredentials(false);
corsConfiguration.setAllowedOrigins(Collections.singletonList(CorsConfiguration.ALL));
// 在SpringBoot 2.4及以上版本处理跨域时，遇到错误提示：当allowCredentials为true时，allowedOrigins不能包含特殊值"*"。
// 解决方法是明确指定allowedOrigins或使用allowedOriginPatterns。
corsConfiguration.setAllowCredentials(true);
corsConfiguration.addAllowedOriginPattern(CorsConfiguration.ALL); // 默认全部允许所有跨域
}
corsConfiguration.setExposedHeaders(Arrays.asList(JwtUtils.getHeader()));