优化推流结束时流类型的获取

lin

2022-01-08 098dd8a04515c9dbb143ab39678c2a71a03ec427

优化推流结束时流类型的获取

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

@@ -343,24 +343,16 @@
                    MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);

                    if (mediaServerItem != null){

                        if (regist) {

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

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

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

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

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

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

                                zlmMediaListManager.addPush(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";

                                }

                            }

                            // 兼容流注销时类型从redis记录获取

                            MediaItem mediaItem = redisCatchStorage.getStreamInfo(app, streamId, mediaServerId);

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

                            zlmMediaListManager.removeMedia(app, streamId);

                            redisCatchStorage.removeStream(mediaServerItem.getId(), type, app, streamId);

                        }


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

@@ -279,18 +279,18 @@
        String type = "PULL";

        // 发送redis消息
        List<StreamInfo> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
        if (streamInfoList.size() > 0) {
            for (StreamInfo streamInfo : streamInfoList) {
        List<MediaItem> mediaItems = redisCatchStorage.getStreams(mediaServerId, type);
        if (mediaItems.size() > 0) {
            for (MediaItem mediaItem : mediaItems) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("serverId", userSetup.getServerId());
                jsonObject.put("app", streamInfo.getApp());
                jsonObject.put("stream", streamInfo.getStreamId());
                jsonObject.put("app", mediaItem.getApp());
                jsonObject.put("stream", mediaItem.getStream());
                jsonObject.put("register", false);
                jsonObject.put("mediaServerId", mediaServerId);
                redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
                // 移除redis内流的信息
                redisCatchStorage.removeStream(mediaServerId, type, streamInfo.getApp(), streamInfo.getStreamId());
                redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
            }
        }
    }

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

@@ -173,16 +173,16 @@
        List<StreamPushItem> pushList = getPushList(mediaServerId);
        Map<String, StreamPushItem> pushItemMap = new HashMap<>();
        // redis记录
        List<StreamInfo> streamInfoPushList = redisCatchStorage.getStreams(mediaServerId, "PUSH");
        Map<String, StreamInfo> streamInfoPushItemMap = new HashMap<>();
        List<MediaItem> mediaItems = redisCatchStorage.getStreams(mediaServerId, "PUSH");
        Map<String, MediaItem> streamInfoPushItemMap = new HashMap<>();
        if (pushList.size() > 0) {
            for (StreamPushItem streamPushItem : pushList) {
                pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem);
            }
        }
        if (streamInfoPushList.size() > 0) {
            for (StreamInfo streamInfo : streamInfoPushList) {
                streamInfoPushItemMap.put(streamInfo.getApp() + streamInfo.getStreamId(), streamInfo);
        if (mediaItems.size() > 0) {
            for (MediaItem mediaItem : mediaItems) {
                streamInfoPushItemMap.put(mediaItem.getApp() + mediaItem.getStream(), mediaItem);
            }
        }
        zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
@@ -221,19 +221,19 @@
                }

            }
            Collection<StreamInfo> offlineStreamInfoItems = streamInfoPushItemMap.values();
            if (offlineStreamInfoItems.size() > 0) {
            Collection<MediaItem> offlineMediaItemList = streamInfoPushItemMap.values();
            if (offlineMediaItemList.size() > 0) {
                String type = "PUSH";
                for (StreamInfo offlineStreamInfoItem : offlineStreamInfoItems) {
                for (MediaItem offlineMediaItem : offlineMediaItemList) {
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("serverId", userSetup.getServerId());
                    jsonObject.put("app", offlineStreamInfoItem.getApp());
                    jsonObject.put("stream", offlineStreamInfoItem.getStreamId());
                    jsonObject.put("app", offlineMediaItem.getApp());
                    jsonObject.put("stream", offlineMediaItem.getStream());
                    jsonObject.put("register", false);
                    jsonObject.put("mediaServerId", mediaServerId);
                    redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
                    // 移除redis内流的信息
                    redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineStreamInfoItem.getApp(), offlineStreamInfoItem.getStreamId());
                    redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineMediaItem.getApp(), offlineMediaItem.getStream());
                }
            }
        }));
@@ -250,15 +250,15 @@
        // 发送流停止消息
        String type = "PUSH";
        // 发送redis消息
        List<StreamInfo> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
        List<MediaItem> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
        if (streamInfoList.size() > 0) {
            for (StreamInfo streamInfo : streamInfoList) {
            for (MediaItem mediaItem : streamInfoList) {
                // 移除redis内流的信息
                redisCatchStorage.removeStream(mediaServerId, type, streamInfo.getApp(), streamInfo.getStreamId());
                redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("serverId", userSetup.getServerId());
                jsonObject.put("app", streamInfo.getApp());
                jsonObject.put("stream", streamInfo.getStreamId());
                jsonObject.put("app", mediaItem.getApp());
                jsonObject.put("stream", mediaItem.getStream());
                jsonObject.put("register", false);
                jsonObject.put("mediaServerId", mediaServerId);
                redisCatchStorage.sendStreamChangeMsg(type, jsonObject);

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

@@ -3,7 +3,9 @@
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.*;
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.StreamPushItem;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;

@@ -143,7 +145,7 @@
     * @param app
     * @param streamId
     */
    void addStream(MediaServerItem mediaServerItem, String type, String app, String streamId, StreamInfo streamInfo);
    void addStream(MediaServerItem mediaServerItem, String type, String app, String streamId, MediaItem item);

    /**
     * 移除流信息从redis
@@ -175,7 +177,7 @@
     */
    ThirdPartyGB queryMemberNoGBId(String queryKey);

    List<StreamInfo> getStreams(String mediaServerId, String pull);
    List<MediaItem> getStreams(String mediaServerId, String pull);

    /**
     * 将device信息写入redis
@@ -206,4 +208,6 @@
    SubscribeInfo getSubscribe(String key);

    void delSubscribe(String key);

    MediaItem getStreamInfo(String app, String streamId, String mediaServerId);
}

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

@@ -5,7 +5,9 @@
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.*;
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.StreamPushItem;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
@@ -386,9 +388,9 @@
    }

    @Override
    public void addStream(MediaServerItem mediaServerItem, String type, String app, String streamId, StreamInfo streamInfo) {
    public void addStream(MediaServerItem mediaServerItem, String type, String app, String streamId, MediaItem mediaItem) {
        String key = VideoManagerConstants.WVP_SERVER_STREAM_PREFIX  + userSetup.getServerId() + "_" + type + "_" + app + "_" + streamId + "_" + mediaServerItem.getId();
        redis.set(key, streamInfo);
        redis.set(key, mediaItem);
    }

    @Override
@@ -421,13 +423,13 @@
    }

    @Override
    public List<StreamInfo> getStreams(String mediaServerId, String type) {
        List<StreamInfo> result = new ArrayList<>();
    public List<MediaItem> getStreams(String mediaServerId, String type) {
        List<MediaItem> result = new ArrayList<>();
        String key = VideoManagerConstants.WVP_SERVER_STREAM_PREFIX + userSetup.getServerId() + "_" + type + "_*_*_" + mediaServerId;
        List<Object> streams = redis.scan(key);
        for (Object stream : streams) {
            StreamInfo streamInfo = (StreamInfo)redis.get((String) stream);
            result.add(streamInfo);
            MediaItem mediaItem = (MediaItem)redis.get((String) stream);
            result.add(mediaItem);
        }
        return result;
    }
@@ -492,4 +494,18 @@

        return result;
    }

    @Override
    public MediaItem getStreamInfo(String app, String streamId, String mediaServerId) {
        String scanKey = VideoManagerConstants.WVP_SERVER_STREAM_PREFIX  + userSetup.getServerId() + "_*_" + app + "_" + streamId + "_" + mediaServerId;

        MediaItem result = null;
        List<Object> keys = redis.scan(scanKey);
        if (keys.size() > 0) {
            String key = (String) keys.get(0);
            result = (MediaItem)redis.get(key);
        }

        return result;
    }
}