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

修复集群模式下国标级联跨平台请求无法发送目录订阅的BUG

Browse Source
This commit is contained in:
lin
2025-11-20 11:34:55 +08:00
parent a97a1113cf
commit 57524b365a
10 changed files with 355 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/main/java/com/genersoft/iot/vmp/gb28181/controller/PlatformController.java
View File

@@ -23,7 +23,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;
/**
* 级联平台管理
@@ -157,15 +156,17 @@ public class PlatformController {
@Parameter(name = "id", description = "上级平台ID")
@DeleteMapping("/delete")
@ResponseBody
public DeferredResult<WVPResult<?>> deletePlatform(Integer id) {
public WVPResult<?> deletePlatform(Integer id) {
if (log.isDebugEnabled()) {
log.debug("删除上级平台API调用");
}
DeferredResult<WVPResult<?>> deferredResult = new DeferredResult<>();
platformService.delete(id, (object)-> deferredResult.setResult(WVPResult.success()));
return deferredResult;
boolean result = platformService.delete(id);
if (result) {
return WVPResult.success();
}else {
return WVPResult.fail(ErrorCode.ERROR100);
}
}
@Operation(summary = "查询上级平台是否存在", security = @SecurityRequirement(name = JwtUtils.HEADER))

18
src/main/java/com/genersoft/iot/vmp/gb28181/controller/bean/ChannelListForRpcParam.java Normal file
View File

@@ -0,0 +1,18 @@
package com.genersoft.iot.vmp.gb28181.controller.bean;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import java.util.List;
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class ChannelListForRpcParam {
private List<Integer> channelIds;
private Integer platformId;
}

3
src/main/java/com/genersoft/iot/vmp/gb28181/service/IPlatformService.java
View File

@@ -1,6 +1,5 @@
package com.genersoft.iot.vmp.gb28181.service;
import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
import com.genersoft.iot.vmp.gb28181.bean.Platform;
import com.genersoft.iot.vmp.gb28181.bean.SipTransactionInfo;
@@ -78,7 +77,7 @@ public interface IPlatformService {
List<Platform> queryEnablePlatformList(String serverId);
void delete(Integer platformId, CommonCallback<Object> callback);
boolean delete(Integer platformId);
List<Platform> queryAll(String serverId);

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

@@ -3,12 +3,14 @@ package com.genersoft.iot.vmp.gb28181.service.impl;
import com.genersoft.iot.vmp.common.enums.ChannelDataType;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.controller.bean.ChannelListForRpcParam;
import com.genersoft.iot.vmp.gb28181.dao.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.channel.ChannelEvent;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.service.IPlatformChannelService;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import lombok.extern.slf4j.Slf4j;
@@ -59,6 +61,9 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
@Autowired
private UserSetting userSetting;
@Autowired
private IRedisRpcService redisRpcService;
// 监听通道信息变化
@EventListener
public void onApplicationEvent(ChannelEvent event) {
@@ -66,12 +71,12 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
log.info("[国标级联-处理通道变化事件] 通道数量为空");
return;
}
String deviceIds = event.getChannels().stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining());
String deviceIds = event.getChannels().stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[国标级联-处理通道变化事件] 类型： {}, 通道: {}", event.getMessageType(), deviceIds);
// 获取通道所关联的平台
List<Platform> allPlatform = platformMapper.queryByServerId(userSetting.getServerId());
if (allPlatform.isEmpty()) {
log.info("[国标级联-处理通道变化事件] 没有关联的平台");
log.info("[国标级联-处理通道变化事件] 没有当前服务负责的平台");
return;
}
// 获取所用订阅
@@ -157,7 +162,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
@EventListener
public void onApplicationEvent(CatalogEvent event) {
String deviceIds = event.getChannels().stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining());
String deviceIds = event.getChannels().stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[Catalog事件: {}] 通道： {}", event.getType(), deviceIds);
Platform platform = event.getPlatform();
if (platform == null || platform.getServerGBId() == null) {
@@ -377,8 +382,20 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
@Transactional
public int addChannelList(Integer platformId, List<CommonGBChannel> channelList) {
Platform platform = platformMapper.query(platformId);
if (platform == null) {
return 0;
Assert.notNull(platform, "平台不存在");
String channelDeviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[共享通道] 平台：{}， 通道：{}", platform.getServerGBId(), channelDeviceIds);
if (!userSetting.getServerId().equals(platform.getServerId())) {
List<Integer> channelIdList = channelList.stream().map(CommonGBChannel::getGbId).toList();
int result = redisRpcService.addPlatformChannelList(platform.getServerId(), new ChannelListForRpcParam(channelIdList, platformId));
if (result > 0) {
log.info("[跨平台-共享通道] 成功， 平台：{}， 通道：{}", platform.getServerGBId(), channelDeviceIds);
}else {
log.info("[跨平台-共享通道] 失败， 平台：{}， 通道：{}", platform.getServerGBId(), channelDeviceIds);
}
return result;
}
int result = platformChannelMapper.addChannels(platformId, channelList);
if (result > 0) {
@@ -422,7 +439,17 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
if (platform == null) {
return 0;
}
log.info("[取消共享通道] 平台：{}， 通道：全部", platform.getServerGBId());
if (!userSetting.getServerId().equals(platform.getServerId())) {
int result = redisRpcService.removeAllPlatformChannel(platform.getServerId(), platformId);
if (result > 0) {
log.info("[跨平台-取消共享通道] 成功， 平台：{}， 通道：全部", platform.getServerGBId());
}else {
log.info("[跨平台-取消共享通道] 失败， 平台：{}， 通道：全部", platform.getServerGBId());
}
return result;
}
List<CommonGBChannel> channelListShare = platformChannelMapper.queryShare(platformId, null);
Assert.notEmpty(channelListShare, "未共享任何通道");
int result = platformChannelMapper.removeChannelsWithPlatform(platformId, channelListShare);
@@ -476,11 +503,22 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
log.info("[移除关联通道] 平台{}未查询到", platformId);
return 0;
}
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining());
log.info("[移除关联通道] 上级平台： {}， 通道： {}", platform.getServerGBId(), deviceIds);
String channelDeviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[取消共享通道] 平台：{}， 通道： {}", platform.getServerGBId(), channelDeviceIds);
if (!userSetting.getServerId().equals(platform.getServerId())) {
List<Integer> channelIds = channelList.stream().map(CommonGBChannel::getGbId).toList();
int result = redisRpcService.removePlatformChannelList(platform.getServerId(), new ChannelListForRpcParam(channelIds, platformId));
if (result > 0) {
log.info("[跨平台-取消共享通道] 成功， 平台：{}， 通道： {}", platform.getServerGBId(), channelDeviceIds);
}else {
log.info("[跨平台-取消共享通道] 失败， 平台：{}， 通道： {}", platform.getServerGBId(), channelDeviceIds);
}
return result;
}
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
int result = platformChannelMapper.removeChannelsWithPlatform(platformId, channelList);
if (result <= 0) {
log.info("[移除关联通道] 平台{}未关联通道： {}", platformId, deviceIds);
log.info("[取消共享通道] 平台{}未关联通道： {}", platformId, deviceIds);
return 0;
}
// 查询通道相关的分组信息
@@ -505,7 +543,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
// 发送catalog
eventPublisher.catalogEventPublish(platform, channelList, CatalogEvent.DEL);
} catch (Exception e) {
log.warn("[移除关联通道] 发送失败，数量：{}", channelList.size(), e);
log.warn("[取消共享通道] 发送失败，数量：{}", channelList.size(), e);
}
return result;
}
@@ -590,6 +628,16 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
public void pushChannel(Integer platformId) {
Platform platform = platformMapper.query(platformId);
Assert.notNull(platform, "平台不存在");
if (!userSetting.getServerId().equals(platform.getServerId())) {
boolean result = redisRpcService.pushPlatformChannel(platform.getServerId(), platformId);
if (result) {
log.info("[跨平台-主动推送通道] 成功， 平台：{}", platform.getServerGBId());
}else {
log.info("[跨平台-主动推送通道] 失败， 平台：{}", platform.getServerGBId());
}
return;
}
List<CommonGBChannel> channelList = queryByPlatform(platform);
if (channelList.isEmpty()){
log.info("[推送通道] 平台：{} 未查询到通道信息", platform.getServerGBId());
@@ -607,15 +655,28 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
@Override
public void updateCustomChannel(PlatformChannel channel) {
platformChannelMapper.updateCustomChannel(channel);
Platform platform = platformMapper.query(channel.getPlatformId());
Assert.notNull(platform, "平台不存在");
log.info("[国标级联-自定义共享通道] 平台：{}， 通道：{}", platform.getServerGBId(), channel);
if (!userSetting.getServerId().equals(platform.getServerId())) {
boolean result = redisRpcService.updateCustomPlatformChannel(platform.getServerId(), channel);
if (result) {
log.info("[国标级联-自定义共享通道] 成功， 平台：{}， 通道：{}", platform.getServerGBId(), channel);
}else {
log.info("[国标级联-自定义共享通道] 失败， 平台：{}， 通道：{}", platform.getServerGBId(), channel);
}
return;
}
platformChannelMapper.updateCustomChannel(channel);
CommonGBChannel commonGBChannel = platformChannelMapper.queryShareChannel(channel.getPlatformId(), channel.getGbId());
// 发送消息
try {
// 发送catalog
eventPublisher.catalogEventPublish(platform, commonGBChannel, CatalogEvent.UPDATE);
} catch (Exception e) {
log.warn("[自定义通道信息] 发送失败， 平台ID： {}， 通道： {}（{}）", channel.getPlatformId(),
log.warn("[国标级联-自定义共享通道] 发送失败， 平台ID： {}， 通道： {}（{}）", channel.getPlatformId(),
channel.getGbName(), channel.getId(), e);
}
}
@@ -631,7 +692,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
// 获取关联这些通道的平台
List<Platform> platformList = platformChannelMapper.queryPlatFormListByChannelList(channelIds);
if (platformList.isEmpty()) {
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining());
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[获取关联这些通道的平台] 未查询到通道关联的平台, 通道如下 {}", deviceIds);
return;
}
@@ -671,7 +732,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
// 获取关联这些通道的平台
List<Platform> platformList = platformChannelMapper.queryPlatFormListByChannelList(channelIds);
if (platformList.isEmpty()) {
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining());
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[获取关联这些通道的平台] 未查询到通道关联的平台, 通道如下 {}", deviceIds);
return;
}
@@ -710,7 +771,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
});
List<Platform> platformList = platformChannelMapper.queryPlatFormListByChannelList(channelIds);
if (platformList.isEmpty()) {
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining());
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[获取关联这些通道的平台] 未查询到通道关联的平台, 通道如下 {}", deviceIds);
return;
}
@@ -743,7 +804,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
});
List<Platform> platformList = platformChannelMapper.queryPlatFormListByChannelList(channelIds);
if (platformList.isEmpty()) {
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining());
String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[获取关联这些通道的平台] 未查询到通道关联的平台, 通道如下 {}", deviceIds);
return;
}

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

@@ -4,6 +4,7 @@ import com.genersoft.iot.vmp.common.*;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.dao.PlatformChannelMapper;
@@ -31,6 +32,7 @@ import com.genersoft.iot.vmp.service.ISendRtpServerService;
import com.genersoft.iot.vmp.service.bean.*;
import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import gov.nist.javax.sip.message.SIPResponse;
@@ -130,7 +132,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
if (stopResult) {
Platform platform = queryPlatformByServerGBId(sendRtpItem.getTargetId());
if (platform != null) {
if (platform != null && userSetting.getServerId().equals(platform.getServerId())) {
try {
commanderForPlatform.streamByeCmd(platform, sendRtpItem, channel);
} catch (InvalidArgumentException | ParseException | SipException e) {
@@ -154,7 +156,9 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
statusTaskRunner.removeRegisterTask(taskInfo.getPlatformServerId());
continue;
}
sendUnRegister(platform, taskInfo.getSipTransactionInfo());
if (userSetting.getServerId().equals(platform.getServerId())) {
sendUnRegister(platform, taskInfo.getSipTransactionInfo());
}
}
// 启动时所有平台默认离线
platformMapper.offlineAll(userSetting.getServerId());
@@ -874,25 +878,38 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
@Override
@Transactional
public void delete(Integer platformId, CommonCallback<Object> callback) {
public boolean delete(Integer platformId) {
Platform platform = platformMapper.query(platformId);
Assert.notNull(platform, "平台不存在");
if (statusTaskRunner.containsRegister(platform.getServerGBId())) {
try {
SipTransactionInfo transactionInfo = statusTaskRunner.getRegisterTransactionInfo(platform.getServerGBId());
sendUnRegister(platform, transactionInfo);
}catch (Exception ignored) {}
log.info("[删除平台] {}/{} {}:{}", platform.getName(), platform.getServerGBId(), platform.getServerIp(), platform.getServerPort());
if (!userSetting.getServerId().equals(platform.getServerId())) {
boolean result = redisRpcService.deletePlatform(platform.getServerId(), platformId);
if (result) {
log.info("[删除平台] 跨平台删除成功 {}/{}", platform.getName(), platform.getServerGBId());
}else {
log.info("[删除平台] 跨平台删除失败 {}/{}", platform.getName(), platform.getServerGBId());
}
return result;
}
platformMapper.delete(platform.getId());
try {
if (statusTaskRunner.containsRegister(platform.getServerGBId())) {
try {
SipTransactionInfo transactionInfo = statusTaskRunner.getRegisterTransactionInfo(platform.getServerGBId());
sendUnRegister(platform, transactionInfo);
}catch (Exception ignored) {}
}
platformMapper.delete(platform.getId());
statusTaskRunner.removeRegisterTask(platform.getServerGBId());
statusTaskRunner.removeKeepAliveTask(platform.getServerGBId());
statusTaskRunner.removeRegisterTask(platform.getServerGBId());
statusTaskRunner.removeKeepAliveTask(platform.getServerGBId());
subscribeHolder.removeCatalogSubscribe(platform.getServerGBId());
subscribeHolder.removeMobilePositionSubscribe(platform.getServerGBId());
if (callback != null) {
callback.run(true);
subscribeHolder.removeCatalogSubscribe(platform.getServerGBId());
subscribeHolder.removeMobilePositionSubscribe(platform.getServerGBId());
}catch (Exception e) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), e.getMessage());
}
return true;
}
@Override

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderForPlatform.java
View File

@@ -44,6 +44,7 @@ import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@Slf4j
@Component
@@ -434,7 +435,8 @@ public class SIPCommanderForPlatform implements ISIPCommanderForPlatform {
Integer finalIndex = index;
String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channels,
deviceChannels.size(), type, subscribeInfo);
log.info("[发送NOTIFY通知]类型： {}，发送数量： {}", type, channels.size());
String channelDeviceIds = channels.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(","));
log.info("[发送NOTIFY通知]类型： {}，通道： {}", type, channelDeviceIds);
sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
log.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
log.error(catalogXmlContent);

14
src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java
View File

@@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.service.redisMsg;
import com.genersoft.iot.vmp.common.CommonCallback;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.controller.bean.ChannelListForRpcParam;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@@ -25,6 +26,7 @@ public interface IRedisRpcService {
void removeCallback(long key);
long onStreamOnlineEvent(String app, String stream, CommonCallback<StreamInfo> callback);
void unPushStreamOnlineEvent(String app, String stream);
void subscribeCatalog(int id, int cycle);
@@ -33,6 +35,18 @@ public interface IRedisRpcService {
boolean updatePlatform(String serverId, Platform platform);
boolean deletePlatform(String serverId, Integer platformId);
int addPlatformChannelList(String serverGBId, ChannelListForRpcParam channelListForRpcParam);
int removeAllPlatformChannel(String serverId, Integer platformId);
int removePlatformChannelList(String serverId, ChannelListForRpcParam channelListForRpcParam);
boolean updateCustomPlatformChannel(String serverId, PlatformChannel channel);
boolean pushPlatformChannel(String serverId, Integer platformId);
void catalogEventPublish(String serverId, CatalogEvent catalogEvent);
WVPResult<SyncStatus> devicesSync(String serverId, String deviceId);

112
src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcPlatformController.java
View File

@@ -8,6 +8,8 @@ import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcRequest;
import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcResponse;
import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
import com.genersoft.iot.vmp.gb28181.bean.Platform;
import com.genersoft.iot.vmp.gb28181.bean.PlatformChannel;
import com.genersoft.iot.vmp.gb28181.controller.bean.ChannelListForRpcParam;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.service.IPlatformChannelService;
import com.genersoft.iot.vmp.gb28181.service.IPlatformService;
@@ -55,7 +57,7 @@ public class RedisRpcPlatformController extends RpcController {
* 更新
*/
@RedisRpcMapping("update")
public RedisRpcResponse play(RedisRpcRequest request) {
public RedisRpcResponse update(RedisRpcRequest request) {
Platform platform = JSONObject.parseObject(request.getParam().toString(), Platform.class);
RedisRpcResponse response = request.getResponse();
boolean update = platformService.update(platform);
@@ -64,6 +66,114 @@ public class RedisRpcPlatformController extends RpcController {
return response;
}
/**
* 删除
*/
@RedisRpcMapping("delete")
public RedisRpcResponse delete(RedisRpcRequest request) {
Integer platformId = Integer.parseInt(request.getParam().toString());
RedisRpcResponse response = request.getResponse();
try {
boolean result = platformService.delete(platformId);
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody(Boolean.toString(result));
}catch (Exception e) {
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody("false");
}
return response;
}
/**
* 主动推送通道
*/
@RedisRpcMapping("pushChannel")
public RedisRpcResponse pushChannel(RedisRpcRequest request) {
Integer platformId = Integer.parseInt(request.getParam().toString());
RedisRpcResponse response = request.getResponse();
try {
platformChannelService.pushChannel(platformId);
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody("true");
}catch (Exception e) {
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody("false");
}
return response;
}
/**
* 共享通道
*/
@RedisRpcMapping("addChannelList")
public RedisRpcResponse addChannelList(RedisRpcRequest request) {
ChannelListForRpcParam param = JSONObject.parseObject(request.getParam().toString(), ChannelListForRpcParam.class);
RedisRpcResponse response = request.getResponse();
try {
int result = platformChannelService.addChannels(param.getPlatformId(), param.getChannelIds());
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody(result + "");
}catch (Exception e) {
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody("0");
}
return response;
}
/**
* 移除全部共享通道
*/
@RedisRpcMapping("removeAllChannel")
public RedisRpcResponse removeAllChannel(RedisRpcRequest request) {
Integer platformId = Integer.parseInt(request.getParam().toString());
RedisRpcResponse response = request.getResponse();
try {
int result = platformChannelService.removeAllChannel(platformId);
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody(result + "");
}catch (Exception e) {
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody("0");
}
return response;
}
/**
* 取消共享通道
*/
@RedisRpcMapping("removeChannelList")
public RedisRpcResponse removeChannelList(RedisRpcRequest request) {
ChannelListForRpcParam param = JSONObject.parseObject(request.getParam().toString(), ChannelListForRpcParam.class);
RedisRpcResponse response = request.getResponse();
try {
int result = platformChannelService.removeChannels(param.getPlatformId(), param.getChannelIds());
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody(result + "");
}catch (Exception e) {
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody("0");
}
return response;
}
/**
* 自定义通道
*/
@RedisRpcMapping("updateCustomChannel")
public RedisRpcResponse updateCustomChannel(RedisRpcRequest request) {
PlatformChannel param = JSONObject.parseObject(request.getParam().toString(), PlatformChannel.class);
RedisRpcResponse response = request.getResponse();
try {
platformChannelService.updateCustomChannel(param);
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody("true");
}catch (Exception e) {
response.setStatusCode(ErrorCode.SUCCESS.getCode());
response.setBody("false");
}
return response;
}
/**
* 目录更新推送
*/

67
src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java
View File

@@ -10,6 +10,7 @@ import com.genersoft.iot.vmp.conf.redis.RedisRpcConfig;
import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcRequest;
import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcResponse;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.controller.bean.ChannelListForRpcParam;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.media.event.hook.Hook;
@@ -238,6 +239,72 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
return Boolean.parseBoolean(response.getBody().toString());
}
@Override
public boolean deletePlatform(String serverId, Integer platformId) {
RedisRpcRequest request = buildRequest("platform/delete", platformId);
request.setToId(serverId);
RedisRpcResponse response = redisRpcConfig.request(request, 20, TimeUnit.SECONDS);
if(response == null) {
return false;
}
return Boolean.parseBoolean(response.getBody().toString());
}
@Override
public int addPlatformChannelList(String serverId, ChannelListForRpcParam channelListForRpcParam) {
RedisRpcRequest request = buildRequest("platform/addChannelList", channelListForRpcParam);
request.setToId(serverId);
RedisRpcResponse response = redisRpcConfig.request(request, 20, TimeUnit.SECONDS);
if(response == null) {
return 0;
}
return Integer.parseInt(response.getBody().toString());
}
@Override
public int removeAllPlatformChannel(String serverId, Integer platformId) {
RedisRpcRequest request = buildRequest("platform/removeAllChannel", platformId);
request.setToId(serverId);
RedisRpcResponse response = redisRpcConfig.request(request, 20, TimeUnit.SECONDS);
if(response == null) {
return 0;
}
return Integer.parseInt(response.getBody().toString());
}
@Override
public int removePlatformChannelList(String serverId, ChannelListForRpcParam channelListForRpcParam) {
RedisRpcRequest request = buildRequest("platform/removeChannelList", channelListForRpcParam);
request.setToId(serverId);
RedisRpcResponse response = redisRpcConfig.request(request, 20, TimeUnit.SECONDS);
if(response == null) {
return 0;
}
return Integer.parseInt(response.getBody().toString());
}
@Override
public boolean updateCustomPlatformChannel(String serverId, PlatformChannel channel) {
RedisRpcRequest request = buildRequest("platform/updateCustomChannel", channel);
request.setToId(serverId);
RedisRpcResponse response = redisRpcConfig.request(request, 20, TimeUnit.SECONDS);
if(response == null) {
return false;
}
return Boolean.parseBoolean(response.getBody().toString());
}
@Override
public boolean pushPlatformChannel(String serverId, Integer platformId) {
RedisRpcRequest request = buildRequest("platform/pushChannel", platformId);
request.setToId(serverId);
RedisRpcResponse response = redisRpcConfig.request(request, 20, TimeUnit.SECONDS);
if(response == null) {
return false;
}
return Boolean.parseBoolean(response.getBody().toString());
}
@Override
public void catalogEventPublish(String serverId, CatalogEvent event) {
JSONObject jsonObject = new JSONObject();