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

临时提交

Browse Source
This commit is contained in:
lin
2025-10-22 22:39:17 +08:00
parent aae673e47e
commit 0e331d6b97
11 changed files with 239 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/com/genersoft/iot/vmp/gb28181/dao/CommonGBChannelMapper.java
View File

@@ -591,6 +591,9 @@ public interface CommonGBChannelMapper {
@SelectProvider(type = ChannelProvider.class, method = "queryOnlineListsByGbDeviceId")
List<CommonGBChannel> queryOnlineListsByGbDeviceId(@Param("deviceId") int deviceId);
@SelectProvider(type = ChannelProvider.class, method = "queryCommonChannelByDeviceChannel")
CommonGBChannel queryCommonChannelByDeviceChannel(DeviceChannel channel);
@Update("UPDATE wvp_device_channel SET stream_id = #{stream} where id = #{gbId}")
void updateStream(int gbId, String stream);
@@ -660,5 +663,6 @@ public interface CommonGBChannelMapper {
@SelectProvider(type = ChannelProvider.class, method = "queryCameraChannelByIds")
List<CameraChannel> queryCameraChannelByIds(List<CommonGBChannel> channelList);
@SelectProvider(type = ChannelProvider.class, method = "queryOldChanelListByChannels")
List<CommonGBChannel> queryOldChanelListByChannels(List<CommonGBChannel> channelList);
}

6
src/main/java/com/genersoft/iot/vmp/gb28181/dao/DeviceChannelMapper.java
View File

@@ -105,8 +105,8 @@ public interface DeviceChannelMapper {
@Delete("DELETE FROM wvp_device_channel WHERE data_type =1 and data_device_id=#{dataDeviceId}")
int cleanChannelsByDeviceId(@Param("dataDeviceId") int dataDeviceId);
@Delete("DELETE FROM wvp_device_channel WHERE id=#{id}")
int del(@Param("id") int id);
@Delete("DELETE FROM wvp_device_channel WHERE WHERE data_type=#{dataType} and data_device_id=#{dataDeviceId} AND device_id=#{deviceId}")
int deleteForNotify(DeviceChannel channel);
@Select(value = {" <script>" +
" SELECT " +
@@ -511,7 +511,6 @@ public interface DeviceChannelMapper {
@Update("UPDATE wvp_device_channel SET status=#{status} WHERE data_type=#{dataType} and data_device_id=#{dataDeviceId} AND device_id=#{deviceId}")
void updateStatus(DeviceChannel channel);
@Update({"<script>" +
" UPDATE" +
" wvp_device_channel" +
@@ -620,4 +619,5 @@ public interface DeviceChannelMapper {
@Update(value = {"UPDATE wvp_device_channel SET status = 'OFF' WHERE data_type = 1 and data_device_id=#{deviceId}"})
void offlineByDeviceId(@Param("deviceId") int deviceId);
}

24
src/main/java/com/genersoft/iot/vmp/gb28181/dao/provider/ChannelProvider.java
View File

@@ -516,6 +516,11 @@ public class ChannelProvider {
return sqlBuild.toString();
}
public String queryCommonChannelByDeviceChannel(Map<String, Object> params ){
return BASE_SQL +
" where data_type=#{dataType} and data_device_id=#{dataDeviceId} AND device_id=#{deviceId}";
}
public String queryOnlineListsByGbDeviceId(Map<String, Object> params ){
StringBuilder sqlBuild = new StringBuilder();
sqlBuild.append(BASE_SQL_TABLE_NAME);
@@ -523,6 +528,24 @@ public class ChannelProvider {
return sqlBuild.toString();
}
public String queryOldChanelListByChannels(Map<String, Object> params ){
StringBuilder sqlBuild = new StringBuilder();
sqlBuild.append(BASE_SQL);
sqlBuild.append(" where id in ( ");
List<CommonGBChannel> channelList = (List<CommonGBChannel>)params.get("channelList");
boolean first = true;
for (CommonGBChannel channel : channelList) {
if (!first) {
sqlBuild.append(",");
}
sqlBuild.append(channel.getGbId());
first = false;
}
sqlBuild.append(" )");
return sqlBuild.toString() ;
}
public String queryAllForUnusualCivilCode(Map<String, Object> params ){
StringBuilder sqlBuild = new StringBuilder();
sqlBuild.append("select wdc.id from wvp_device_channel wdc ");
@@ -828,6 +851,7 @@ public class ChannelProvider {
sqlBuild.append(" )");
return sqlBuild.toString() ;
}
public String queryCameraChannelByIds(Map<String, Object> params ){
StringBuilder sqlBuild = new StringBuilder();
sqlBuild.append(BASE_SQL_FOR_CAMERA_DEVICE);

5
src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java
View File

@@ -69,6 +69,11 @@ public class EventPublisher {
applicationEventPublisher.publishEvent(channelEvent);
}
public void channelEventPublishForUpdate(List<CommonGBChannel> channelList, List<CommonGBChannel> channelListForOld) {
ChannelEvent channelEvent = ChannelEvent.getInstanceForUpdate(this, channelList, channelListForOld);
applicationEventPublisher.publishEvent(channelEvent);
}
public void channelEventPublish(List<CommonGBChannel> channelList, ChannelEvent.ChannelEventMessageType type) {
ChannelEvent channelEvent = ChannelEvent.getInstance(this, type, channelList);
applicationEventPublisher.publishEvent(channelEvent);

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

@@ -52,7 +52,7 @@ public interface IDeviceChannelService {
void offline(DeviceChannel channel);
void delete(DeviceChannel channel);
void deleteForNotify(DeviceChannel channel);
void cleanChannelsForDevice(int deviceId);

2
src/main/java/com/genersoft/iot/vmp/gb28181/service/IGbChannelService.java
View File

@@ -102,4 +102,6 @@ public interface IGbChannelService {
List<CommonGBChannel> queryListForMap(String query, Boolean online, Boolean hasRecordPlan, Integer channelType);
void saveLevel(List<ChannelForThin> channels);
CommonGBChannel queryCommonChannelByDeviceChannel(DeviceChannel channel);
}

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

@@ -219,8 +219,8 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
@Override
public void delete(DeviceChannel channel) {
channelMapper.del(channel.getId());
public void deleteForNotify(DeviceChannel channel) {
channelMapper.deleteForNotify(channel);
}
@Override

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

@@ -11,6 +11,7 @@ import com.genersoft.iot.vmp.gb28181.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.gb28181.dao.DeviceMapper;
import com.genersoft.iot.vmp.gb28181.dao.PlatformChannelMapper;
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.IDeviceService;
import com.genersoft.iot.vmp.gb28181.service.IInviteStreamService;
@@ -422,7 +423,7 @@ public class DeviceServiceImpl implements IDeviceService, CommandLineRunner {
}
deviceChannelMapper.offlineByDeviceId(device.getId());
// 发送通道离线通知
eventPublisher.catalogEventPublish(null, channelList, CatalogEvent.OFF);
eventPublisher.channelEventPublish(channelList, ChannelEvent.ChannelEventMessageType.OFF);
}
private boolean isDevice(String deviceId) {
@@ -827,7 +828,7 @@ public class DeviceServiceImpl implements IDeviceService, CommandLineRunner {
try {
// 发送catalog
eventPublisher.catalogEventPublish(null, commonGBChannels, CatalogEvent.DEL);
eventPublisher.channelEventPublish(commonGBChannels, ChannelEvent.ChannelEventMessageType.DEL);
} catch (Exception e) {
log.warn("[多个通道删除] 发送失败，数量：{}", commonGBChannels.size(), e);
}

39
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GbChannelServiceImpl.java
View File

@@ -122,7 +122,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
commonGBChannelMapper.batchDelete(channelListInDb);
try {
// 发送通知
eventPublisher.catalogEventPublish(null, channelListInDb, CatalogEvent.DEL);
eventPublisher.channelEventPublish(channelListInDb, ChannelEvent.ChannelEventMessageType.DEL);
} catch (Exception e) {
log.warn("[通道移除通知] 发送失败", e);
}
@@ -289,6 +289,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
log.warn("[更新多个通道] 通道数量为0，更新失败");
return;
}
List<CommonGBChannel> oldCommonGBChannelList = commonGBChannelMapper.queryOldChanelListByChannels(commonGBChannels);
// 批量保存
int limitCount = 1000;
int result = 0;
@@ -307,8 +308,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
// 发送通过更新通知
try {
// 发送通知
eventPublisher.catalogEventPublish(null, commonGBChannels, CatalogEvent.UPDATE);
// eventPublisher.channelEventPublishForUpdate(commonGBChannels, ChannelEvent.ChannelEventMessageType.ADD);
eventPublisher.channelEventPublishForUpdate(commonGBChannels, oldCommonGBChannelList);
} catch (Exception e) {
log.warn("[更新多个通道] 发送失败，{}个", commonGBChannels.size(), e);
}
@@ -321,6 +321,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
log.warn("[更新多个通道状态] 通道数量为0，更新失败");
return;
}
List<CommonGBChannel> oldChanelListByChannels = commonGBChannelMapper.queryOldChanelListByChannels(commonGBChannels);
int limitCount = 1000;
int result = 0;
if (commonGBChannels.size() > limitCount) {
@@ -338,7 +339,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
// 发送通过更新通知
try {
// 发送通知
eventPublisher.catalogEventPublish(null, commonGBChannels, CatalogEvent.UPDATE);
eventPublisher.channelEventPublishForUpdate(commonGBChannels, oldChanelListByChannels);
} catch (Exception e) {
log.warn("[更新多个通道] 发送失败，{}个", commonGBChannels.size(), e);
}
@@ -445,6 +446,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
if (channelList.isEmpty()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "所有通道Id不存在");
}
List<CommonGBChannel> channelListForOld = new ArrayList<>(channelList);
for (CommonGBChannel channel : channelList) {
channel.setGbCivilCode(civilCode);
}
@@ -454,7 +456,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
platformChannelService.checkRegionAdd(channelList);
try {
// 发送catalog
eventPublisher.catalogEventPublish(null, channelList, CatalogEvent.UPDATE);
eventPublisher.channelEventPublishForUpdate(channelList, channelListForOld);
} catch (Exception e) {
log.warn("[多个通道添加行政区划] 发送失败，数量：{}", channelList.size(), e);
}
@@ -524,6 +526,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
if (channelList.isEmpty()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "所有通道Id不存在");
}
List<CommonGBChannel> channelListForOld = new ArrayList<>(channelList);
for (CommonGBChannel channel : channelList) {
channel.setGbCivilCode(civilCode);
}
@@ -532,7 +535,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
if (result > 0) {
try {
// 发送catalog
eventPublisher.catalogEventPublish(null, channelList, CatalogEvent.UPDATE);
eventPublisher.channelEventPublishForUpdate(channelList, channelListForOld);
} catch (Exception e) {
log.warn("[多个通道添加行政区划] 发送失败，数量：{}", channelList.size(), e);
}
@@ -579,7 +582,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
log.info("[更新业务分组] 发现未关联任何通道： {}", oldBusinessGroup);
return;
}
List<CommonGBChannel> channelListForOld = new ArrayList<>(channelList);
int result = commonGBChannelMapper.updateBusinessGroupByChannelList(newBusinessGroup, channelList);
if (result > 0) {
for (CommonGBChannel channel : channelList) {
@@ -587,7 +590,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
}
// 发送catalog
try {
eventPublisher.catalogEventPublish(null, channelList, CatalogEvent.UPDATE);
eventPublisher.channelEventPublishForUpdate(channelList, channelListForOld);
} catch (Exception e) {
log.warn("[多个通道业务分组] 发送失败，数量：{}", channelList.size(), e);
}
@@ -600,7 +603,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
if (channelList.isEmpty()) {
return;
}
List<CommonGBChannel> channelListForOld = new ArrayList<>(channelList);
int result = commonGBChannelMapper.updateParentIdByChannelList(newParentId, channelList);
if (result > 0) {
for (CommonGBChannel channel : channelList) {
@@ -608,7 +611,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
}
// 发送catalog
try {
eventPublisher.catalogEventPublish(null, channelList, CatalogEvent.UPDATE);
eventPublisher.channelEventPublishForUpdate(channelList, channelListForOld);
} catch (Exception e) {
log.warn("[多个通道业务分组] 发送失败，数量：{}", channelList.size(), e);
}
@@ -622,6 +625,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
if (channelList.isEmpty()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "所有通道Id不存在");
}
List<CommonGBChannel> channelListForOld = new ArrayList<>(channelList);
int result = commonGBChannelMapper.updateGroup(parentId, businessGroup, channelList);
for (CommonGBChannel commonGBChannel : channelList) {
commonGBChannel.setGbParentId(parentId);
@@ -633,7 +637,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
platformChannelService.checkGroupAdd(channelList);
try {
// 发送catalog
eventPublisher.catalogEventPublish(null, channelList, CatalogEvent.UPDATE);
eventPublisher.channelEventPublishForUpdate(channelList, channelListForOld);
} catch (Exception e) {
log.warn("[多个通道添加行政区划] 发送失败，数量：{}", channelList.size(), e);
}
@@ -665,6 +669,8 @@ public class GbChannelServiceImpl implements IGbChannelService {
if (channelList.isEmpty()) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "所有通道Id不存在");
}
List<CommonGBChannel> channelListForOld = new ArrayList<>(channelList);
for (CommonGBChannel channel : channelList) {
channel.setGbParentId(parentId);
channel.setGbBusinessGroupId(businessGroup);
@@ -680,7 +686,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
platformChannelService.checkGroupAdd(channelList);
try {
// 发送catalog
eventPublisher.catalogEventPublish(null, channelList, CatalogEvent.UPDATE);
eventPublisher.channelEventPublishForUpdate(channelList, channelListForOld);
} catch (Exception e) {
log.warn("[多个通道添加行政区划] 发送失败，数量：{}", channelList.size(), e);
}
@@ -714,7 +720,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
if (channelList.isEmpty()) {
return;
}
List<CommonGBChannel> channelListForOld = new ArrayList<>(channelList);
int result = commonGBChannelMapper.updateCivilCodeByChannelList(newCivilCode, channelList);
if (result > 0) {
for (CommonGBChannel channel : channelList) {
@@ -722,7 +728,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
}
// 发送catalog
try {
eventPublisher.catalogEventPublish(null, channelList, CatalogEvent.UPDATE);
eventPublisher.channelEventPublishForUpdate(channelList, channelListForOld);
} catch (Exception e) {
log.warn("[多个通道业务分组] 发送失败，数量：{}", channelList.size(), e);
}
@@ -839,4 +845,9 @@ public class GbChannelServiceImpl implements IGbChannelService {
commonGBChannelMapper.saveLevel(channels);
}
}
@Override
public CommonGBChannel queryCommonChannelByDeviceChannel(DeviceChannel channel) {
return commonGBChannelMapper.queryCommonChannelByDeviceChannel(channel);
}
}

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

@@ -1,9 +1,11 @@
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.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;
@@ -11,6 +13,7 @@ import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
@@ -49,6 +52,153 @@ public class PlatformChannelServiceImpl implements IPlatformChannelService {
@Autowired
private ISIPCommanderForPlatform sipCommanderFroPlatform;
@Autowired
private SubscribeHolder subscribeHolder;
@Autowired
private UserSetting userSetting;
// 监听通道信息变化
@EventListener
public void onApplicationEvent(ChannelEvent event) {
if (event.getChannels().isEmpty()) {
return;
}
// 获取通道所关联的平台
List<Platform> allPlatform = platformMapper.queryByServerId(userSetting.getServerId());
// 获取所用订阅
List<String> platforms = subscribeHolder.getAllCatalogSubscribePlatform(allPlatform);
Map<String, List<Platform>> platformMap = new HashMap<>();
Map<String, CommonGBChannel> channelMap = new HashMap<>();
if (platforms.isEmpty()) {
return;
}
for (CommonGBChannel deviceChannel : event.getChannels()) {
List<Platform> parentPlatformsForGB = queryPlatFormListByChannelDeviceId(
deviceChannel.getGbId(), platforms);
platformMap.put(deviceChannel.getGbDeviceId(), parentPlatformsForGB);
channelMap.put(deviceChannel.getGbDeviceId(), deviceChannel);
}
if (platformMap.isEmpty()) {
return;
}
switch (event.getMessageType()) {
case ON:
case OFF:
case DEL:
for (String serverGbId : platformMap.keySet()) {
List<Platform> platformList = platformMap.get(serverGbId);
if (platformList != null && !platformList.isEmpty()) {
for (Platform platform : platformList) {
SubscribeInfo subscribeInfo = subscribeHolder.getCatalogSubscribe(platform.getServerGBId());
if (subscribeInfo == null) {
continue;
}
log.info("[Catalog事件: {}]平台：{}，影响通道{}", event.getMessageType(), platform.getServerGBId(), serverGbId);
List<CommonGBChannel> deviceChannelList = new ArrayList<>();
CommonGBChannel deviceChannel = new CommonGBChannel();
deviceChannel.setGbDeviceId(serverGbId);
deviceChannelList.add(deviceChannel);
try {
sipCommanderFroPlatform.sendNotifyForCatalogOther(event.getMessageType().name(), platform, deviceChannelList, subscribeInfo, null);
} catch (InvalidArgumentException | ParseException | NoSuchFieldException | SipException |
IllegalAccessException e) {
log.error("[命令发送失败] 国标级联 Catalog通知: {}", e.getMessage());
}
}
}else {
log.info("[Catalog事件: {}] 未找到上级平台： {}", event.getMessageType(), serverGbId);
}
}
break;
case VLOST:
break;
case DEFECT:
break;
case ADD:
case UPDATE:
for (String gbId : platformMap.keySet()) {
List<Platform> parentPlatforms = platformMap.get(gbId);
if (parentPlatforms != null && !parentPlatforms.isEmpty()) {
for (Platform platform : parentPlatforms) {
SubscribeInfo subscribeInfo = subscribeHolder.getCatalogSubscribe(platform.getServerGBId());
if (subscribeInfo == null) {
continue;
}
log.info("[Catalog事件: {}]平台：{}，影响通道{}", event.getMessageType(), platform.getServerGBId(), gbId);
List<CommonGBChannel> channelList = new ArrayList<>();
CommonGBChannel deviceChannel = channelMap.get(gbId);
channelList.add(deviceChannel);
try {
sipCommanderFroPlatform.sendNotifyForCatalogAddOrUpdate(event.getMessageType().name(), platform, channelList, subscribeInfo, null);
} catch (InvalidArgumentException | ParseException | NoSuchFieldException |
SipException | IllegalAccessException e) {
log.error("[命令发送失败] 国标级联 Catalog通知: {}", e.getMessage());
}
}
}
}
break;
default:
break;
}
}
@EventListener
public void onApplicationEvent(CatalogEvent event) {
log.info("[Catalog事件: {}]通道数量： {}", event.getType(), event.getChannels().size());
Platform platform = event.getPlatform();
if (platform == null || platform.getServerGBId() == null) {
return;
}
SubscribeInfo subscribe = subscribeHolder.getCatalogSubscribe(platform.getServerGBId());
if (subscribe == null) {
return;
}
switch (event.getType()) {
case CatalogEvent.ON:
case CatalogEvent.OFF:
case CatalogEvent.DEL:
List<CommonGBChannel> channels = new ArrayList<>();
if (event.getChannels() != null) {
channels.addAll(event.getChannels());
}
if (!channels.isEmpty()) {
log.info("[Catalog事件: {}]平台：{}，影响通道{}个", event.getType(), platform.getServerGBId(), channels.size());
try {
sipCommanderFroPlatform.sendNotifyForCatalogOther(event.getType(), platform, channels, subscribe, null);
} catch (InvalidArgumentException | ParseException | NoSuchFieldException | SipException |
IllegalAccessException e) {
log.error("[命令发送失败] 国标级联 Catalog通知: {}", e.getMessage());
}
}
break;
case CatalogEvent.VLOST:
break;
case CatalogEvent.DEFECT:
break;
case CatalogEvent.ADD:
case CatalogEvent.UPDATE:
List<CommonGBChannel> deviceChannelList = new ArrayList<>();
if (event.getChannels() != null) {
deviceChannelList.addAll(event.getChannels());
}
if (!deviceChannelList.isEmpty()) {
log.info("[Catalog事件: {}]平台：{}，影响通道{}个", event.getType(), platform.getServerGBId(), deviceChannelList.size());
try {
sipCommanderFroPlatform.sendNotifyForCatalogAddOrUpdate(event.getType(), platform, deviceChannelList, subscribe, null);
} catch (InvalidArgumentException | ParseException | NoSuchFieldException | SipException |
IllegalAccessException e) {
log.error("[命令发送失败] 国标级联 Catalog通知: {}", e.getMessage());
}
}
break;
default:
break;
}
}
@Override
public PageInfo<PlatformChannel> queryChannelList(int page, int count, String query, Integer channelType, Boolean online, Integer platformId, Boolean hasShare) {

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

@@ -3,8 +3,10 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
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.IDeviceChannelService;
import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
@@ -50,6 +52,9 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
@Autowired
private IDeviceChannelService deviceChannelService;
@Autowired
private IGbChannelService channelService;
// @Scheduled(fixedRate = 2000) //每400毫秒执行一次
// public void showSize(){
// log.warn("[notify-目录订阅] 待处理消息数量： {}", taskQueue.size() );
@@ -256,8 +261,6 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
log.warn("[ NotifyCatalog ] event not found ： {}", catalogChannelEvent.getEvent());
}
// 转发变化信息
eventPublisher.catalogEventPublish(null, catalogChannelEvent.getChannel(), catalogChannelEvent.getEvent());
}
}
@@ -283,15 +286,28 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
switch (notifyCatalogChannel.getType()) {
case STATUS_CHANGED:
deviceChannelService.updateChannelStatusForNotify(notifyCatalogChannel.getChannel());
CommonGBChannel channelForStatus = channelService.queryCommonChannelByDeviceChannel(notifyCatalogChannel.getChannel());
if ("ON".equals(notifyCatalogChannel.getChannel().getStatus()) ) {
eventPublisher.channelEventPublish(channelForStatus, ChannelEvent.ChannelEventMessageType.ON);
}else {
eventPublisher.channelEventPublish(channelForStatus, ChannelEvent.ChannelEventMessageType.OFF);
}
break;
case ADD:
deviceChannelService.addChannel(notifyCatalogChannel.getChannel());
CommonGBChannel channelForAdd = channelService.getOne(notifyCatalogChannel.getChannel().getId());
eventPublisher.channelEventPublish(channelForAdd, ChannelEvent.ChannelEventMessageType.ADD);
break;
case UPDATE:
CommonGBChannel oldCommonChannel = channelService.queryCommonChannelByDeviceChannel(notifyCatalogChannel.getChannel());
deviceChannelService.updateChannelForNotify(notifyCatalogChannel.getChannel());
CommonGBChannel channel = channelService.getOne(oldCommonChannel.getGbId());
eventPublisher.channelEventPublishForUpdate(channel, oldCommonChannel);
break;
case DELETE:
deviceChannelService.delete(notifyCatalogChannel.getChannel());
CommonGBChannel oldCommonChannelForDelete = channelService.queryCommonChannelByDeviceChannel(notifyCatalogChannel.getChannel());
deviceChannelService.deleteForNotify(notifyCatalogChannel.getChannel());
eventPublisher.channelEventPublish(oldCommonChannelForDelete, ChannelEvent.ChannelEventMessageType.DEL);
break;
}
}catch (Exception e) {