Merge pull request #689 from AlphaWu/Zafu-Dev-1129

648540858

2022-12-01 b96d580168ed877fc4cf24f3208066d6fea6e114

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

@@ -5,29 +5,25 @@
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.TreeType;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
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.StreamProxyItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.service.IGbStreamService;
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.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.StreamProxyMapper;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaceInfo;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.github.pagehelper.PageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,10 +33,10 @@
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.net.InetAddress;
import java.util.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 视频代理业务
@@ -112,7 +108,6 @@
                param.getStream() );
        param.setDst_url(dstUrl);
        StringBuffer resultMsg = new StringBuffer();
        boolean streamLive = false;
        param.setMediaServerId(mediaInfo.getId());
        boolean saveResult;
        // 更新
@@ -129,7 +124,6 @@
        if (param.isEnable()) {
            JSONObject jsonObject = addStreamProxyToZlm(param);
            if (jsonObject == null || jsonObject.getInteger("code") != 0) {
                streamLive = false;
                resultMsg.append(", 但是启用失败，请检查流地址是否可用");
                param.setEnable(false);
                // 直接移除
@@ -139,26 +133,10 @@
                    updateStreamProxy(param);
                }


            }else {
                streamLive = true;
                resultForStreamInfo = mediaService.getStreamInfoByAppAndStream(
                        mediaInfo, param.getApp(), param.getStream(), null, null);

            }
        }
        if ( !ObjectUtils.isEmpty(param.getPlatformGbId()) && streamLive) {
            List<GbStream> gbStreams = new ArrayList<>();
            gbStreams.add(param);
            if (gbStreamService.addPlatformInfo(gbStreams, param.getPlatformGbId(), param.getCatalogId())){
                return resultForStreamInfo;
            }else {
                resultMsg.append(",  关联国标平台[ " + param.getPlatformGbId() + " ]失败");
                throw new ControllerException(ErrorCode.ERROR100.getCode(), resultMsg.toString());
            }
        }else {
            if (!streamLive) {
                throw new ControllerException(ErrorCode.ERROR100.getCode(), resultMsg.toString());
            }
        }
        return resultForStreamInfo;
@@ -250,10 +228,10 @@
        }
        if ("default".equals(param.getType())){
            result = zlmresTfulUtils.addStreamProxy(mediaServerItem, param.getApp(), param.getStream(), param.getUrl(),
                    param.isEnable_hls(), param.isEnable_mp4(), param.getRtp_type());
                    param.isEnable_audio(), param.isEnable_mp4(), param.getRtp_type());
        }else if ("ffmpeg".equals(param.getType())) {
            result = zlmresTfulUtils.addFFmpegSource(mediaServerItem, param.getSrc_url(), param.getDst_url(),
                    param.getTimeout_ms() + "", param.isEnable_hls(), param.isEnable_mp4(),
                    param.getTimeout_ms() + "", param.isEnable_audio(), param.isEnable_mp4(),
                    param.getFfmpeg_cmd_key());
        }
        return result;
@@ -306,6 +284,9 @@
                result = true;
                streamProxy.setEnable(true);
                updateStreamProxy(streamProxy);
            }else {
                logger.info("启用代理失败： {}/{}->{}({})", app, stream, jsonObject.getString("msg"),
                        streamProxy.getSrc_url() == null? streamProxy.getUrl():streamProxy.getSrc_url());
            }
        }
        return result;
@@ -389,18 +370,18 @@
        String type = "PULL";

        // 发送redis消息
        List<MediaItem> mediaItems = redisCatchStorage.getStreams(mediaServerId, type);
        if (mediaItems.size() > 0) {
            for (MediaItem mediaItem : mediaItems) {
        List<OnStreamChangedHookParam> onStreamChangedHookParams = redisCatchStorage.getStreams(mediaServerId, type);
        if (onStreamChangedHookParams.size() > 0) {
            for (OnStreamChangedHookParam onStreamChangedHookParam : onStreamChangedHookParams) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("serverId", userSetting.getServerId());
                jsonObject.put("app", mediaItem.getApp());
                jsonObject.put("stream", mediaItem.getStream());
                jsonObject.put("app", onStreamChangedHookParam.getApp());
                jsonObject.put("stream", onStreamChangedHookParam.getStream());
                jsonObject.put("register", false);
                jsonObject.put("mediaServerId", mediaServerId);
                redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
                // 移除redis内流的信息
                redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
                redisCatchStorage.removeStream(mediaServerId, type, onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream());
            }
        }
    }
@@ -418,7 +399,7 @@
    private void syncPullStream(String mediaServerId){
        MediaServerItem mediaServer = mediaServerService.getOne(mediaServerId);
        if (mediaServer != null) {
            List<MediaItem> allPullStream = redisCatchStorage.getStreams(mediaServerId, "PULL");
            List<OnStreamChangedHookParam> allPullStream = redisCatchStorage.getStreams(mediaServerId, "PULL");
            if (allPullStream.size() > 0) {
                zlmresTfulUtils.getMediaList(mediaServer, jsonObject->{
                    Map<String, StreamInfo> stringStreamInfoMap = new HashMap<>();