From 098dd8a04515c9dbb143ab39678c2a71a03ec427 Mon Sep 17 00:00:00 2001 From: lin <18010473990@163.com> Date: Sat, 8 Jan 2022 18:00:31 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=8E=A8=E6=B5=81=E7=BB=93?= =?UTF-8?q?=E6=9D=9F=E6=97=B6=E6=B5=81=E7=B1=BB=E5=9E=8B=E7=9A=84=E8=8E=B7?= =?UTF-8?q?=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../vmp/media/zlm/ZLMHttpHookListener.java | 16 +++------- .../service/impl/StreamProxyServiceImpl.java | 12 +++---- .../service/impl/StreamPushServiceImpl.java | 32 +++++++++---------- .../iot/vmp/storager/IRedisCatchStorage.java | 8 +++-- .../storager/impl/RedisCatchStorageImpl.java | 28 ++++++++++++---- 5 files changed, 54 insertions(+), 42 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java index 118b2afb..c1f473df 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -343,24 +343,16 @@ public class ZLMHttpHookListener { 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); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java index 8ec9474b..1efc6145 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java @@ -279,18 +279,18 @@ public class StreamProxyServiceImpl implements IStreamProxyService { String type = "PULL"; // 发送redis消息 - List streamInfoList = redisCatchStorage.getStreams(mediaServerId, type); - if (streamInfoList.size() > 0) { - for (StreamInfo streamInfo : streamInfoList) { + List 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()); } } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java index 2a707541..3fac37ad 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java @@ -173,16 +173,16 @@ public class StreamPushServiceImpl implements IStreamPushService { List pushList = getPushList(mediaServerId); Map pushItemMap = new HashMap<>(); // redis记录 - List streamInfoPushList = redisCatchStorage.getStreams(mediaServerId, "PUSH"); - Map streamInfoPushItemMap = new HashMap<>(); + List mediaItems = redisCatchStorage.getStreams(mediaServerId, "PUSH"); + Map 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 @@ public class StreamPushServiceImpl implements IStreamPushService { } } - Collection offlineStreamInfoItems = streamInfoPushItemMap.values(); - if (offlineStreamInfoItems.size() > 0) { + Collection 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 @@ public class StreamPushServiceImpl implements IStreamPushService { // 发送流停止消息 String type = "PUSH"; // 发送redis消息 - List streamInfoList = redisCatchStorage.getStreams(mediaServerId, type); + List 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); diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index f7b16a84..8dc21ff3 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -3,7 +3,9 @@ package com.genersoft.iot.vmp.storager; 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 @@ public interface IRedisCatchStorage { * @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 @@ public interface IRedisCatchStorage { */ ThirdPartyGB queryMemberNoGBId(String queryKey); - List getStreams(String mediaServerId, String pull); + List getStreams(String mediaServerId, String pull); /** * 将device信息写入redis @@ -206,4 +208,6 @@ public interface IRedisCatchStorage { SubscribeInfo getSubscribe(String key); void delSubscribe(String key); + + MediaItem getStreamInfo(String app, String streamId, String mediaServerId); } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java index 2dcc2171..1d689434 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java @@ -5,7 +5,9 @@ import com.genersoft.iot.vmp.common.StreamInfo; 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 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { } @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 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { } @Override - public List getStreams(String mediaServerId, String type) { - List result = new ArrayList<>(); + public List getStreams(String mediaServerId, String type) { + List result = new ArrayList<>(); String key = VideoManagerConstants.WVP_SERVER_STREAM_PREFIX + userSetup.getServerId() + "_" + type + "_*_*_" + mediaServerId; List 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 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { 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 keys = redis.scan(scanKey); + if (keys.size() > 0) { + String key = (String) keys.get(0); + result = (MediaItem)redis.get(key); + } + + return result; + } }