Merge pull request #2110 from q792602257/fix/ssrc_recycle

修复ssrc回收逻辑

src/main/java/com/genersoft/iot/vmp/gb28181/bean/InviteMessageInfo.java

@@ -10,6 +10,8 @@ public class InviteMessageInfo {
     private String sourceChannelId;
     private String sessionName;
     private String ssrc;
+    private String allocatedSsrc;
+    private String allocatedSsrcMediaServerId;
     private boolean tcp;
     private boolean tcpActive;
     private String callId;

src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpInfo.java

@@ -24,6 +24,11 @@ public class SendRtpInfo {
      */
     private String ssrc;
 
+    /**
+     * 从SSRC池中分配的SSRC
+     */
+    private String allocatedSsrc;
+
     /**
      * 目标平台或设备的编号
      */
@@ -247,4 +252,8 @@ public class SendRtpInfo {
             this.setPlayType("Play".equalsIgnoreCase(sessionName) ? InviteStreamType.PLAY : InviteStreamType.PLAYBACK);
         }
     }
+
+    public String getSsrcToRelease() {
+        return allocatedSsrc;
+    }
 }

src/main/java/com/genersoft/iot/vmp/gb28181/bean/SsrcTransaction.java

@@ -47,6 +47,11 @@ public class SsrcTransaction {
      */
     private String ssrc;
 
+    /**
+     * 从SSRC池中分配的SSRC
+     */
+    private String allocatedSsrc;
+
     /**
      * 事务信息
      */
@@ -88,4 +93,8 @@ public class SsrcTransaction {
 
     public SsrcTransaction() {
     }
+
+    public String getSsrcToRelease() {
+        return allocatedSsrc;
+    }
 }

src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java

@@ -268,7 +268,7 @@ public class DeviceServiceImpl implements IDeviceService, CommandLineRunner {
         List<SsrcTransaction> ssrcTransactions = sessionManager.getSsrcTransactionByDeviceId(device.getDeviceId());
         if (ssrcTransactions != null && !ssrcTransactions.isEmpty()) {
             for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
-                mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
+                mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrcToRelease());
                 receiveRtpServerService.closeRTPServerByMediaServerId(ssrcTransaction.getMediaServerId(), ssrcTransaction.getApp(), ssrcTransaction.getStream());
                 sessionManager.removeByCallId(ssrcTransaction.getCallId());
             }

src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlatformServiceImpl.java

@@ -132,7 +132,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                 }
                 sendRtpServerService.delete(sendRtpItem);
                 if (mediaServerItem != null) {
-                    ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
+                    ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrcToRelease());
                     boolean stopResult = mediaServerService.initStopSendRtp(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc());
                     if (stopResult) {
                         Platform platform = queryPlatformByServerGBId(sendRtpItem.getTargetId());
@@ -339,7 +339,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                 if (sendRtpItem != null && sendRtpItem.getApp().equals(event.getApp()) && sendRtpItem.isSendToPlatform()) {
                     Platform platform = platformMapper.getParentPlatByServerGBId(sendRtpItem.getTargetId());
                     CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId());
-                    ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
+                    ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrcToRelease());
                     try {
                         commanderForPlatform.streamByeCmd(platform, sendRtpItem, channel);
                     } catch (SipException | InvalidArgumentException | ParseException e) {
@@ -526,7 +526,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
         List<SendRtpInfo> sendRtpItems = sendRtpServerService.queryForPlatform(platformId);
         if (sendRtpItems != null && !sendRtpItems.isEmpty()) {
             for (SendRtpInfo sendRtpItem : sendRtpItems) {
-                ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
+                ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrcToRelease());
                 sendRtpServerService.delete(sendRtpItem);
                 MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
                 mediaServerService.stopSendRtp(mediaInfo, sendRtpItem.getApp(), sendRtpItem.getStream(), null);
@@ -647,7 +647,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                                 log.error("[点播超时]， 发送BYE失败 {}", e.getMessage());
                             } finally {
                                 timeoutCallback.run(1, "收流超时");
-                                mediaServerService.releaseSsrc(mediaServerItem.getId(), data.getSsrcInfo().getSsrc());
+                                mediaServerService.releaseSsrc(mediaServerItem.getId(), data.getSsrcInfo().getSsrcToRelease());
                                 receiveRtpServerService.closeRTPServer(mediaServerItem, data.getSsrcInfo().getApp(), data.getSsrcInfo().getStream());
                                 sessionManager.removeByStream(data.getSsrcInfo().getApp(), data.getSsrcInfo().getStream());
                             }
@@ -728,8 +728,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                     // ssrc检验
                     // 更新ssrc
                     log.info("[Invite 200OK] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
-                    // 释放ssrc
+                    releaseAllocatedSsrc(mediaServerItem, ssrcInfo);
-                    mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
                     Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInResponse);
                     if (!result) {
                         try {
@@ -739,7 +738,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                             log.error("[命令发送失败] 停止播放， 发送BYE: {}", e.getMessage());
                         } finally {
                             // 释放ssrc
-                            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
+                            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrcToRelease());
                             receiveRtpServerService.closeRTPServer(mediaServerItem, ssrcInfo.getApp(), ssrcInfo.getStream());
                             sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream());
 
@@ -751,6 +750,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                         }
                     }else {
                         ssrcInfo.setSsrc(ssrcInResponse);
+                        updateSsrcTransaction(ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInResponse, null);
                         inviteInfo.setSsrcInfo(ssrcInfo);
                         inviteInfo.setStream(ssrcInfo.getStream());
                         if (tcpMode == 2) {
@@ -764,7 +764,9 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                         inviteStreamService.updateInviteInfo(inviteInfo);
                     }
                 }else {
+                    releaseAllocatedSsrc(mediaServerItem, ssrcInfo);
                     ssrcInfo.setSsrc(ssrcInResponse);
+                    updateSsrcTransaction(ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInResponse, null);
                     inviteInfo.setSsrcInfo(ssrcInfo);
                     inviteInfo.setStream(ssrcInfo.getStream());
                     if (tcpMode == 2) {
@@ -782,6 +784,10 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                     // 单端口
                     // 重新订阅流上线
                     SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(ssrcInfo.getApp(), inviteInfo.getStream());
+                    if (ssrcTransaction == null) {
+                        return;
+                    }
+                    releaseAllocatedSsrc(mediaServerItem, ssrcInfo);
                     sessionManager.removeByStream(ssrcInfo.getApp(), inviteInfo.getStream());
                     inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse);
 
@@ -790,6 +796,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
                     ssrcTransaction.setApp(ssrcInfo.getApp());
                     ssrcTransaction.setStream(inviteInfo.getStream());
                     ssrcTransaction.setSsrc(ssrcInResponse);
+                    ssrcTransaction.setAllocatedSsrc(null);
                     ssrcTransaction.setMediaServerId(mediaServerItem.getId());
                     ssrcTransaction.setSipTransactionInfo(new SipTransactionInfo((SIPResponse) responseEvent.getResponse()));
                     ssrcTransaction.setType(inviteSessionType);
@@ -800,6 +807,24 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
         }
     }
 
+    private void releaseAllocatedSsrc(MediaServer mediaServerItem, SSRCInfo ssrcInfo) {
+        if (ssrcInfo == null || ssrcInfo.getAllocatedSsrc() == null) {
+            return;
+        }
+        mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getAllocatedSsrc());
+        ssrcInfo.setAllocatedSsrc(null);
+    }
+
+    private void updateSsrcTransaction(String app, String stream, String ssrc, String allocatedSsrc) {
+        SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(app, stream);
+        if (ssrcTransaction == null) {
+            return;
+        }
+        ssrcTransaction.setSsrc(ssrc);
+        ssrcTransaction.setAllocatedSsrc(allocatedSsrc);
+        sessionManager.put(ssrcTransaction);
+    }
+
 
     private void tcpActiveHandler(Platform platform, CommonGBChannel channel, String contentString,
                                   MediaServer mediaServerItem, int tcpMode, boolean ssrcCheck,
@@ -836,7 +861,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
             log.error("[TCP主动连接对方] serverGbId: {}, channelId: {}, 解析200OK的SDP信息失败", platform.getServerGBId(), channel.getGbDeviceId(), e);
             receiveRtpServerService.closeRTPServer(mediaServerItem, ssrcInfo.getApp(), ssrcInfo.getStream());
             // 释放ssrc
-            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
+            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrcToRelease());
             sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream());
 
             callback.run(InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(),
@@ -861,7 +886,7 @@ public class PlatformServiceImpl implements IPlatformService, CommandLineRunner
             InviteInfo inviteInfo = inviteStreamService.getInviteInfo(null, channel.getGbId(), stream);
             if (inviteInfo != null) {
                 // 释放ssrc
-                mediaServerService.releaseSsrc(mediaServerItem.getId(), inviteInfo.getSsrcInfo().getSsrc());
+                mediaServerService.releaseSsrc(mediaServerItem.getId(), inviteInfo.getSsrcInfo().getSsrcToRelease());
                 inviteStreamService.removeInviteInfo(inviteInfo);
             }
             sessionManager.removeByStream(app, stream);

src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlayServiceImpl.java

@@ -495,8 +495,15 @@ public class PlayServiceImpl implements IPlayService {
         try {
             sendRtpInfo = sendRtpServerService.createSendRtpInfo(mediaServerItem, null, null, playSsrc, device.getDeviceId(), MediaApp.GB28181_TALK, stream,
                     channel.getId(), true, false);
+            if (sendRtpInfo == null) {
+                ssrcFactory.releaseSsrc(mediaServerItem.getId(), playSsrc);
+                audioEvent.call("获取发流端口失败");
+                return;
+            }
+            sendRtpInfo.setAllocatedSsrc(playSsrc);
             sendRtpInfo.setPlayType(InviteStreamType.TALK);
         }catch (PlayException e) {
+            ssrcFactory.releaseSsrc(mediaServerItem.getId(), playSsrc);
             log.info("[语音对讲]开始 获取发流端口失败 deviceId: {}, channelId: {},", device.getDeviceId(), channel.getDeviceId());
             return;
         }
@@ -523,7 +530,7 @@ public class PlayServiceImpl implements IPlayService {
                 log.error("[语音对讲]超时， 发送BYE失败 {}", e.getMessage());
             } finally {
                 timeoutCallback.run();
-                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
+                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrcToRelease());
                 sessionManager.removeByStream(sendRtpInfo.getApp(), sendRtpInfo.getStream());
             }
         }, userSetting.getPlayTimeout());
@@ -532,13 +539,13 @@ public class PlayServiceImpl implements IPlayService {
             Integer localPort = mediaServerService.startSendRtpPassive(mediaServerItem, sendRtpInfo, userSetting.getPlayTimeout() * 1000);
             if (localPort == null || localPort <= 0) {
                 timeoutCallback.run();
-                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
+                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrcToRelease());
                 sessionManager.removeByStream(sendRtpInfo.getApp(), sendRtpInfo.getStream());
                 return;
             }
             sendRtpInfo.setPort(localPort);
         }catch (ControllerException e) {
-            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
+            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrcToRelease());
             log.info("[语音对讲]失败 deviceId: {}, channelId: {}", device.getDeviceId(), channel.getDeviceId());
             audioEvent.call("失败, " + e.getMessage());
             // 查看是否已经建立了通道，存在则发送bye
@@ -584,7 +591,7 @@ public class PlayServiceImpl implements IPlayService {
                 dynamicTask.stop(timeOutTaskKey);
                 receiveRtpServerService.closeRTPServer(mediaServerItem, sendRtpInfo.getApp(), sendRtpInfo.getStream());
                 // 释放ssrc
-                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
+                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrcToRelease());
                 sessionManager.removeByStream(sendRtpInfo.getApp(), sendRtpInfo.getStream());
                 errorEvent.response(event);
             }, userSetting.getPlayTimeout().longValue());
@@ -594,7 +601,7 @@ public class PlayServiceImpl implements IPlayService {
             dynamicTask.stop(timeOutTaskKey);
             receiveRtpServerService.closeRTPServer(mediaServerItem, sendRtpInfo.getApp(), sendRtpInfo.getStream());
             // 释放ssrc
-            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
+            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrcToRelease());
 
             sessionManager.removeByStream(sendRtpInfo.getApp(), sendRtpInfo.getStream());
             SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult();
@@ -879,8 +886,7 @@ public class PlayServiceImpl implements IPlayService {
                     // ssrc检验
                     // 更新ssrc
                     log.info("[Invite 200OK] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
-                    // 释放ssrc
+                    releaseAllocatedSsrc(mediaServerItem, ssrcInfo);
-                    mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
                     Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInResponse);
                     if (!result) {
                         try {
@@ -890,8 +896,7 @@ public class PlayServiceImpl implements IPlayService {
                             log.error("[命令发送失败] 停止播放， 发送BYE: {}", e.getMessage());
                         }
 
-                        // 释放ssrc
+                        mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrcToRelease());
-                        mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
 
                         sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream());
 
@@ -903,6 +908,7 @@ public class PlayServiceImpl implements IPlayService {
 
                     }else {
                         ssrcInfo.setSsrc(ssrcInResponse);
+                        updateSsrcTransaction(ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInResponse, null);
                         inviteInfo.setSsrcInfo(ssrcInfo);
                         inviteInfo.setStream(ssrcInfo.getStream());
                         if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
@@ -914,18 +920,37 @@ public class PlayServiceImpl implements IPlayService {
                         }
                         inviteStreamService.updateInviteInfo(inviteInfo);
                     }
+                } else {
+                    releaseAllocatedSsrc(mediaServerItem, ssrcInfo);
+                    ssrcInfo.setSsrc(ssrcInResponse);
+                    updateSsrcTransaction(ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInResponse, null);
+                    inviteInfo.setSsrcInfo(ssrcInfo);
+                    inviteInfo.setStream(ssrcInfo.getStream());
+                    if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
+                        if (mediaServerItem.isRtpEnable()) {
+                            tcpActiveHandler(device, channel, contentString, mediaServerItem,  ssrcInfo, callback);
+                        }else {
+                            log.warn("[Invite 200OK] 单端口收流模式不支持tcp主动模式收流");
+                        }
+                    }
+                    inviteStreamService.updateInviteInfo(inviteInfo);
                 }
             }else {
                 if (ssrcInResponse != null) {
                     // 单端口
                     // 重新订阅流上线
                     SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(MediaApp.GB28181, inviteInfo.getStream());
+                    if (ssrcTransaction == null) {
+                        return;
+                    }
+                    releaseAllocatedSsrc(mediaServerItem, ssrcInfo);
                     sessionManager.removeByStream(MediaApp.GB28181, inviteInfo.getStream());
                     inviteStreamService.updateInviteInfoForSSRC(inviteInfo, ssrcInResponse);
                     ssrcTransaction.setDeviceId(device.getDeviceId());
                     ssrcTransaction.setChannelId(ssrcTransaction.getChannelId());
                     ssrcTransaction.setCallId(ssrcTransaction.getCallId());
                     ssrcTransaction.setSsrc(ssrcInResponse);
+                    ssrcTransaction.setAllocatedSsrc(null);
                     ssrcTransaction.setApp(MediaApp.GB28181);
                     ssrcTransaction.setStream(inviteInfo.getStream());
                     ssrcTransaction.setMediaServerId(mediaServerItem.getId());
@@ -938,6 +963,24 @@ public class PlayServiceImpl implements IPlayService {
         }
     }
 
+    private void releaseAllocatedSsrc(MediaServer mediaServerItem, SSRCInfo ssrcInfo) {
+        if (ssrcInfo == null || ssrcInfo.getAllocatedSsrc() == null) {
+            return;
+        }
+        mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getAllocatedSsrc());
+        ssrcInfo.setAllocatedSsrc(null);
+    }
+
+    private void updateSsrcTransaction(String app, String stream, String ssrc, String allocatedSsrc) {
+        SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(app, stream);
+        if (ssrcTransaction == null) {
+            return;
+        }
+        ssrcTransaction.setSsrc(ssrc);
+        ssrcTransaction.setAllocatedSsrc(allocatedSsrc);
+        sessionManager.put(ssrcTransaction);
+    }
+
     @Override
     public void download(Device device, DeviceChannel channel, String startTime, String endTime, int downloadSpeed, ErrorCallback<StreamInfo> callback) {
 
@@ -1577,7 +1620,7 @@ public class PlayServiceImpl implements IPlayService {
             mediaServerService.stopSendRtp(mediaServer, sendRtpInfo.getApp(), sendRtpInfo.getStream(), sendRtpInfo.getSsrc());
         }
 
-        ssrcFactory.releaseSsrc(mediaServerId, sendRtpInfo.getSsrc());
+        ssrcFactory.releaseSsrc(mediaServerId, sendRtpInfo.getSsrcToRelease());
 
         SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(sendRtpInfo.getApp(), sendRtpInfo.getStream());
         if (ssrcTransaction != null) {

src/main/java/com/genersoft/iot/vmp/gb28181/session/SSRCFactory.java

@@ -39,8 +39,7 @@ public class SSRCFactory {
 
 
     public void initMediaServerSSRC(String mediaServerId, Set<String> usedSet) {
-        String sipDomain = sipConfig.getDomain();
+        String ssrcPrefix = getSsrcPrefix();
-        String ssrcPrefix = sipDomain.length() >= 8 ? sipDomain.substring(3, 8) : sipDomain;
         String redisKey = SSRC_INFO_KEY + userSetting.getServerId() + "_" + mediaServerId;
         List<String> ssrcList = new ArrayList<>();
         for (int i = 1; i < MAX_STREAM_COUNT; i++) {
@@ -83,7 +82,12 @@ public class SSRCFactory {
         if (ssrc == null) {
             return;
         }
+        if (!isFactorySsrc(ssrc)) {
+            log.warn("[释放 SSRC] 忽略非SSRC池分配的值: {}", ssrc);
+            return;
+        }
         String sn = ssrc.substring(1);
+        log.debug("[释放 SSRC] SSRC:{} -> SN: {}", ssrc, sn);
         String redisKey = SSRC_INFO_KEY + userSetting.getServerId() + "_" + mediaServerId;
         redisTemplate.opsForSet().add(redisKey, sn);
     }
@@ -122,4 +126,18 @@ public class SSRCFactory {
         return Boolean.TRUE.equals(redisTemplate.hasKey(redisKey));
     }
 
+    private String getSsrcPrefix() {
+        String sipDomain = sipConfig.getDomain();
+        return sipDomain.length() >= 8 ? sipDomain.substring(3, 8) : sipDomain;
+    }
+
+    private boolean isFactorySsrc(String ssrc) {
+        if (ssrc.length() < 2) {
+            return false;
+        }
+        String sn = ssrc.substring(1);
+        String ssrcPrefix = getSsrcPrefix();
+        return sn.length() == ssrcPrefix.length() + 4 && sn.startsWith(ssrcPrefix);
+    }
+
 }

src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java

@@ -285,7 +285,7 @@ public class SIPCommander implements ISIPCommander {
         Request request = headerProvider.createInviteRequest(device, channel.getDeviceId(), content.toString(), SipUtils.getNewViaTag(), SipUtils.getNewFromTag(), null, ssrcInfo.getSsrc(),sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()));
         sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, (e -> {
             sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream());
-            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
+            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrcToRelease());
             errorEvent.response(e);
         }), e -> {
             ResponseEvent responseEvent = (ResponseEvent) e.event;
@@ -294,6 +294,7 @@ public class SIPCommander implements ISIPCommander {
             SsrcTransaction ssrcTransaction = SsrcTransaction.buildForDevice(device.getDeviceId(), channel.getId(),
                     callId,ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response,
                     InviteSessionType.PLAY);
+            ssrcTransaction.setAllocatedSsrc(ssrcInfo.getAllocatedSsrc());
             sessionManager.put(ssrcTransaction);
             okEvent.response(e);
         }, timeout);
@@ -391,6 +392,7 @@ public class SIPCommander implements ISIPCommander {
                     channel.getId(), sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),
                             device.getTransport()).getCallId(), ssrcInfo.getApp(), ssrcInfo.getStream(), ssrcInfo.getSsrc(),
                     mediaServerItem.getId(), response, InviteSessionType.PLAYBACK);
+            ssrcTransaction.setAllocatedSsrc(ssrcInfo.getAllocatedSsrc());
             sessionManager.put(ssrcTransaction);
             okEvent.response(event);
         }, timeout);
@@ -484,6 +486,7 @@ public class SIPCommander implements ISIPCommander {
             SsrcTransaction ssrcTransaction = SsrcTransaction.buildForDevice(device.getDeviceId(), channel.getId(),
                     response.getCallIdHeader().getCallId(), ssrcInfo.getApp(), ssrcInfo.getStream(), ssrc,
                     mediaServerItem.getId(), response, InviteSessionType.DOWNLOAD);
+            ssrcTransaction.setAllocatedSsrc(ssrcInfo.getAllocatedSsrc());
             sessionManager.put(ssrcTransaction);
             okEvent.response(event);
         }, timeout);
@@ -544,13 +547,14 @@ public class SIPCommander implements ISIPCommander {
                 SipUtils.getNewViaTag(), SipUtils.getNewFromTag(), null, sendRtpItem.getSsrc(), callIdHeader);
         sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, (e -> {
             sessionManager.removeByStream(sendRtpItem.getApp(), sendRtpItem.getStream());
-            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
+            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrcToRelease());
             errorEvent.response(e);
         }), e -> {
             // 这里为例避免一个通道的点播只有一个callID这个参数使用一个固定值
             ResponseEvent responseEvent = (ResponseEvent) e.event;
             SIPResponse response = (SIPResponse) responseEvent.getResponse();
             SsrcTransaction ssrcTransaction = SsrcTransaction.buildForDevice(device.getDeviceId(), channel.getId(), MediaApp.GB28181_TALK,sendRtpItem.getApp(), stream, sendRtpItem.getSsrc(), mediaServerItem.getId(), response, InviteSessionType.TALK);
+            ssrcTransaction.setAllocatedSsrc(sendRtpItem.getAllocatedSsrc());
             sessionManager.put(ssrcTransaction);
             okEvent.response(e);
         }, timeout);

src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderForPlatform.java

@@ -642,7 +642,7 @@ public class SIPCommanderForPlatform implements ISIPCommanderForPlatform {
         String mediaServerId = sendRtpItem.getMediaServerId();
         MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId);
         if (mediaServerItem != null) {
-            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
+            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrcToRelease());
             receiveRtpServerService.closeRTPServer(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream());
         }
         SIPRequest byeRequest = headerProviderPlatformProvider.createByeRequest(platform, sendRtpItem, channel);
@@ -744,13 +744,14 @@ public class SIPCommanderForPlatform implements ISIPCommanderForPlatform {
                 callIdHeader);
         sipSender.transmitRequest(sipLayer.getLocalIp(platform.getDeviceIp()), request, (e -> {
             sessionManager.removeByStream(ssrcInfo.getApp(), ssrcInfo.getStream());
-            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
+            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrcToRelease());
             errorEvent.response(e);
         }), e -> {
             ResponseEvent responseEvent = (ResponseEvent) e.event;
             SIPResponse response = (SIPResponse) responseEvent.getResponse();
             SsrcTransaction ssrcTransaction = SsrcTransaction.buildForPlatform(platform.getServerGBId(), channel.getGbId(),
                     callIdHeader.getCallId(), ssrcInfo.getApp(),  stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response, InviteSessionType.BROADCAST);
+            ssrcTransaction.setAllocatedSsrc(ssrcInfo.getAllocatedSsrc());
             sessionManager.put(ssrcTransaction);
             okEvent.response(e);
         });

src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java

@@ -132,7 +132,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
 						if (mediaServer != null) {
 							mediaServerService.stopSendRtp(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc());
 							if (userSetting.getUseCustomSsrcForParentInvite()) {
-								mediaServerService.releaseSsrc(mediaServer.getId(), sendRtpItem.getSsrc());
+								mediaServerService.releaseSsrc(mediaServer.getId(), sendRtpItem.getSsrcToRelease());
 							}
 						}
 					}
@@ -144,7 +144,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
 				sendRtpServerService.delete(sendRtpItem);
 				mediaServerService.stopSendRtp(mediaInfo, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc());
 				if (userSetting.getUseCustomSsrcForParentInvite()) {
-					mediaServerService.releaseSsrc(mediaInfo.getId(), sendRtpItem.getSsrc());
+					mediaServerService.releaseSsrc(mediaInfo.getId(), sendRtpItem.getSsrcToRelease());
 				}
 			}
 			if (sendRtpItem.getServerId().equals(userSetting.getServerId())) {
@@ -254,7 +254,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
 			// 释放ssrc
 			MediaServer mediaServerItem = mediaServerService.getOne(ssrcTransaction.getMediaServerId());
 			if (mediaServerItem != null) {
-				mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransaction.getSsrc());
+				mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransaction.getSsrcToRelease());
 			}
 			sessionManager.removeByCallId(ssrcTransaction.getCallId());
 		}

src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java

@@ -179,6 +179,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                                         ? ssrcFactory.getPlaySsrc(streamInfo.getMediaServer().getId())
                                         : ssrcFactory.getPlayBackSsrc(streamInfo.getMediaServer().getId());
                                 inviteInfo.setSsrc(ssrc);
+                                inviteInfo.setAllocatedSsrc(ssrc);
+                                inviteInfo.setAllocatedSsrcMediaServerId(streamInfo.getMediaServer().getId());
                             }
                         }
                         // 构建sendRTP内容
@@ -186,6 +188,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                                 inviteInfo.getIp(), inviteInfo.getPort(), inviteInfo.getSsrc(), platform.getServerGBId(),
                                 streamInfo.getApp(), streamInfo.getStream(),
                                 channel.getGbId(), inviteInfo.isTcp(), platform.isRtcp());
+                        sendRtpItem.setAllocatedSsrc(inviteInfo.getAllocatedSsrc());
                         if (inviteInfo.isTcp() && inviteInfo.isTcpActive()) {
                             sendRtpItem.setTcpActive(true);
                         }
@@ -203,7 +206,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                         // 超时未收到Ack应该回复bye,当前等待时间为10秒
                         dynamicTask.startDelay(inviteInfo.getCallId(), () -> {
                             log.info("[Ack ] 等待超时, {}/{}", inviteInfo.getCallId(), channel.getGbDeviceId());
-                            mediaServerService.releaseSsrc(streamInfo.getMediaServer().getId(), sendRtpItem.getSsrc());
+                            mediaServerService.releaseSsrc(streamInfo.getMediaServer().getId(), sendRtpItem.getSsrcToRelease());
                             // 回复bye
                             sendBye(platform, inviteInfo.getCallId());
                         }, 60 * 1000);
@@ -244,6 +247,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                 log.error("[命令发送失败] invite BAD_REQUEST: {}", sendException.getMessage());
             }
         }catch (PlayException e) {
+            releaseAllocatedSsrc(inviteInfo);
             try {
                 responseAck(request, e.getCode(), e.getMsg());
             } catch (SipException | InvalidArgumentException | ParseException sendException) {
@@ -251,6 +255,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
             }
         }catch (Exception e) {
             log.error("[Invite处理异常] ", e);
+            releaseAllocatedSsrc(inviteInfo);
             try {
                 responseAck(request, Response.SERVER_INTERNAL_ERROR, "");
             } catch (SipException | InvalidArgumentException | ParseException sendException) {
@@ -259,6 +264,15 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
         }
     }
 
+    private void releaseAllocatedSsrc(InviteMessageInfo inviteInfo) {
+        if (inviteInfo == null || inviteInfo.getAllocatedSsrc() == null || inviteInfo.getAllocatedSsrcMediaServerId() == null) {
+            return;
+        }
+        mediaServerService.releaseSsrc(inviteInfo.getAllocatedSsrcMediaServerId(), inviteInfo.getAllocatedSsrc());
+        inviteInfo.setAllocatedSsrc(null);
+        inviteInfo.setAllocatedSsrcMediaServerId(null);
+    }
+
     private InviteMessageInfo decode(RequestEvent evt) throws SdpException {
 
         InviteMessageInfo inviteInfo = new InviteMessageInfo();

src/main/java/com/genersoft/iot/vmp/service/bean/SSRCInfo.java

@@ -7,6 +7,7 @@ public class SSRCInfo {
 
     private int port;
     private String ssrc;
+    private String allocatedSsrc;
     private String app;
     private String Stream;
 
@@ -17,4 +18,8 @@ public class SSRCInfo {
         this.Stream = stream;
     }
 
+    public String getSsrcToRelease() {
+        return allocatedSsrc;
+    }
+
 }

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

@@ -102,6 +102,9 @@ public class RtpServerServiceImpl implements IReceiveRtpServerService {
         }
 
         SSRCInfo ssrcInfo = new SSRCInfo(0, ssrc, MediaApp.GB28181, streamId);
+        if (presetSSRC == null) {
+            ssrcInfo.setAllocatedSsrc(ssrc);
+        }
         RTPServerParam rtpServerParam = new RTPServerParam(mediaServer, MediaApp.GB28181, streamId, ssrcCheck ? Long.parseLong(ssrc): 0L, null, onlyAuto, disableAuto, false, tcpMode);
         int rtpServerPort = openRTPServer(rtpServerParam, ((code, msg, data) -> {
             if (code == InviteErrorCode.SUCCESS.getCode()) {
@@ -113,6 +116,7 @@ public class RtpServerServiceImpl implements IReceiveRtpServerService {
                 // 释放ssrc
                 if (presetSSRC == null) {
                     ssrcFactory.releaseSsrc(mediaServer.getId(), ssrc);
+                    ssrcInfo.setAllocatedSsrc(null);
                 }
                 OpenRTPServerResult openRTPServerResult = new OpenRTPServerResult();
                 openRTPServerResult.setSsrcInfo(ssrcInfo);

src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcSendRtpController.java

@@ -56,6 +56,7 @@ public class RedisRpcSendRtpController extends RpcController {
         if (mediaServerItem == null) {
             RedisRpcResponse response = request.getResponse();
             response.setStatusCode(ErrorCode.SUCCESS.getCode());
+            return response;
         }
         // 自平台内容
         int localPort = sendRtpServerService.getNextPort(mediaServerItem);
@@ -63,6 +64,7 @@ public class RedisRpcSendRtpController extends RpcController {
             log.info("[redis-rpc] getSendRtpItem->服务器端口资源不足" );
             RedisRpcResponse response = request.getResponse();
             response.setStatusCode(ErrorCode.SUCCESS.getCode());
+            return response;
         }
         // 写入redis， 超时时回复
         sendRtpItem.setStatus(1);
@@ -72,6 +74,7 @@ public class RedisRpcSendRtpController extends RpcController {
             // 上级平台点播时不使用上级平台指定的ssrc，使用自定义的ssrc，参考国标文档-点播外域设备媒体流SSRC处理方式
             String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(mediaServerItem.getId()) : ssrcFactory.getPlayBackSsrc(mediaServerItem.getId());
             sendRtpItem.setSsrc(ssrc);
+            sendRtpItem.setAllocatedSsrc(ssrc);
         }
         sendRtpServerService.update(sendRtpItem);
         RedisRpcResponse response = request.getResponse();
@@ -99,6 +102,7 @@ public class RedisRpcSendRtpController extends RpcController {
         MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId());
         if (mediaServer == null) {
             log.info("[redis-rpc] startSendRtp->未找到MediaServer： {}", sendRtpItem.getMediaServerId() );
+            clearSendRtpItem(sendRtpItem);
             WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "未找到MediaServer");
             response.setBody(wvpResult);
             return response;
@@ -106,6 +110,7 @@ public class RedisRpcSendRtpController extends RpcController {
         MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream());
         if (mediaInfo == null) {
             log.info("[redis-rpc] startSendRtp->流不在线： {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() );
+            clearSendRtpItem(sendRtpItem);
             WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "流不在线");
             response.setBody(wvpResult);
             return response;
@@ -114,6 +119,7 @@ public class RedisRpcSendRtpController extends RpcController {
             mediaServerService.startSendRtp(mediaServer, sendRtpItem);
         }catch (ControllerException exception) {
             log.info("[redis-rpc] 发流失败： {}/{}, 目标地址： {}：{}， {}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort(), exception.getMsg());
+            clearSendRtpItem(sendRtpItem);
             WVPResult wvpResult = WVPResult.fail(exception.getCode(), exception.getMsg());
             response.setBody(wvpResult);
             return response;
@@ -143,6 +149,7 @@ public class RedisRpcSendRtpController extends RpcController {
         MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId());
         if (mediaServer == null) {
             log.info("[redis-rpc] stopSendRtp->未找到MediaServer： {}", sendRtpItem.getMediaServerId() );
+            clearSendRtpItem(sendRtpItem);
             WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "未找到MediaServer");
             response.setBody(wvpResult);
             return response;
@@ -155,9 +162,20 @@ public class RedisRpcSendRtpController extends RpcController {
             response.setBody(WVPResult.fail(exception.getCode(), exception.getMsg()));
             return response;
         }
+        clearSendRtpItem(sendRtpItem);
         log.info("[redis-rpc] 停止推流成功： {}/{}, 目标地址： {}：{}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() );
         response.setBody(WVPResult.success());
         return response;
     }
 
+    private void clearSendRtpItem(SendRtpInfo sendRtpItem) {
+        if (sendRtpItem == null) {
+            return;
+        }
+        sendRtpServerService.delete(sendRtpItem);
+        if (sendRtpItem.getMediaServerId() != null) {
+            mediaServerService.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrcToRelease());
+        }
+    }
+
 }

src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcStreamPushController.java

@@ -76,6 +76,7 @@ public class RedisRpcStreamPushController extends RpcController {
                 // 上级平台点播时不使用上级平台指定的ssrc，使用自定义的ssrc，参考国标文档-点播外域设备媒体流SSRC处理方式
                 String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(mediaServer.getId()) : ssrcFactory.getPlayBackSsrc(mediaServer.getId());
                 sendRtpItem.setSsrc(ssrc);
+                sendRtpItem.setAllocatedSsrc(ssrc);
             }
             sendRtpItem.setMediaServerId(mediaServer.getId());
             sendRtpItem.setLocalIp(mediaServer.getSdpIp());
@@ -95,6 +96,7 @@ public class RedisRpcStreamPushController extends RpcController {
                 // 上级平台点播时不使用上级平台指定的ssrc，使用自定义的ssrc，参考国标文档-点播外域设备媒体流SSRC处理方式
                 String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(hookData.getMediaServer().getId()) : ssrcFactory.getPlayBackSsrc(hookData.getMediaServer().getId());
                 sendRtpItem.setSsrc(ssrc);
+                sendRtpItem.setAllocatedSsrc(ssrc);
             }
             sendRtpItem.setMediaServerId(hookData.getMediaServer().getId());
             sendRtpItem.setLocalIp(hookData.getMediaServer().getSdpIp());

src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java

@@ -110,6 +110,7 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
                 // 上级平台点播时不使用上级平台指定的ssrc，使用自定义的ssrc，参考国标文档-点播外域设备媒体流SSRC处理方式
                 String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(hookData.getMediaServer().getId()) : ssrcFactory.getPlayBackSsrc(hookData.getMediaServer().getId());
                 sendRtpItem.setSsrc(ssrc);
+                sendRtpItem.setAllocatedSsrc(ssrc);
             }
             sendRtpItem.setMediaServerId(hookData.getMediaServer().getId());
             sendRtpItem.setLocalIp(hookData.getMediaServer().getSdpIp());