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

648540858

2022-01-27 f33c3a36302749d8552b281de3dbe37f672bba86

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

 sql/mysql.sql

@@ -444,6 +444,7 @@
  `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`)

 sql/update.sql

New file
@@ -0,0 +1 @@
ALTER TABLE stream_proxy ADD status bit(1) not null;

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java

@@ -332,6 +332,11 @@
            }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 @@
                                    || 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);


 src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java

@@ -151,4 +151,5 @@
    public void setEnable_remove_none_reader(boolean enable_remove_none_reader) {
        this.enable_remove_none_reader = enable_remove_none_reader;
    }

}

 src/main/java/com/genersoft/iot/vmp/service/IStreamProxyService.java

@@ -56,6 +56,16 @@
    boolean start(String app, String stream);

    /**
     * 更新状态
     * @param status 状态
     * @param app
     * @param stream
     */
    int updateStatus(boolean status, String app, String stream);



    /**
     * 停用用视频代理
     * @param app
     * @param stream

 src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java

@@ -14,8 +14,11 @@
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;
@@ -69,6 +72,12 @@

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private IVideoManagerStorager storager;

    @Autowired
    private IStreamProxyService streamProxyService;

    @Autowired
    private EventPublisher publisher;
@@ -231,6 +240,7 @@
    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 @@
                result.add((MediaServerItem) redisUtil.get(serverKey));
            }
        }
        Collections.reverse(result);
        return result;
    }

@@ -374,6 +385,7 @@
        resetOnlineServerItem(serverItem);
        updateMediaServerKeepalive(serverItem.getId(), null);
        setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable()));

        publisher.zlmOnlineEventPublish(serverItem.getId());

    }

 src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java

@@ -59,6 +59,9 @@
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private IVideoManagerStorager storager;

    @Autowired
    private UserSetup userSetup;

    @Autowired
@@ -278,7 +281,27 @@

    @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 @@
            gbStreamMapper.batchDel(streamProxyItemList);
        }
        streamProxyMapper.deleteAutoRemoveItemByMediaServerId(mediaServerId);
        // 其他的流设置未启用
        streamProxyMapper.updateStatus(false, mediaServerId);
        // 其他的流设置离线
        streamProxyMapper.updateStatusByMediaServerId(false, mediaServerId);
        String type = "PULL";

        // 发送redis消息
@@ -314,4 +337,9 @@
    public void clean() {

    }

    @Override
    public int updateStatus(boolean status, String app, String stream) {
        return streamProxyMapper.updateStatus(status, app, stream);
    }
}

 src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java

@@ -398,10 +398,11 @@
    /**

     * 根据媒体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获取其所在设备


 src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java

@@ -11,9 +11,9 @@
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 @@
            "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 @@

    @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 @@
    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);

 src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java

@@ -860,8 +860,8 @@
    }

    @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);
    }



 web_src/src/components/StreamProxyList.vue

@@ -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">

 web_src/src/components/dialog/StreamProxyEdit.vue

@@ -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 @@
          enable_mp4: false,
          enable_remove_none_reader: false,
          platformGbId: null,
          mediaServerId: "auto",
          mediaServerId: null,
      },
      mediaServerList:{},
      ffmpegCmdList:{},
@@ -206,7 +205,8 @@
        console.log(error);
      });
      this.mediaServer.getOnlineMediaServerList((data)=>{
        this.mediaServerList = data;
        this.mediaServerList = data.data;
        this.proxyParam.mediaServerId = this.mediaServerList[0].id
      })
    },
    mediaServerIdChange:function (){