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

lin
1 parent abb60593
Showing 5 changed files with 54 additions and 42 deletions
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.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);
 						}
@@ -279,18 +279,18 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
         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());
             }
         }
     }
@@ -173,16 +173,16 @@ public class StreamPushServiceImpl implements IStreamPushService {
         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 @@ public class StreamPushServiceImpl implements IStreamPushService {
                 }
             }
-            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 @@ public class StreamPushServiceImpl implements IStreamPushService {
         // 发送流停止消息
         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);
@@ -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<StreamInfo> getStreams(String mediaServerId, String pull);
+    List<MediaItem> 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);
 }
@@ -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<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 @@ 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<Object> keys = redis.scan(scanKey);
+        if (keys.size() > 0) {
+            String key = (String) keys.get(0);
+            result = (MediaItem)redis.get(key);
+        }
+
+        return result;
+    }
 }