修复重启服务后拉流代理回复失败的问题

648540858
1 parent c9f091d5
Showing 6 changed files with 52 additions and 10 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorage.java
src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/media/MediaController.java
@@ -175,3 +175,4 @@ public class CatalogEventLister implements ApplicationListener&lt;CatalogEvent&gt; {
         }
     }
 }
+ 
 \ No newline at end of file
 package com.genersoft.iot.vmp.service.impl;
+import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.SipConfig;
@@ -285,9 +286,12 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
         }
         streamProxyMapper.deleteAutoRemoveItemByMediaServerId(mediaServerId);
+        // 移除拉流代理生成的流信息
+//        syncPullStream(mediaServerId);
+
         // 恢复流代理, 只查找这个这个流媒体
         List<StreamProxyItem> streamProxyListForEnable = storager.getStreamProxyListForEnableInMediaServer(
-                mediaServerId, true, false);
+                mediaServerId, true);
         for (StreamProxyItem streamProxyDto : streamProxyListForEnable) {
             logger.info("恢复流代理，" + streamProxyDto.getApp() + "/" + streamProxyDto.getStream());
             JSONObject jsonObject = addStreamProxyToZlm(streamProxyDto);
@@ -339,4 +343,45 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     public int updateStatus(boolean status, String app, String stream) {
         return streamProxyMapper.updateStatus(status, app, stream);
     }
+
+    private void syncPullStream(String mediaServerId){
+        MediaServerItem mediaServer = mediaServerService.getOne(mediaServerId);
+        if (mediaServer != null) {
+            List<MediaItem> allPullStream = redisCatchStorage.getStreams(mediaServerId, "PULL");
+            if (allPullStream.size() > 0) {
+                zlmresTfulUtils.getMediaList(mediaServer, jsonObject->{
+                    Map<String, StreamInfo> stringStreamInfoMap = new HashMap<>();
+                    if (jsonObject.getInteger("code") == 0) {
+                        JSONArray data = jsonObject.getJSONArray("data");
+                        if(data != null && data.size() > 0) {
+                            for (int i = 0; i < data.size(); i++) {
+                                JSONObject streamJSONObj = data.getJSONObject(i);
+                                if ("rtmp".equals(streamJSONObj.getString("schema"))) {
+                                    StreamInfo streamInfo = new StreamInfo();
+                                    String app = streamJSONObj.getString("app");
+                                    String stream = streamJSONObj.getString("stream");
+                                    streamInfo.setApp(app);
+                                    streamInfo.setStream(stream);
+                                    stringStreamInfoMap.put(app+stream, streamInfo);
+                                }
+                            }
+                        }
+                    }
+                    if (stringStreamInfoMap.size() == 0) {
+                        redisCatchStorage.removeStream(mediaServerId, "PULL");
+                    }else {
+                        for (String key : stringStreamInfoMap.keySet()) {
+                            StreamInfo streamInfo = stringStreamInfoMap.get(key);
+                            if (stringStreamInfoMap.get(streamInfo.getApp() + streamInfo.getStream()) == null) {
+                                redisCatchStorage.removeStream(mediaServerId, "PULL", streamInfo.getApp(),
+                                        streamInfo.getStream());
+                            }
+                        }
+                    }
+                });
+            }
+
+        }
+
+    }
 }
@@ -401,10 +401,9 @@ public interface IVideoManagerStorage {
 	 * 根据媒体ID获取启用/不启用的代理列表
 	 * @param id 媒体ID
 	 * @param enable 启用/不启用
-	 * @param status 状态
 	 * @return
 	 */
-	List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id,  boolean enable, boolean status);
+	List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id,  boolean enable);
 	/**
 	 * 根据通道ID获取其所在设备
@@ -51,8 +51,8 @@ public interface StreamProxyMapper {
     @Select("SELECT st.*, pgs.gbId, pgs.name, pgs.longitude, pgs.latitude FROM stream_proxy st " +
             "LEFT JOIN gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream " +
-            "WHERE st.enable=${enable} and st.status=${status} and st.mediaServerId = '${id}' order by st.createTime desc")
-    List<StreamProxyItem> selectForEnableInMediaServer(String id, boolean enable, boolean status);
+            "WHERE st.enable=${enable} and st.mediaServerId = #{id} order by st.createTime desc")
+    List<StreamProxyItem> selectForEnableInMediaServer(String id, boolean enable);
     @Select("SELECT st.*, pgs.gbId, pgs.name, pgs.longitude, pgs.latitude FROM stream_proxy st " +
             "LEFT JOIN gb_stream pgs on st.app = pgs.app AND st.stream = pgs.stream " +
@@ -911,8 +911,8 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
 	}
 	@Override
-	public List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id, boolean enable, boolean status) {
-		return streamProxyMapper.selectForEnableInMediaServer(id, enable, status);
+	public List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id, boolean enable) {
+		return streamProxyMapper.selectForEnableInMediaServer(id, enable);
 	}
@@ -65,7 +65,4 @@ public class MediaController {
         }
         return result;
     }
-
-
-
 }