src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
@@ -104,6 +104,10 @@ // 设备状态订阅的通知 public static final String VM_MSG_SUBSCRIBE_DEVICE_STATUS = "device"; //************************** 第三方 **************************************** public static final String WVP_STREAM_GB_ID_PREFIX = "memberNo_"; public static final String WVP_STREAM_GPS_MSG_PREFIX = "WVP_STREAM_GPS_MSG_"; src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java
@@ -1,10 +1,7 @@ package com.genersoft.iot.vmp.conf; import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.service.impl.RedisAlarmMsgListener; import com.genersoft.iot.vmp.service.impl.RedisGpsMsgListener; import com.genersoft.iot.vmp.service.impl.RedisGbPlayMsgListener; import com.genersoft.iot.vmp.service.impl.RedisStreamMsgListener; import com.genersoft.iot.vmp.service.impl.*; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; @@ -60,6 +57,9 @@ @Autowired private RedisGbPlayMsgListener redisGbPlayMsgListener; @Autowired private RedisPushStreamStatusMsgListener redisPushStreamStatusMsgListener; @Bean public JedisPool jedisPool() { if (StringUtils.isBlank(password)) { @@ -108,6 +108,7 @@ container.addMessageListener(redisAlarmMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_SUBSCRIBE_ALARM_RECEIVE)); container.addMessageListener(redisStreamMsgListener, new PatternTopic(VideoManagerConstants.WVP_MSG_STREAM_CHANGE_PREFIX + "PUSH")); container.addMessageListener(redisGbPlayMsgListener, new PatternTopic(RedisGbPlayMsgListener.WVP_PUSH_STREAM_KEY)); container.addMessageListener(redisPushStreamStatusMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_STATUS_CHANGE)); return container; }
@@ -58,7 +58,7 @@ ParentPlatform parentPlatform = null; Map<String, List<ParentPlatform>> parentPlatformMap = new HashMap<>(); if (event.getPlatformId() != null) { if (!StringUtils.isEmpty(event.getPlatformId())) { parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformId()); if (parentPlatform != null && !parentPlatform.isStatus()) { return; src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
@@ -292,7 +292,7 @@ logger.warn("查询流({}/{})是否有其它观看者时得到： {}", app, streamId, mediaInfo.getString("msg")); return -1; } if ( code == 0 && ! mediaInfo.getBoolean("online")) { if ( code == 0 && mediaInfo.getBoolean("online") != null && !mediaInfo.getBoolean("online")) { logger.warn("查询流({}/{})是否有其它观看者时得到： {}", app, streamId, mediaInfo.getString("msg")); return -1; } src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
@@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis; import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto; import com.github.pagehelper.PageInfo; @@ -44,31 +45,55 @@ * 停止一路推流 * @param app 应用名 * @param streamId 流ID * @return */ boolean stop(String app, String streamId); /** * 新的节点加入 * @param mediaServerId * @return */ void zlmServerOnline(String mediaServerId); /** * 节点离线 * @param mediaServerId * @return */ void zlmServerOffline(String mediaServerId); /** * 清空 */ void clean(); boolean saveToRandomGB(); /** * 批量添加 */ void batchAdd(List<StreamPushItem> streamPushExcelDtoList); /** * 中止多个推流 */ boolean batchStop(List<GbStream> streamPushItems); /** * 导入时批量增加 */ void batchAddForUpload(List<StreamPushItem> streamPushItems, Map<String, List<String[]>> streamPushItemsForAll); /** * 全部离线 */ void allStreamOffline(); /** * 推流离线 */ void offline(List<StreamPushItemFromRedis> offlineStreams); /** * 推流上线 */ void online(List<StreamPushItemFromRedis> onlineStreams); } src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusMsgListener.java
New file @@ -0,0 +1,89 @@ package com.genersoft.iot.vmp.service.impl; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.GbStream; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.service.bean.PushStreamStatusChangeFromRedisDto; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.connection.MessageListener; import org.springframework.stereotype.Component; import java.util.List; /** * 接收redis发送的推流设备上线下线通知 * @author lin */ @Component public class RedisPushStreamStatusMsgListener implements MessageListener, ApplicationRunner { private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusMsgListener.class); @Autowired private IRedisCatchStorage redisCatchStorage; @Autowired private IStreamPushService streamPushService; @Autowired private EventPublisher eventPublisher; @Autowired private UserSetting userSetting; @Autowired private DynamicTask dynamicTask; @Override public void onMessage(Message message, byte[] bytes) { PushStreamStatusChangeFromRedisDto statusChangeFromPushStream = JSON.parseObject(message.getBody(), PushStreamStatusChangeFromRedisDto.class); if (statusChangeFromPushStream == null) { logger.warn("[REDIS 消息]推流设备状态变化消息解析失败"); return; } if (statusChangeFromPushStream.isSetAllOffline()) { // 所有设备离线 streamPushService.allStreamOffline(); } if (statusChangeFromPushStream.getOfflineStreams().size() > 0) { // 更新部分设备离线 streamPushService.offline(statusChangeFromPushStream.getOfflineStreams()); } if (statusChangeFromPushStream.getOnlineStreams().size() > 0) { // 更新部分设备上线 streamPushService.online(statusChangeFromPushStream.getOnlineStreams()); } } @Override public void run(ApplicationArguments args) throws Exception { // 启动时设置所有推流通道离线，发起查询请求 redisCatchStorage.sendStreamPushRequestedMsgForStatus(); dynamicTask.startDelay(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED, ()->{ logger.info("[REDIS 消息]未收到redis回复推流设备状态，执行推流设备离线"); // 五秒收不到请求就设置通道离线，然后通知上级离线 streamPushService.allStreamOffline(); }, 5000); } } src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java
@@ -3,16 +3,12 @@ import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.AlarmChannelMessage; import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.DeviceAlarm; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -22,6 +18,7 @@ /** * 接收其他wvp发送流变化通知 * @author lin */ @Component src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
@@ -13,6 +13,7 @@ import com.genersoft.iot.vmp.service.IGbStreamService; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.dao.*; import com.genersoft.iot.vmp.utils.DateUtil; @@ -181,7 +182,6 @@ @Override public StreamPushItem getPush(String app, String streamId) { return streamPushMapper.selectOne(app, streamId); } @@ -481,4 +481,34 @@ } return true; } @Override public void allStreamOffline() { List<GbStream> onlinePushers = streamPushMapper.getOnlinePusherForGb(); if (onlinePushers.size() == 0) { return; } streamPushMapper.allStreamOffline(); // 发送通知 eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.OFF); } @Override public void offline(List<StreamPushItemFromRedis> offlineStreams) { // 更新部分设备离线 List<GbStream> onlinePushers = streamPushMapper.getOnlinePusherForGbInList(offlineStreams); streamPushMapper.offline(offlineStreams); // 发送通知 eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.OFF); } @Override public void online(List<StreamPushItemFromRedis> onlineStreams) { // 更新部分设备上线streamPushService List<GbStream> onlinePushers = streamPushMapper.getOfflinePusherForGbInList(onlineStreams); streamPushMapper.online(onlineStreams); // 发送通知 eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.ON); } } src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
@@ -233,4 +233,9 @@ * @return */ StreamAuthorityInfo getStreamAuthorityInfo(String app, String stream); /** * 发送redis消息 查询所有推流设备的状态 */ void sendStreamPushRequestedMsgForStatus(); } src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java
@@ -2,6 +2,7 @@ import com.genersoft.iot.vmp.gb28181.bean.GbStream; import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis; import org.apache.ibatis.annotations.*; // import org.omg.PortableInterceptor.INACTIVE; import org.springframework.stereotype.Repository; @@ -117,4 +118,45 @@ "SET status=#{status} " + "WHERE mediaServerId=#{mediaServerId}") void updateStatusByMediaServerId(String mediaServerId, boolean status); @Select("<script> "+ "SELECT gs.* FROM stream_push sp left join gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream " + "where sp.status = 1 and (gs.app, gs.stream) in" + "<foreach collection='offlineStreams' item='item' separator=','>" + "(#{item.app}, {item.stream}) " + "</foreach>" + "</script>") List<GbStream> getOnlinePusherForGbInList(List<StreamPushItemFromRedis> offlineStreams); @Update("<script> "+ "UPDATE stream_push SET status=0 where (app, stream) in" + "<foreach collection='offlineStreams' item='item' separator=','>" + "(#{item.app}, {item.stream}) " + "</foreach>" + "</script>") void offline(List<StreamPushItemFromRedis> offlineStreams); @Select("<script> "+ "SELECT * FROM stream_push sp left join gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream " + "where sp.status = 0 and (gs.app, gs.stream) in" + "<foreach collection='offlineStreams' item='item' separator=','>" + "(#{item.app}, {item.stream}) " + "</foreach>" + "</script>") List<GbStream> getOfflinePusherForGbInList(List<StreamPushItemFromRedis> onlineStreams); @Update("<script> "+ "UPDATE stream_push SET status=1 where (app, stream) in" + "<foreach collection='offlineStreams' item='item' separator=','>" + "(#{item.app}, {item.stream}) " + "</foreach>" + "</script>") void online(List<StreamPushItemFromRedis> onlineStreams); @Select("SELECT gs.* FROM stream_push sp left join gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream") List<GbStream> getOnlinePusherForGb(); @Update("UPDATE stream_push SET status=0") void allStreamOffline(); }
