兼容流注销时流产生类型错误的问题

648540858

2021-12-03 42d8fff57403500b880886e13093e1841bf47e54

兼容流注销时流产生类型错误的问题

 src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java

@@ -56,5 +56,5 @@
    public static final String MEDIA_TRANSACTION_USED_PREFIX = "VMP_media_transaction_";



    //************************** redis 消息*********************************

    public static final String WVP_MSG_STREAM_PUSH_CHANGE_PREFIX = "WVP_MSG_STREAM_PUSH_CHANGE";

    public static final String WVP_MSG_STREAM_PUSH_CHANGE_PREFIX = "WVP_MSG_STREAM_CHANGE_";

}


 src/main/java/com/genersoft/iot/vmp/conf/ThreadPoolTaskConfig.java

@@ -38,7 +38,7 @@
    /**
     * 线程池名前缀
     */
    private static final String threadNamePrefix = "wvp-sip-";
    private static final String threadNamePrefix = "wvp-";

    @Bean("taskExecutor") // bean的名称，默认为首字母小写的方法名
    public ThreadPoolTaskExecutor taskExecutor() {

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

@@ -9,17 +9,11 @@
import com.genersoft.iot.vmp.conf.MediaConfig;

import com.genersoft.iot.vmp.conf.UserSetup;

import com.genersoft.iot.vmp.gb28181.bean.Device;

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.OriginType;

import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;

import com.genersoft.iot.vmp.service.IMediaServerService;

import com.genersoft.iot.vmp.service.IMediaService;

import com.genersoft.iot.vmp.service.IStreamProxyService;

import com.genersoft.iot.vmp.media.zlm.dto.*;

import com.genersoft.iot.vmp.service.*;

import com.genersoft.iot.vmp.service.bean.SSRCInfo;

import com.genersoft.iot.vmp.storager.IRedisCatchStorage;

import com.genersoft.iot.vmp.storager.IVideoManagerStorager;

import com.genersoft.iot.vmp.service.IPlayService;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

@@ -66,6 +60,9 @@
    private IStreamProxyService streamProxyService;



    @Autowired

    private IStreamPushService streamPushService;



    @Autowired

    private IMediaService mediaService;



    @Autowired

@@ -84,6 +81,32 @@
    private MediaConfig mediaConfig;



    /**

     * 服务器定时上报时间，上报间隔可配置，默认10s上报一次

     *

     */

    @ResponseBody

    @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")

    public ResponseEntity<String> onServerKeepalive(@RequestBody JSONObject json){



        if (logger.isDebugEnabled()) {

            logger.debug("[ ZLM HOOK ]on_server_keepalive API调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");



        List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(ZLMHttpHookSubscribe.HookType.on_server_keepalive);

        if (subscribes != null  && subscribes.size() > 0) {

            for (ZLMHttpHookSubscribe.Event subscribe : subscribes) {

                subscribe.response(null, json);

            }

        }



        JSONObject ret = new JSONObject();

        ret.put("code", 0);

        ret.put("msg", "success");

        return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);

    }



    /**

     * 流量统计事件，播放器或推流器断开时并且耗用流量超过特定阈值时会触发此事件，阈值通过配置文件general.flowThreshold配置；此事件对回复不敏感。

     *  

     */

@@ -92,7 +115,7 @@
    public ResponseEntity<String> onFlowReport(@RequestBody JSONObject json){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_flow_report API调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_flow_report API调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");

        JSONObject ret = new JSONObject();

@@ -110,7 +133,7 @@
    public ResponseEntity<String> onHttpAccess(@RequestBody JSONObject json){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_http_access API 调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_http_access API 调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");

        JSONObject ret = new JSONObject();

@@ -130,7 +153,7 @@
    public ResponseEntity<String> onPlay(@RequestBody JSONObject json){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_play API调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_play API调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");

        ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_play, json);

@@ -155,7 +178,7 @@
    @PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8")

    public ResponseEntity<String> onPublish(@RequestBody JSONObject json) {



        logger.debug("ZLM HOOK on_publish API调用，参数：" + json.toString());

        logger.debug("[ ZLM HOOK ]on_publish API调用，参数：" + json.toString());



        String mediaServerId = json.getString("mediaServerId");

        ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, json);

@@ -191,7 +214,7 @@
    public ResponseEntity<String> onRecordMp4(@RequestBody JSONObject json){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_record_mp4 API调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_record_mp4 API调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");

        JSONObject ret = new JSONObject();

@@ -209,7 +232,7 @@
    public ResponseEntity<String> onRtspRealm(@RequestBody JSONObject json){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_rtsp_realm API调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_rtsp_realm API调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");

        JSONObject ret = new JSONObject();

@@ -228,7 +251,7 @@
    public ResponseEntity<String> onRtspAuth(@RequestBody JSONObject json){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_rtsp_auth API调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_rtsp_auth API调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");

        JSONObject ret = new JSONObject();

@@ -247,7 +270,7 @@
    public ResponseEntity<String> onShellLogin(@RequestBody JSONObject json){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_shell_login API调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_shell_login API调用，参数：" + json.toString());

        }

        // TODO 如果是带有rtpstream则开启按需拉流

        // String app = json.getString("app");

@@ -277,7 +300,7 @@
    public ResponseEntity<String> onStreamChanged(@RequestBody MediaItem item){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_stream_changed API调用，参数：" + JSONObject.toJSONString(item));

            logger.debug("[ ZLM HOOK ]on_stream_changed API调用，参数：" + JSONObject.toJSONString(item));

        }

        String mediaServerId = item.getMediaServerId();

        JSONObject json = (JSONObject) JSON.toJSON(item);

@@ -315,33 +338,39 @@
                }

            }else {

                if (!"rtp".equals(app)){



                    String type = OriginType.values()[item.getOriginType()].getType();

                    MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);

                    if (regist) {

                        StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks);

                        redisCatchStorage.addStream(mediaServerItem, OriginType.values()[item.getOriginType()].getType(), app, streamId, streamInfo);

                        redisCatchStorage.addStream(mediaServerItem, type, app, streamId, streamInfo);

                        if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()

                                || item.getOriginType() == OriginType.RTMP_PUSH.ordinal()

                                || item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {

                            zlmMediaListManager.addMedia(item);

                        }

                    }else {

                        // 兼容流注销时类型错误的问题，等zlm更新后删除

                        StreamPushItem streamPushItem = streamPushService.getPush(app, streamId);

                        if (streamPushItem != null) {

                            type = "PUSH";

                        }else {

                            StreamProxyItem streamProxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(app, streamId);

                            if (streamProxyByAppAndStream != null) {

                                type = "PULL";

                            }

                        }

                        zlmMediaListManager.removeMedia( app, streamId);

                        redisCatchStorage.removeStream(mediaServerItem, OriginType.values()[item.getOriginType()].getType(), app, streamId);

                    }



                    }

                    if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()

                            || item.getOriginType() == OriginType.RTMP_PUSH.ordinal()

                            || item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {

                        // 发送流变化redis消息

                        JSONObject jsonObject = new JSONObject();

                        jsonObject.put("serverId", userSetup.getServerId());

                        jsonObject.put("app", app);

                        jsonObject.put("stream", streamId);

                        jsonObject.put("register", regist);

                        jsonObject.put("mediaServerId", mediaServerId);

                        redisCatchStorage.sendStreamChangeMsg(jsonObject);

                    }

                    // 发送流变化redis消息

                    JSONObject jsonObject = new JSONObject();

                    jsonObject.put("serverId", userSetup.getServerId());

                    jsonObject.put("app", app);

                    jsonObject.put("stream", streamId);

                    jsonObject.put("register", regist);

                    jsonObject.put("mediaServerId", mediaServerId);

                    redisCatchStorage.sendStreamChangeMsg(type, jsonObject);

                }

            }

        }

@@ -361,7 +390,7 @@
    public ResponseEntity<String> onStreamNoneReader(@RequestBody JSONObject json){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_stream_none_reader API调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_stream_none_reader API调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");

        String streamId = json.getString("stream");

@@ -421,7 +450,7 @@
    @PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8")

    public ResponseEntity<String> onStreamNotFound(@RequestBody JSONObject json){

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_stream_not_found API调用，参数：" + json.toString());

            logger.debug("[ ZLM HOOK ]on_stream_not_found API调用，参数：" + json.toString());

        }

        String mediaServerId = json.getString("mediaServerId");

        MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);

@@ -468,7 +497,7 @@
    public ResponseEntity<String> onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){

        

        if (logger.isDebugEnabled()) {

            logger.debug("ZLM HOOK on_server_started API调用，参数：" + jsonObject.toString());

            logger.debug("[ ZLM HOOK ]on_server_started API调用，参数：" + jsonObject.toString());

        }

        String remoteAddr = request.getRemoteAddr();

        jsonObject.put("ip", remoteAddr);


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

@@ -27,7 +27,8 @@
        on_stream_changed,
        on_stream_none_reader,
        on_stream_not_found,
        on_server_started
        on_server_started,
        on_server_keepalive
    }

    public interface Event{

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

@@ -3,7 +3,6 @@
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -11,9 +10,11 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.*;
@@ -41,8 +42,13 @@
    @Autowired
    private MediaConfig mediaConfig;

    @Qualifier("taskExecutor")
    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    @Override
    public void run(String... strings) throws Exception {
        mediaServerService.clearMediaServerForOnline();
        if (mediaServerService.getDefaultMediaServer() == null) {
            mediaServerService.addToDatabase(mediaConfig.getMediaSerItem());
        }
@@ -59,6 +65,9 @@
            }
        });

        // TODO 订阅 zlm保活事件, 当zlm离线时做业务的处理


        // 获取zlm信息
        logger.info("等待默认zlm接入...");

@@ -70,7 +79,9 @@
        for (MediaServerItem mediaServerItem : all) {
            if (startGetMedia == null) startGetMedia = new HashMap<>();
            startGetMedia.put(mediaServerItem.getId(), true);
            connectZlmServer(mediaServerItem);
            taskExecutor.execute(()->{
                connectZlmServer(mediaServerItem);
            });
        }
        Timer timer = new Timer();
        // 2分钟后未连接到则不再去主动连接, TODO 并对重启前使用此在zlm的通道发送bye

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

@@ -1,15 +1,16 @@
package com.genersoft.iot.vmp.media.zlm.dto;

public enum OriginType {
    // 不可调整顺序
    UNKNOWN("UNKNOWN"),
    RTMP_PUSH("PUSH"),
    RTSP_PUSH("PUSH"),
    RTP_PUSH("RTP"),
    RTC_PUSH("PUSH"),
    PULL("PULL"),
    FFMPEG_PULL("PULL"),
    MP4_VOD("MP4_VOD"),
    DEVICE_CHN("DEVICE_CHN");
    DEVICE_CHN("DEVICE_CHN"),
    RTC_PUSH("PUSH");

    private final String type;
    OriginType(String type) {

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

@@ -35,4 +35,6 @@
    PageInfo<StreamPushItem> getPushList(Integer page, Integer count);

    StreamPushItem transform(MediaItem item);

    StreamPushItem getPush(String app, String streamId);
}

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

@@ -310,7 +310,7 @@
     */
    @Override
    public void handLeZLMServerConfig(ZLMServerConfig zlmServerConfig) {
        logger.info("[ {} ]-[ {}:{} ]已连接",
        logger.info("[ ZLM：{} ]-[ {}:{} ]已连接",
                zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort());

        MediaServerItem serverItem = mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId());
@@ -469,7 +469,7 @@
     */
    @Override
    public void setZLMConfig(MediaServerItem mediaServerItem) {
        logger.info("[ {} ]-[ {}:{} ]设置zlm",
        logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm",
                mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
        String protocol = sslEnabled ? "https" : "http";
        String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
@@ -494,16 +494,17 @@
        param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrex));
        param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
        param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
        param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
        param.put("hook.timeoutSec","20");
        param.put("general.streamNoneReaderDelayMS","-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() );

        JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param);

        if (responseJSON != null && responseJSON.getInteger("code") == 0) {
            logger.info("[ {} ]-[ {}:{} ]设置zlm成功",
            logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm成功",
                    mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
        }else {
            logger.info("[ {} ]-[ {}:{} ]设置zlm失败",
            logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm失败",
                    mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
        }
    }

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

@@ -109,4 +109,11 @@
        }
        return del > 0;
    }


    @Override
    public StreamPushItem getPush(String app, String streamId) {

        return streamPushMapper.selectOne(app, streamId);
    }
}

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

@@ -127,7 +127,7 @@
     * 发送推流生成与推流消失消息
     * @param jsonObject 消息内容
     */
    void sendStreamChangeMsg(JSONObject jsonObject);
    void sendStreamChangeMsg(String type, JSONObject jsonObject);

    /**
     * 添加流信息到redis

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

@@ -9,6 +9,8 @@
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@@ -18,6 +20,8 @@
@SuppressWarnings("rawtypes")
@Component
public class RedisCatchStorageImpl implements IRedisCatchStorage {

    private Logger logger = LoggerFactory.getLogger(RedisCatchStorageImpl.class);

    @Autowired
    private RedisUtil redis;
@@ -311,8 +315,9 @@
    }

    @Override
    public void sendStreamChangeMsg(JSONObject jsonObject) {
        String key = VideoManagerConstants.WVP_MSG_STREAM_PUSH_CHANGE_PREFIX;
    public void sendStreamChangeMsg(String type, JSONObject jsonObject) {
        String key = VideoManagerConstants.WVP_MSG_STREAM_PUSH_CHANGE_PREFIX + type;
        logger.debug("[redis 流变化事件] {}: {}", key, jsonObject.toString());
        redis.convertAndSend(key, jsonObject);
    }