From 81f69eb6f47b69cd89da7621889629f4f456dce1 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期二, 12 七月 2022 17:33:17 +0800 Subject: [PATCH] 支持从redis消息更新推流设备状态 --- src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 32 +++++++ src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java | 7 - src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java | 2 src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java | 9 +- src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java | 5 + src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java | 42 ++++++++++ src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusMsgListener.java | 89 ++++++++++++++++++++++ src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java | 2 src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java | 4 + src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java | 35 +++++++- 10 files changed, 210 insertions(+), 17 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index 702e357..40a7352 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/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_"; diff --git a/src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java index ec1f9ba..6a862ae 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java +++ b/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; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java index e38733d..ac30450 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java @@ -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; diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java index 366ed22..cbef9ce 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java +++ b/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; } diff --git a/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java b/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java index acf0d27..5dd45ef 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java +++ b/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 娴両D - * @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); + } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusMsgListener.java new file mode 100644 index 0000000..a34315b --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusMsgListener.java @@ -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 娑堟伅]鏈敹鍒皉edis鍥炲鎺ㄦ祦璁惧鐘舵�侊紝鎵ц鎺ㄦ祦璁惧绂荤嚎"); + // 浜旂鏀朵笉鍒拌姹傚氨璁剧疆閫氶亾绂荤嚎锛岀劧鍚庨�氱煡涓婄骇绂荤嚎 + streamPushService.allStreamOffline(); + }, 5000); + } + +} diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java index 07fffdc..83116f3 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java +++ b/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 diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java index 837e135..646d287 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java +++ b/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); + } } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index 79e6b26..b9811da 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -233,4 +233,9 @@ * @return */ StreamAuthorityInfo getStreamAuthorityInfo(String app, String stream); + + /** + * 鍙戦�乺edis娑堟伅 鏌ヨ鎵�鏈夋帹娴佽澶囩殑鐘舵�� + */ + void sendStreamPushRequestedMsgForStatus(); } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java index 6c1e72d..bcf57a6 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamPushMapper.java +++ b/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(); } -- Gitblit v1.8.0