添加重启后拉流代理自动恢复

648540858
1 parent 66cadafd
Showing 12 changed files with 90 additions and 16 deletions
sql/mysql.sql
sql/update.sql
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java
src/main/java/com/genersoft/iot/vmp/service/IStreamProxyService.java
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java
src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java
web_src/src/components/StreamProxyList.vue
web_src/src/components/dialog/StreamProxyEdit.vue
@@ -444,6 +444,7 @@ CREATE TABLE `stream_proxy` (
   `enable_hls` bit(1) DEFAULT NULL,
   `enable_mp4` bit(1) DEFAULT NULL,
   `enable` bit(1) NOT NULL,
+  `status` bit(1) NOT NULL,
   `enable_remove_none_reader` bit(1) NOT NULL,
   `createTime` varchar(50) NOT NULL,
   PRIMARY KEY (`app`,`stream`)
+ALTER TABLE stream_proxy ADD status bit(1) not null;
 \ No newline at end of file
@@ -332,6 +332,11 @@ public class ZLMHttpHookListener {
 			}else {
 				mediaServerService.removeCount(mediaServerId);
 			}
+			if (item.getOriginType() == OriginType.PULL.ordinal()
+					|| item.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {
+				// 设置拉流代理上线/离线
+				streamProxyService.updateStatus(regist, app, streamId);
+			}
 			if ("rtp".equals(app) && !regist ) {
 				StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
 				if (streamInfo!=null){
@@ -355,6 +360,7 @@ public class ZLMHttpHookListener {
 									|| item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
 								streamPushItem = zlmMediaListManager.addPush(item);
 							}
+
 							List<GbStream> gbStreams = new ArrayList<>();
 							if (streamPushItem == null || streamPushItem.getGbId() == null) {
 								GbStream gbStream = storager.getGbStream(app, streamId);
@@ -151,4 +151,5 @@ public class StreamProxyItem extends GbStream {
     public void setEnable_remove_none_reader(boolean enable_remove_none_reader) {
         this.enable_remove_none_reader = enable_remove_none_reader;
     }
+
 }
@@ -56,6 +56,16 @@ public interface IStreamProxyService {
     boolean start(String app, String stream);
     /**
+     * 更新状态
+     * @param status 状态
+     * @param app
+     * @param stream
+     */
+    int updateStatus(boolean status, String app, String stream);
+
+
+
+    /**
      * 停用用视频代理
      * @param app
      * @param stream
@@ -14,8 +14,11 @@ import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
 import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
 import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
 import com.genersoft.iot.vmp.service.IMediaServerService;
+import com.genersoft.iot.vmp.service.IStreamProxyService;
 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
 import com.genersoft.iot.vmp.storager.dao.MediaServerMapper;
 import com.genersoft.iot.vmp.utils.redis.JedisUtil;
 import com.genersoft.iot.vmp.utils.redis.RedisUtil;
@@ -71,6 +74,12 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
     private RedisUtil redisUtil;
     @Autowired
+    private IVideoManagerStorager storager;
+
+    @Autowired
+    private IStreamProxyService streamProxyService;
+
+    @Autowired
     private EventPublisher publisher;
     @Autowired
@@ -231,6 +240,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
     public List<MediaServerItem> getAllOnline() {
         String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId();
         Set<String> mediaServerIdSet = redisUtil.zRevRange(key, 0, -1);
+
         List<MediaServerItem> result = new ArrayList<>();
         if (mediaServerIdSet != null && mediaServerIdSet.size() > 0) {
             for (String mediaServerId : mediaServerIdSet) {
@@ -238,6 +248,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
                 result.add((MediaServerItem) redisUtil.get(serverKey));
             }
         }
+        Collections.reverse(result);
         return result;
     }
@@ -374,6 +385,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
         resetOnlineServerItem(serverItem);
         updateMediaServerKeepalive(serverItem.getId(), null);
         setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable()));
+
         publisher.zlmOnlineEventPublish(serverItem.getId());
     }
@@ -59,6 +59,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     private IRedisCatchStorage redisCatchStorage;
     @Autowired
+    private IVideoManagerStorager storager;
+
+    @Autowired
     private UserSetup userSetup;
     @Autowired
@@ -278,7 +281,27 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     @Override
     public void zlmServerOnline(String mediaServerId) {
-        zlmServerOffline(mediaServerId);
+        // 移除开启了无人观看自动移除的流
+        List<StreamProxyItem> streamProxyItemList = streamProxyMapper.selecAutoRemoveItemByMediaServerId(mediaServerId);
+        if (streamProxyItemList.size() > 0) {
+            gbStreamMapper.batchDel(streamProxyItemList);
+        }
+        streamProxyMapper.deleteAutoRemoveItemByMediaServerId(mediaServerId);
+
+        // 恢复流代理, 只查找这个这个流媒体
+        List<StreamProxyItem> streamProxyListForEnable = storager.getStreamProxyListForEnableInMediaServer(
+                mediaServerId, true, false);
+        for (StreamProxyItem streamProxyDto : streamProxyListForEnable) {
+            logger.info("恢复流代理，" + streamProxyDto.getApp() + "/" + streamProxyDto.getStream());
+            JSONObject jsonObject = addStreamProxyToZlm(streamProxyDto);
+            if (jsonObject == null) {
+                // 设置为离线
+                logger.info("恢复流代理失败" + streamProxyDto.getApp() + "/" + streamProxyDto.getStream());
+                updateStatus(false, streamProxyDto.getApp(), streamProxyDto.getStream());
+            }else {
+                updateStatus(true, streamProxyDto.getApp(), streamProxyDto.getStream());
+            }
+        }
     }
     @Override
@@ -289,8 +312,8 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
             gbStreamMapper.batchDel(streamProxyItemList);
         }
         streamProxyMapper.deleteAutoRemoveItemByMediaServerId(mediaServerId);
-        // 其他的流设置未启用
-        streamProxyMapper.updateStatus(false, mediaServerId);
+        // 其他的流设置离线
+        streamProxyMapper.updateStatusByMediaServerId(false, mediaServerId);
         String type = "PULL";
         // 发送redis消息
@@ -314,4 +337,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     public void clean() {
     }
+
+    @Override
+    public int updateStatus(boolean status, String app, String stream) {
+        return streamProxyMapper.updateStatus(status, app, stream);
+    }
 }
@@ -398,10 +398,11 @@ public interface IVideoManagerStorager {
 	/**
 	 * 根据媒体ID获取启用/不启用的代理列表
 	 * @param id 媒体ID
-	 * @param b 启用/不启用
+	 * @param enable 启用/不启用
+	 * @param status 状态
 	 * @return
 	 */
-	List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id, boolean b);
+	List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id,  boolean enable, boolean status);
 	/**
 	 * 根据通道ID获取其所在设备
@@ -11,9 +11,9 @@ import java.util.List;
 public interface StreamProxyMapper {
     @Insert("INSERT INTO stream_proxy (type, app, stream,mediaServerId, url, src_url, dst_url, " +
-            "timeout_ms, ffmpeg_cmd_key, rtp_type, enable_hls, enable_mp4, enable, enable_remove_none_reader, createTime) VALUES" +
+            "timeout_ms, ffmpeg_cmd_key, rtp_type, enable_hls, enable_mp4, enable, status, enable_remove_none_reader, createTime) VALUES" +
             "('${type}','${app}', '${stream}', '${mediaServerId}','${url}', '${src_url}', '${dst_url}', " +
-            "'${timeout_ms}', '${ffmpeg_cmd_key}', '${rtp_type}', ${enable_hls}, ${enable_mp4}, ${enable}, " +
+            "'${timeout_ms}', '${ffmpeg_cmd_key}', '${rtp_type}', ${enable_hls}, ${enable_mp4}, ${enable}, ${status}, " +
             "${enable_remove_none_reader}, '${createTime}' )")
     int add(StreamProxyItem streamProxyDto);
@@ -30,6 +30,7 @@ public interface StreamProxyMapper {
             "rtp_type=#{rtp_type}, " +
             "enable_hls=#{enable_hls}, " +
             "enable=#{enable}, " +
+            "status=#{status}, " +
             "enable_remove_none_reader=#{enable_remove_none_reader}, " +
             "enable_mp4=#{enable_mp4} " +
             "WHERE app=#{app} AND stream=#{stream}")
@@ -49,8 +50,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.mediaServerId = '${id}' order by st.createTime desc")
-    List<StreamProxyItem> selectForEnableInMediaServer(String id, boolean enable);
+            "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);
     @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 " +
@@ -58,9 +59,14 @@ public interface StreamProxyMapper {
     List<StreamProxyItem> selectInMediaServer(String id);
     @Update("UPDATE stream_proxy " +
-            "SET enable=#{status} " +
+            "SET status=#{status} " +
             "WHERE mediaServerId=#{mediaServerId}")
-    void updateStatus(boolean status, String mediaServerId);
+    void updateStatusByMediaServerId(boolean status, String mediaServerId);
+
+    @Update("UPDATE stream_proxy " +
+            "SET status=${status} " +
+            "WHERE app=#{app} AND stream=#{stream}")
+    int updateStatus(boolean status, String app, String stream);
     @Delete("DELETE FROM stream_proxy WHERE enable_remove_none_reader=true AND mediaServerId=#{mediaServerId}")
     void deleteAutoRemoveItemByMediaServerId(String mediaServerId);
@@ -860,8 +860,8 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
 	}
 	@Override
-	public List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id, boolean enable) {
-		return streamProxyMapper.selectForEnableInMediaServer(id, enable);
+	public List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id, boolean enable, boolean status) {
+		return streamProxyMapper.selectForEnableInMediaServer(id, enable, status);
 	}
@@ -42,6 +42,14 @@
           </el-table-column>
 					<el-table-column prop="gbId" label="国标编码" width="180" align="center" show-overflow-tooltip/>
+          <el-table-column label="状态" width="120" align="center">
+            <template slot-scope="scope">
+              <div slot="reference" class="name-wrapper">
+                <el-tag size="medium" v-if="scope.row.status">在线</el-tag>
+                <el-tag size="medium" type="info" v-if="!scope.row.status">离线</el-tag>
+              </div>
+            </template>
+          </el-table-column>
           <el-table-column label="启用" width="120" align="center">
             <template slot-scope="scope">
               <div slot="reference" class="name-wrapper">
@@ -46,7 +46,6 @@
                   style="width: 100%"
                   placeholder="请选择拉流节点"
                 >
-                  <el-option label="自动选择" value="auto"></el-option>
                   <el-option
                     v-for="item in mediaServerList"
                     :key="item.id"
@@ -172,7 +171,7 @@ export default {
           enable_mp4: false,
           enable_remove_none_reader: false,
           platformGbId: null,
-          mediaServerId: "auto",
+          mediaServerId: null,
       },
       mediaServerList:{},
       ffmpegCmdList:{},
@@ -206,7 +205,8 @@ export default {
         console.log(error);
       });
       this.mediaServer.getOnlineMediaServerList((data)=>{
-        this.mediaServerList = data;
+        this.mediaServerList = data.data;
+        this.proxyParam.mediaServerId = this.mediaServerList[0].id
       })
     },
     mediaServerIdChange:function (){