diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlatformChannelServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlatformChannelServiceImpl.java index 15316c7e0..267f795ee 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlatformChannelServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlatformChannelServiceImpl.java @@ -36,6 +36,8 @@ import java.util.stream.Collectors; @RequiredArgsConstructor public class PlatformChannelServiceImpl implements IPlatformChannelService { + private static final int BATCH_SIZE = 500; + private final PlatformChannelMapper platformChannelMapper; private final EventPublisher eventPublisher; @@ -450,34 +452,7 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService { } List channelListShare = platformChannelMapper.queryShare(platformId, null); Assert.notEmpty(channelListShare, "未共享任何通道"); - int result = platformChannelMapper.removeChannelsWithPlatform(platformId, channelListShare); - if (result > 0) { - // 查询通道相关的分组信息 - Set regionSet = regionMapper.queryByChannelList(channelListShare); - Set deleteRegion = deleteEmptyRegion(regionSet, platformId); - if (!deleteRegion.isEmpty()) { - for (Region region : deleteRegion) { - channelListShare.add(0, CommonGBChannel.build(region)); - } - } - - // 查询通道相关的分组信息 - Set groupSet = groupMapper.queryByChannelList(channelListShare); - Set deleteGroup = deleteEmptyGroup(groupSet, platformId); - if (!deleteGroup.isEmpty()) { - for (Group group : deleteGroup) { - channelListShare.add(0, CommonGBChannel.build(group)); - } - } - // 发送消息 - try { - // 发送catalog - eventPublisher.catalogEventPublish(platform, channelListShare, CatalogEvent.DEL); - } catch (Exception e) { - log.warn("[移除全部关联通道] 发送失败，数量：{}", channelListShare.size(), e); - } - } - return result; + return removeChannelsFromDb(platform, platformId, channelListShare); } @Override @@ -494,6 +469,44 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService { removeChannels(platformId, channelList); } + private List> partition(List list, int size) { + List> result = new ArrayList<>(); + for (int i = 0; i < list.size(); i += size) { + result.add(list.subList(i, Math.min(i + size, list.size()))); + } + return result; + } + + private int removeChannelsFromDb(Platform platform, Integer platformId, List channelList) { + List> batches = partition(channelList, BATCH_SIZE); + int totalResult = 0; + for (List batch : batches) { + totalResult += platformChannelMapper.removeChannelsWithPlatform(platformId, batch); + } + if (totalResult > 0) { + Set regionSet = regionMapper.queryByChannelList(channelList); + Set deleteRegion = deleteEmptyRegion(regionSet, platformId); + if (!deleteRegion.isEmpty()) { + for (Region region : deleteRegion) { + channelList.add(0, CommonGBChannel.build(region)); + } + } + Set groupSet = groupMapper.queryByChannelList(channelList); + Set deleteGroup = deleteEmptyGroup(groupSet, platformId); + if (!deleteGroup.isEmpty()) { + for (Group group : deleteGroup) { + channelList.add(0, CommonGBChannel.build(group)); + } + } + try { + eventPublisher.catalogEventPublish(platform, channelList, CatalogEvent.DEL); + } catch (Exception e) { + log.warn("[取消共享通道] 发送失败，数量：{}", channelList.size(), e); + } + } + return totalResult; + } + @Transactional public int removeChannelList(Integer platformId, List channelList) { Platform platform = platformMapper.query(platformId); @@ -513,36 +526,12 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService { } return result; } - String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(",")); - int result = platformChannelMapper.removeChannelsWithPlatform(platformId, channelList); + int result = removeChannelsFromDb(platform, platformId, channelList); if (result <= 0) { + String deviceIds = channelList.stream().map(CommonGBChannel::getGbDeviceId).collect(Collectors.joining(",")); log.info("[取消共享通道] 平台{}未关联通道： {}", platformId, deviceIds); return 0; } - // 查询通道相关的分组信息 - Set regionSet = regionMapper.queryByChannelList(channelList); - Set deleteRegion = deleteEmptyRegion(regionSet, platformId); - if (!deleteRegion.isEmpty()) { - for (Region region : deleteRegion) { - channelList.add(0, CommonGBChannel.build(region)); - } - } - - // 查询通道相关的分组信息 - Set groupSet = groupMapper.queryByChannelList(channelList); - Set deleteGroup = deleteEmptyGroup(groupSet, platformId); - if (!deleteGroup.isEmpty()) { - for (Group group : deleteGroup) { - channelList.add(0, CommonGBChannel.build(group)); - } - } - // 发送消息 - try { - // 发送catalog - eventPublisher.catalogEventPublish(platform, channelList, CatalogEvent.DEL); - } catch (Exception e) { - log.warn("[取消共享通道] 发送失败，数量：{}", channelList.size(), e); - } return result; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlayServiceImpl.java index 47af047e9..f0967ea57 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlayServiceImpl.java @@ -861,7 +861,9 @@ public class PlayServiceImpl implements IPlayService { log.info("[Invite 200OK] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse); String oldStreamId = String.format("%08x", Long.parseLong(ssrcInfo.getSsrc())).toUpperCase(); String newStreamId = String.format("%08x", Long.parseLong(ssrcInResponse)).toUpperCase(); - receiveRtpServerService.refreshAuthenticateInfo(oldStreamId, newStreamId); + if (!mediaServerItem.isRtpEnable()) { // 多端口时按照端口绑定了stream，即使stream与ssrc不一致，也不会影响 + receiveRtpServerService.refreshAuthenticateInfo(oldStreamId, newStreamId); + } // ssrc 不一致 if (mediaServerItem.isRtpEnable()) { // 多端口 diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 6135d80c7..8d4238aec 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -177,6 +177,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements String sendSsrc = sendSsrcFactory.getSendSsrc( "Play".equalsIgnoreCase(finalInviteInfo.getSessionName()) ? "0" : "1"); finalInviteInfo.setSsrc(sendSsrc); + log.info("[上级INVITE] 使用自定义SSRC: {}", sendSsrc); } // 构建sendRTP内容 SendRtpInfo sendRtpItem = sendRtpServerService.createSendRtpInfo(streamInfo.getMediaServer(), @@ -224,8 +225,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements } catch (SipException | InvalidArgumentException | ParseException e) { log.error("[命令发送失败] 上级INVITE 发送 200（SDP）: {}", e.getMessage()); } - - } })); }