Merge pull request #658 from itouxin/patch-2

648540858

2022-11-03 fb1d9ca50ab23a5361ea66ab0527be31b21cfd89

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

@@ -4,6 +4,7 @@
import com.alibaba.fastjson.JSONObject;
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;
@@ -24,6 +25,8 @@
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;
@@ -33,6 +36,7 @@
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.net.InetAddress;
@@ -93,25 +97,21 @@


    @Override
    public WVPResult<StreamInfo> save(StreamProxyItem param) {
    public StreamInfo save(StreamProxyItem param) {
        MediaServerItem mediaInfo;
        WVPResult<StreamInfo> wvpResult = new WVPResult<>();
        wvpResult.setCode(0);
        if (param.getMediaServerId() == null || "auto".equals(param.getMediaServerId())){
        if (ObjectUtils.isEmpty(param.getMediaServerId()) || "auto".equals(param.getMediaServerId())){
            mediaInfo = mediaServerService.getMediaServerForMinimumLoad();
        }else {
            mediaInfo = mediaServerService.getOne(param.getMediaServerId());
        }
        if (mediaInfo == null) {
            logger.warn("保存代理未找到在线的ZLM...");
            wvpResult.setMsg("保存失败");
            return wvpResult;
            throw new ControllerException(ErrorCode.ERROR100.getCode(), "保存代理未找到在线的ZLM");
        }

        String dstUrl = String.format("rtmp://%s:%s/%s/%s", "127.0.0.1", mediaInfo.getRtmpPort(), param.getApp(),
                param.getStream() );
        param.setDst_url(dstUrl);
        StringBuffer result = new StringBuffer();
        StringBuffer resultMsg = new StringBuffer();
        boolean streamLive = false;
        param.setMediaServerId(mediaInfo.getId());
        boolean saveResult;
@@ -121,43 +121,47 @@
        }else { // 新增
            saveResult = addStreamProxy(param);
        }
        if (saveResult) {
            result.append("保存成功");
            if (param.isEnable()) {
                JSONObject jsonObject = addStreamProxyToZlm(param);
                if (jsonObject == null || jsonObject.getInteger("code") != 0) {
                    streamLive = false;
                    result.append(", 但是启用失败，请检查流地址是否可用");
                    param.setEnable(false);
                    // 直接移除
                    if (param.isEnable_remove_none_reader()) {
                        del(param.getApp(), param.getStream());
                    }else {
                        updateStreamProxy(param);
                    }

                }else {
                    streamLive = true;
                    StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
                            mediaInfo, param.getApp(), param.getStream(), null, null);
                    wvpResult.setData(streamInfo);

                }
            }
        }else {
            result.append("保存失败");
        if (!saveResult) {
            throw new ControllerException(ErrorCode.ERROR100.getCode(),"保存失败");
        }
        if ( !StringUtils.isEmpty(param.getPlatformGbId()) && streamLive) {
        StreamInfo resultForStreamInfo = null;
        resultMsg.append("保存成功");
        if (param.isEnable()) {
            JSONObject jsonObject = addStreamProxyToZlm(param);
            if (jsonObject == null || jsonObject.getInteger("code") != 0) {
                streamLive = false;
                resultMsg.append(", 但是启用失败，请检查流地址是否可用");
                param.setEnable(false);
                // 直接移除
                if (param.isEnable_remove_none_reader()) {
                    del(param.getApp(), param.getStream());
                }else {
                    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())){
                result.append(",  关联国标平台[ " + param.getPlatformGbId() + " ]成功");
                return resultForStreamInfo;
            }else {
                result.append(",  关联国标平台[ " + param.getPlatformGbId() + " ]失败");
                resultMsg.append(",  关联国标平台[ " + param.getPlatformGbId() + " ]失败");
                throw new ControllerException(ErrorCode.ERROR100.getCode(), resultMsg.toString());
            }
        }else {
            if (!streamLive) {
                throw new ControllerException(ErrorCode.ERROR100.getCode(), resultMsg.toString());
            }
        }
        wvpResult.setMsg(result.toString());
        return wvpResult;
        return resultForStreamInfo;
    }

    /**
@@ -174,7 +178,7 @@
        streamProxyItem.setCreateTime(now);
        try {
            if (streamProxyMapper.add(streamProxyItem) > 0) {
                if (!StringUtils.isEmpty(streamProxyItem.getGbId())) {
                if (!ObjectUtils.isEmpty(streamProxyItem.getGbId())) {
                    if (gbStreamMapper.add(streamProxyItem) < 0) {
                        //事务回滚
                        dataSourceTransactionManager.rollback(transactionStatus);
@@ -209,7 +213,7 @@
        streamProxyItem.setStreamType("proxy");
        try {
            if (streamProxyMapper.update(streamProxyItem) > 0) {
                if (!StringUtils.isEmpty(streamProxyItem.getGbId())) {
                if (!ObjectUtils.isEmpty(streamProxyItem.getGbId())) {
                    if (gbStreamMapper.updateByAppAndStream(streamProxyItem) == 0) {
                        //事务回滚
                        dataSourceTransactionManager.rollback(transactionStatus);
@@ -293,12 +297,11 @@
    public boolean start(String app, String stream) {
        boolean result = false;
        StreamProxyItem streamProxy = videoManagerStorager.queryStreamProxy(app, stream);
        if (!streamProxy.isEnable() ) {
        if (streamProxy != null && !streamProxy.isEnable() ) {
            JSONObject jsonObject = addStreamProxyToZlm(streamProxy);
            if (jsonObject == null) {
                return false;
            }
            System.out.println(jsonObject);
            if (jsonObject.getInteger("code") == 0) {
                result = true;
                streamProxy.setEnable(true);
@@ -424,7 +427,7 @@
                        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"))) {
                                if ("rtsp".equals(streamJSONObj.getString("schema"))) {
                                    StreamInfo streamInfo = new StreamInfo();
                                    String app = streamJSONObj.getString("app");
                                    String stream = streamJSONObj.getString("stream");
@@ -452,4 +455,9 @@
        }

    }

    @Override
    public ResourceBaceInfo getOverview() {
        return streamProxyMapper.getOverview();
    }
}