使用阿里代码规范。规范代码写法

648540858

2022-05-06 5d901b5e3f033e8b04e53420d68626cbd87431c8

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

@@ -1,21 +1,28 @@
package com.genersoft.iot.vmp.media.zlm;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.StreamPushMapper;
import org.checkerframework.checker.units.qual.C;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Component
public class ZLMMediaListManager {
@@ -29,7 +36,7 @@
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private IVideoManagerStorager storager;
    private IVideoManagerStorage storager;

    @Autowired
    private GbStreamMapper gbStreamMapper;
@@ -41,25 +48,38 @@
    private IStreamPushService streamPushService;

    @Autowired
    private IStreamProxyService streamProxyService;

    @Autowired
    private StreamPushMapper streamPushMapper;

    @Autowired
    private ZLMHttpHookSubscribe subscribe;

    @Autowired
    private UserSetting userSetting;

    public void updateMediaList() {
    private Map<String, ChannelOnlineEvent> channelOnlineEvents = new ConcurrentHashMap<>();


    public void updateMediaList(MediaServerItem mediaServerItem) {
        storager.clearMediaList();

        // 使用异步的当时更新媒体流列表
        zlmresTfulUtils.getMediaList((mediaList ->{
            if (mediaList == null) return;
        zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
            if (mediaList == null) {
                return;
            }
            String dataStr = mediaList.getString("data");

            Integer code = mediaList.getInteger("code");
            Map<String, StreamPushItem> result = new HashMap<>();
            List<StreamPushItem> streamPushItems = null;
            // 获取所有的国标关联
            List<GbStream> gbStreams = gbStreamMapper.selectAll();
//            List<GbStream> gbStreams = gbStreamMapper.selectAllByMediaServerId(mediaServerItem.getId());
            if (code == 0 ) {
                if (dataStr != null) {
                    streamPushItems = streamPushService.handleJSON(dataStr);
                    streamPushItems = streamPushService.handleJSON(dataStr, mediaServerItem);
                }
            }else {
                logger.warn("更新视频流失败，错误code： " + code);
@@ -71,26 +91,96 @@
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("app", streamPushItem.getApp());
                    jsonObject.put("stream", streamPushItem.getStream());
                    subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_play,jsonObject,(response)->{
                        updateMedia(response.getString("app"), response.getString("stream"));
                    });
                    jsonObject.put("mediaServerId", mediaServerItem.getId());
                    subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_play,jsonObject,
                            (MediaServerItem mediaServerItemInuse, JSONObject response)->{
                                updateMedia(mediaServerItem, response.getString("app"), response.getString("stream"));
                            }
                    );
                }
            }
        }));

    }

    public void addMedia(String app, String streamId) {
    public void addMedia(MediaServerItem mediaServerItem, String app, String streamId) {
        //使用异步更新推流
        updateMedia(app, streamId);
        updateMedia(mediaServerItem, app, streamId);
    }

    public StreamPushItem addPush(MediaItem mediaItem) {
        // 查找此直播流是否存在redis预设gbId
        StreamPushItem transform = streamPushService.transform(mediaItem);
        // 从streamId取出查询关键值
        Pattern pattern = Pattern.compile(userSetting.getThirdPartyGBIdReg());
        Matcher matcher = pattern.matcher(mediaItem.getStream());// 指定要匹配的字符串
        String queryKey = null;
        if (matcher.find()) { //此处find（）每次被调用后，会偏移到下一个匹配
            queryKey = matcher.group();
        }
        if (queryKey != null) {
            ThirdPartyGB thirdPartyGB = redisCatchStorage.queryMemberNoGBId(queryKey);
            if (thirdPartyGB != null && !StringUtils.isEmpty(thirdPartyGB.getNationalStandardNo())) {
                transform.setGbId(thirdPartyGB.getNationalStandardNo());
                transform.setName(thirdPartyGB.getName());
            }
        }
        if (!StringUtils.isEmpty(transform.getGbId())) {
            // 如果这个国标ID已经给了其他推流且流已离线，则移除其他推流
            List<GbStream> gbStreams = gbStreamMapper.selectByGBId(transform.getGbId());
            if (gbStreams.size() > 0) {
                for (GbStream gbStream : gbStreams) {
                    // 出现使用相同国标Id的视频流时，使用新流替换旧流，
                    if (queryKey != null && gbStream.getApp().equals(mediaItem.getApp())) {
                        Matcher matcherForStream = pattern.matcher(gbStream.getStream());
                        String queryKeyForStream = null;
                        if (matcherForStream.find()) { //此处find（）每次被调用后，会偏移到下一个匹配
                            queryKeyForStream = matcherForStream.group();
                        }
                        if (queryKeyForStream == null || !queryKeyForStream.equals(queryKey)) {
                            // 此时不是同一个流
                            gbStreamMapper.del(gbStream.getApp(), gbStream.getStream());
                            if (!gbStream.isStatus()) {
                                streamPushMapper.del(gbStream.getApp(), gbStream.getStream());
                            }
                        }
                    }
                }
            }
            //            StreamProxyItem streamProxyItem = gbStreamMapper.selectOne(transform.getApp(), transform.getStream());
            List<GbStream> gbStreamList = gbStreamMapper.selectByGBId(transform.getGbId());
            if (gbStreamList != null && gbStreamList.size() == 1) {
                transform.setGbStreamId(gbStreamList.get(0).getGbStreamId());
                transform.setPlatformId(gbStreamList.get(0).getPlatformId());
                transform.setCatalogId(gbStreamList.get(0).getCatalogId());
                transform.setGbId(gbStreamList.get(0).getGbId());
                gbStreamMapper.update(transform);
                streamPushMapper.del(gbStreamList.get(0).getApp(), gbStreamList.get(0).getStream());
            }else {
                transform.setCreateStamp(System.currentTimeMillis());
                gbStreamMapper.add(transform);
            }
            if (transform != null) {
                if (channelOnlineEvents.get(transform.getGbId()) != null)  {
                    channelOnlineEvents.get(transform.getGbId()).run(transform.getApp(), transform.getStream());
                    channelOnlineEvents.remove(transform.getGbId());
                }
            }
        }


        storager.updateMedia(transform);
        return transform;
    }


    public void updateMedia(String app, String streamId) {
    public void updateMedia(MediaServerItem mediaServerItem, String app, String streamId) {
        //使用异步更新推流
        zlmresTfulUtils.getMediaList(app, streamId, "rtmp", json->{
        zlmresTfulUtils.getMediaList(mediaServerItem, app, streamId, "rtmp", json->{

            if (json == null) return;
            if (json == null) {
                return;
            }
            String dataStr = json.getString("data");

            Integer code = json.getInteger("code");
@@ -98,7 +188,7 @@
            List<StreamPushItem> streamPushItems = null;
            if (code == 0 ) {
                if (dataStr != null) {
                    streamPushItems = streamPushService.handleJSON(dataStr);
                    streamPushItems = streamPushService.handleJSON(dataStr, mediaServerItem);
                }
            }else {
                logger.warn("更新视频流失败，错误code： " + code);
@@ -111,42 +201,55 @@
    }


    public void removeMedia(String app, String streamId) {
    public int removeMedia(String app, String streamId) {
        // 查找是否关联了国标， 关联了不删除， 置为离线
        StreamProxyItem streamProxyItem = gbStreamMapper.selectOne(app, streamId);
        int result = 0;
        if (streamProxyItem == null) {
            storager.removeMedia(app, streamId);
            result = storager.removeMedia(app, streamId);
        }else {
            storager.mediaOutline(app, streamId);
            // TODO 暂不设置为离线
            result =storager.mediaOutline(app, streamId);
        }
        return result;
    }

    public void clearAllSessions() {
        logger.info("清空所有国标相关的session");
        JSONObject allSessionJSON = zlmresTfulUtils.getAllSession();
        ZLMServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
        HashSet<String> allLocalPorts = new HashSet();
        if (allSessionJSON.getInteger("code") == 0) {
            JSONArray data = allSessionJSON.getJSONArray("data");
            if (data.size() > 0) {
                for (int i = 0; i < data.size(); i++) {
                    JSONObject sessionJOSN = data.getJSONObject(i);
                    Integer local_port = sessionJOSN.getInteger("local_port");
                    if (!local_port.equals(Integer.valueOf(mediaInfo.getHttpPort())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getHttpSSLport())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getRtmpPort())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getRtspPort())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getRtspSSlport())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getHookOnFlowReport()))){
                        allLocalPorts.add(sessionJOSN.getInteger("local_port") + "");
                     }
                }
            }
        }
        if (allLocalPorts.size() > 0) {
            List<String> result = new ArrayList<>(allLocalPorts);
            String localPortSStr = String.join(",", result);
            zlmresTfulUtils.kickSessions(localPortSStr);
        }
    public void addChannelOnlineEventLister(String key, ChannelOnlineEvent callback) {
        this.channelOnlineEvents.put(key,callback);
    }

    public void removedChannelOnlineEventLister(String key) {
        this.channelOnlineEvents.remove(key);
    }



//    public void clearAllSessions() {
//        logger.info("清空所有国标相关的session");
//        JSONObject allSessionJSON = zlmresTfulUtils.getAllSession();
//        ZLMServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
//        HashSet<String> allLocalPorts = new HashSet();
//        if (allSessionJSON.getInteger("code") == 0) {
//            JSONArray data = allSessionJSON.getJSONArray("data");
//            if (data.size() > 0) {
//                for (int i = 0; i < data.size(); i++) {
//                    JSONObject sessionJOSN = data.getJSONObject(i);
//                    Integer local_port = sessionJOSN.getInteger("local_port");
//                    if (!local_port.equals(Integer.valueOf(mediaInfo.getHttpPort())) &&
//                        !local_port.equals(Integer.valueOf(mediaInfo.getHttpSSLport())) &&
//                        !local_port.equals(Integer.valueOf(mediaInfo.getRtmpPort())) &&
//                        !local_port.equals(Integer.valueOf(mediaInfo.getRtspPort())) &&
//                        !local_port.equals(Integer.valueOf(mediaInfo.getRtspSSlport())) &&
//                        !local_port.equals(Integer.valueOf(mediaInfo.getHookOnFlowReport()))){
//                        allLocalPorts.add(sessionJOSN.getInteger("local_port") + "");
//                     }
//                }
//            }
//        }
//        if (allLocalPorts.size() > 0) {
//            List<String> result = new ArrayList<>(allLocalPorts);
//            String localPortSStr = String.join(",", result);
//            zlmresTfulUtils.kickSessions(localPortSStr);
//        }
//    }
}