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