From 764d04b497356ba6bcbb75fd42b51eca750f7223 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 29 五月 2024 15:02:51 +0800
Subject: [PATCH] 调整上级观看消息的发送

---
 src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java |  138 ++++++++++++++++++++++++++++++++++++----------
 1 files changed, 108 insertions(+), 30 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java
index 5e91151..a0b3341 100755
--- a/src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java
@@ -7,22 +7,21 @@
 import com.genersoft.iot.vmp.conf.MediaConfig;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.conf.exception.ControllerException;
-import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType;
-import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
 import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
 import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
 import com.genersoft.iot.vmp.media.bean.MediaInfo;
+import com.genersoft.iot.vmp.media.bean.MediaServer;
 import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
 import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
 import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerChangeEvent;
 import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerDeleteEvent;
 import com.genersoft.iot.vmp.media.service.IMediaNodeServerService;
 import com.genersoft.iot.vmp.media.service.IMediaServerService;
-import com.genersoft.iot.vmp.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
 import com.genersoft.iot.vmp.service.IInviteStreamService;
 import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
-import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.dao.MediaServerMapper;
@@ -83,6 +82,9 @@
     @Autowired
     private MediaConfig mediaConfig;
 
+    @Autowired
+    private SendRtpPortManager sendRtpPortManager;
+
 
 
     /**
@@ -94,6 +96,8 @@
         if ("rtsp".equals(event.getSchema())) {
             logger.info("娴佸彉鍖栵細娉ㄥ唽 app->{}, stream->{}", event.getApp(), event.getStream());
             addCount(event.getMediaServer().getId());
+            String type = OriginType.values()[event.getMediaInfo().getOriginType()].getType();
+            redisCatchStorage.addStream(event.getMediaServer(), type, event.getApp(), event.getStream(), event.getMediaInfo());
         }
     }
 
@@ -106,7 +110,15 @@
         if ("rtsp".equals(event.getSchema())) {
             logger.info("娴佸彉鍖栵細娉ㄩ攢, app->{}, stream->{}", event.getApp(), event.getStream());
             removeCount(event.getMediaServer().getId());
+            MediaInfo mediaInfo = redisCatchStorage.getStreamInfo(
+                    event.getApp(), event.getStream(), event.getMediaServer().getId());
+            if (mediaInfo == null) {
+                return;
+            }
+            String type = OriginType.values()[mediaInfo.getOriginType()].getType();
+            redisCatchStorage.removeStream(mediaInfo.getMediaServer().getId(), type, event.getApp(), event.getStream());
         }
+
     }
 
 
@@ -125,7 +137,7 @@
                 ssrcFactory.initMediaServerSSRC(mediaServer.getId(), null);
             }
             // 鏌ヨredis鏄惁瀛樺湪姝ediaServer
-            String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServer.getId();
+            String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + ":" + mediaServer.getId();
             Boolean hasKey = redisTemplate.hasKey(key);
             if (hasKey != null && ! hasKey) {
                 redisTemplate.opsForValue().set(key, mediaServer);
@@ -261,7 +273,7 @@
         if (mediaServerInRedis == null || !ssrcFactory.hasMediaServerSSRC(mediaServerInDataBase.getId())) {
             ssrcFactory.initMediaServerSSRC(mediaServerInDataBase.getId(),null);
         }
-        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerInDataBase.getId();
+        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + ":" + mediaServerInDataBase.getId();
         redisTemplate.opsForValue().set(key, mediaServerInDataBase);
         if (mediaServerInDataBase.isStatus()) {
             resetOnlineServerItem(mediaServerInDataBase);
@@ -277,8 +289,8 @@
     @Override
     public List<MediaServer> getAllOnlineList() {
         List<MediaServer> result = new ArrayList<>();
-        List<Object> mediaServerKeys = RedisUtil.scan(redisTemplate, String.format("%S*", VideoManagerConstants.MEDIA_SERVER_PREFIX+ userSetting.getServerId() + "_" ));
-        String onlineKey = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
+        List<Object> mediaServerKeys = RedisUtil.scan(redisTemplate, String.format("%S*", VideoManagerConstants.MEDIA_SERVER_PREFIX+ userSetting.getServerId() + ":" ));
+        String onlineKey = VideoManagerConstants.ONLINE_MEDIA_SERVERS_PREFIX + userSetting.getServerId();
         for (Object mediaServerKey : mediaServerKeys) {
             String key = (String) mediaServerKey;
             MediaServer mediaServer = JsonUtil.redisJsonToObject(redisTemplate, key, MediaServer.class);
@@ -326,14 +338,14 @@
 
     @Override
     public List<MediaServer> getAllOnline() {
-        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
+        String key = VideoManagerConstants.ONLINE_MEDIA_SERVERS_PREFIX + userSetting.getServerId();
         Set<Object> mediaServerIdSet = redisTemplate.opsForZSet().reverseRange(key, 0, -1);
 
         List<MediaServer> result = new ArrayList<>();
         if (mediaServerIdSet != null && mediaServerIdSet.size() > 0) {
             for (Object mediaServerId : mediaServerIdSet) {
                 String mediaServerIdStr = (String) mediaServerId;
-                String serverKey = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerIdStr;
+                String serverKey = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + ":" + mediaServerIdStr;
                 result.add((MediaServer) redisTemplate.opsForValue().get(serverKey));
             }
         }
@@ -351,7 +363,7 @@
         if (mediaServerId == null) {
             return null;
         }
-        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;
+        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + ":" + mediaServerId;
         return JsonUtil.redisJsonToObject(redisTemplate, key, MediaServer.class);
     }
 
@@ -363,7 +375,7 @@
 
     @Override
     public void clearMediaServerForOnline() {
-        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
+        String key = VideoManagerConstants.ONLINE_MEDIA_SERVERS_PREFIX + userSetting.getServerId();
         redisTemplate.delete(key);
     }
 
@@ -387,6 +399,7 @@
             logger.info("[娣诲姞濯掍綋鑺傜偣] 澶辫触, mediaServer鐨勭被鍨嬶細 {}锛屾湭鎵惧埌瀵瑰簲鐨勫疄鐜扮被", mediaServer.getType());
             return;
         }
+
         mediaServerMapper.add(mediaServer);
         if (mediaServer.isStatus()) {
             mediaNodeServerService.online(mediaServer);
@@ -401,7 +414,7 @@
     @Override
     public void resetOnlineServerItem(MediaServer serverItem) {
         // 鏇存柊缂撳瓨
-        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
+        String key = VideoManagerConstants.ONLINE_MEDIA_SERVERS_PREFIX + userSetting.getServerId();
         // 浣跨敤zset鐨勫垎鏁颁綔涓哄綋鍓嶅苟鍙戦噺锛� 榛樿鍊艰缃负0
         if (redisTemplate.opsForZSet().score(key, serverItem.getId()) == null) {  // 涓嶅瓨鍦ㄥ垯璁剧疆榛樿鍊� 宸插瓨鍦ㄥ垯閲嶇疆
             redisTemplate.opsForZSet().add(key, serverItem.getId(), 0L);
@@ -424,14 +437,14 @@
         if (mediaServerId == null) {
             return;
         }
-        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
+        String key = VideoManagerConstants.ONLINE_MEDIA_SERVERS_PREFIX + userSetting.getServerId();
         redisTemplate.opsForZSet().incrementScore(key, mediaServerId, 1);
 
     }
 
     @Override
     public void removeCount(String mediaServerId) {
-        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
+        String key = VideoManagerConstants.ONLINE_MEDIA_SERVERS_PREFIX + userSetting.getServerId();
         redisTemplate.opsForZSet().incrementScore(key, mediaServerId, - 1);
     }
 
@@ -441,7 +454,7 @@
      */
     @Override
     public MediaServer getMediaServerForMinimumLoad(Boolean hasAssist) {
-        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
+        String key = VideoManagerConstants.ONLINE_MEDIA_SERVERS_PREFIX + userSetting.getServerId();
         Long size = redisTemplate.opsForZSet().zCard(key);
         if (size  == null || size == 0) {
             logger.info("鑾峰彇璐熻浇鏈�浣庣殑鑺傜偣鏃舵棤鍦ㄧ嚎鑺傜偣");
@@ -520,8 +533,8 @@
     @Override
     public void delete(String id) {
         mediaServerMapper.delOne(id);
-        redisTemplate.opsForZSet().remove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), id);
-        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + id;
+        redisTemplate.opsForZSet().remove(VideoManagerConstants.ONLINE_MEDIA_SERVERS_PREFIX + userSetting.getServerId(), id);
+        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + ":" + id;
         redisTemplate.delete(key);
         // 鍙戦�佽妭鐐圭Щ闄ら�氱煡
         MediaServerDeleteEvent event = new MediaServerDeleteEvent(this);
@@ -577,6 +590,16 @@
             return false;
         }
         return mediaNodeServerService.stopSendRtp(mediaInfo, app, stream, ssrc);
+    }
+
+    @Override
+    public boolean initStopSendRtp(MediaServer mediaInfo, String app, String stream, String ssrc) {
+        IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaInfo.getType());
+        if (mediaNodeServerService == null) {
+            logger.info("[stopSendRtp] 澶辫触, mediaServer鐨勭被鍨嬶細 {}锛屾湭鎵惧埌瀵瑰簲鐨勫疄鐜扮被", mediaInfo.getType());
+            return false;
+        }
+        return mediaNodeServerService.initStopSendRtp(mediaInfo, app, stream, ssrc);
     }
 
     @Override
@@ -801,18 +824,17 @@
     }
 
     @Override
-    public void startSendRtpPassive(MediaServer mediaServer, ParentPlatform platform, SendRtpItem sendRtpItem, Integer timeout) {
+    public void startSendRtpPassive(MediaServer mediaServer, SendRtpItem sendRtpItem, Integer timeout) {
         IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
         if (mediaNodeServerService == null) {
             logger.info("[startSendRtpPassive] 澶辫触, mediaServer鐨勭被鍨嬶細 {}锛屾湭鎵惧埌瀵瑰簲鐨勫疄鐜扮被", mediaServer.getType());
             throw new ControllerException(ErrorCode.ERROR100.getCode(), "鏈壘鍒癿ediaServer瀵瑰簲鐨勫疄鐜扮被");
         }
         mediaNodeServerService.startSendRtpPassive(mediaServer, sendRtpItem, timeout);
-        sendPlatformStartPlayMsg(platform, sendRtpItem);
     }
 
     @Override
-    public void startSendRtpStream(MediaServer mediaServer, ParentPlatform platform, SendRtpItem sendRtpItem) {
+    public void startSendRtp(MediaServer mediaServer, SendRtpItem sendRtpItem) {
         IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
         if (mediaNodeServerService == null) {
             logger.info("[startSendRtpStream] 澶辫触, mediaServer鐨勭被鍨嬶細 {}锛屾湭鎵惧埌瀵瑰簲鐨勫疄鐜扮被", mediaServer.getType());
@@ -821,17 +843,73 @@
         logger.info("[寮�濮嬫帹娴乚 rtp/{}, 鐩爣={}:{}锛孲SRC={}, RTCP={}", sendRtpItem.getStream(),
                 sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isRtcp());
         mediaNodeServerService.startSendRtpStream(mediaServer, sendRtpItem);
-        sendPlatformStartPlayMsg(platform, sendRtpItem);
-
     }
 
-    private void sendPlatformStartPlayMsg(ParentPlatform platform, SendRtpItem sendRtpItem) {
-        if (sendRtpItem.getPlayType() == InviteStreamType.PUSH && platform  != null) {
-            MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0, sendRtpItem.getApp(), sendRtpItem.getStream(),
-                    sendRtpItem.getChannelId(), platform.getServerGBId(), platform.getName(), userSetting.getServerId(),
-                    sendRtpItem.getMediaServerId());
-            messageForPushChannel.setPlatFormIndex(platform.getId());
-            redisCatchStorage.sendPlatformStartPlayMsg(messageForPushChannel);
+    @Override
+    public SendRtpItem createSendRtpItem(MediaServer mediaServer, String ip, int port, String ssrc, String requesterId, String deviceId, String channelId, boolean isTcp, boolean rtcp) {
+        int localPort = sendRtpPortManager.getNextPort(mediaServer);
+        if (localPort == 0) {
+            return null;
         }
+        SendRtpItem sendRtpItem = new SendRtpItem();
+        sendRtpItem.setIp(ip);
+        sendRtpItem.setPort(port);
+        sendRtpItem.setSsrc(ssrc);
+        sendRtpItem.setPlatformId(deviceId);
+        sendRtpItem.setDeviceId(deviceId);
+        sendRtpItem.setChannelId(channelId);
+        sendRtpItem.setTcp(isTcp);
+        sendRtpItem.setRtcp(rtcp);
+        sendRtpItem.setApp("rtp");
+        sendRtpItem.setLocalPort(localPort);
+        sendRtpItem.setServerId(userSetting.getServerId());
+        sendRtpItem.setMediaServerId(mediaServer.getId());
+        return sendRtpItem;
+    }
+
+    @Override
+    public SendRtpItem createSendRtpItem(MediaServer serverItem, String ip, int port, String ssrc, String platformId,
+                                         String app, String stream, String channelId, boolean tcp, boolean rtcp){
+
+        int localPort = sendRtpPortManager.getNextPort(serverItem);
+        if (localPort == 0) {
+            return null;
+        }
+        SendRtpItem sendRtpItem = new SendRtpItem();
+        sendRtpItem.setIp(ip);
+        sendRtpItem.setPort(port);
+        sendRtpItem.setSsrc(ssrc);
+        sendRtpItem.setApp(app);
+        sendRtpItem.setStream(stream);
+        sendRtpItem.setPlatformId(platformId);
+        sendRtpItem.setChannelId(channelId);
+        sendRtpItem.setTcp(tcp);
+        sendRtpItem.setLocalPort(localPort);
+        sendRtpItem.setServerId(userSetting.getServerId());
+        sendRtpItem.setMediaServerId(serverItem.getId());
+        sendRtpItem.setRtcp(rtcp);
+        return sendRtpItem;
+    }
+
+    @Override
+    public MediaServer getMediaServerByAppAndStream(String app, String stream) {
+        List<MediaServer> mediaServerList = getAll();
+        for (MediaServer mediaServer : mediaServerList) {
+            MediaInfo mediaInfo = getMediaInfo(mediaServer, app, stream);
+            if (mediaInfo != null) {
+                return mediaServer;
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public Long updateDownloadProcess(MediaServer mediaServer, String app, String stream) {
+        IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
+        if (mediaNodeServerService == null) {
+            logger.info("[updateDownloadProcess] 澶辫触, mediaServer鐨勭被鍨嬶細 {}锛屾湭鎵惧埌瀵瑰簲鐨勫疄鐜扮被", mediaServer.getType());
+            throw new ControllerException(ErrorCode.ERROR100.getCode(), "鏈壘鍒癿ediaServer瀵瑰簲鐨勫疄鐜扮被");
+        }
+        return mediaNodeServerService.updateDownloadProcess(mediaServer, app, stream);
     }
 }

--
Gitblit v1.8.0