From 23ce55260fdb9277cd72987c1e6a3d4e058f3bb4 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 25 八月 2023 08:51:06 +0800
Subject: [PATCH] 修复级联点播

---
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java |  159 +++++++++++++++++++++++++++++------------------------
 1 files changed, 87 insertions(+), 72 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
index 3d3c772..89381d9 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
@@ -1,10 +1,12 @@
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
 
 import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
+import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
 import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
 import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
@@ -72,6 +74,9 @@
 
     @Autowired
     private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private SSRCFactory ssrcFactory;
 
     @Autowired
     private DynamicTask dynamicTask;
@@ -179,14 +184,11 @@
                             return;
                         } else {
                             streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream());
-                            if (streamPushItem == null || streamPushItem.getServerId().equals(userSetting.getServerId())) {
-                                logger.info("[ app={}, stream={} ]鎵句笉鍒皕lm {}锛岃繑鍥�410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
-                                try {
-                                    responseAck(request, Response.GONE);
-                                } catch (SipException | InvalidArgumentException | ParseException e) {
-                                    logger.error("[鍛戒护鍙戦�佸け璐 invite GONE: {}", e.getMessage());
-                                }
-                                return;
+                            if (streamPushItem != null) {
+                                mediaServerItem = mediaServerService.getOne(streamPushItem.getMediaServerId());
+                            }
+                            if (mediaServerItem == null) {
+                                mediaServerItem = mediaServerService.getDefaultMediaServer();
                             }
                         }
                     } else {
@@ -228,7 +230,7 @@
                     }
                     return;
                 } else {
-                    logger.info("閫氶亾涓嶅瓨鍦紝杩斿洖404");
+                    logger.info("閫氶亾涓嶅瓨鍦紝杩斿洖404: {}", channelId);
                     try {
                         // 閫氶亾涓嶅瓨鍦紝鍙�404锛岃祫婧愪笉瀛樺湪
                         responseAck(request, Response.NOT_FOUND);
@@ -240,21 +242,8 @@
                 // 瑙f瀽sdp娑堟伅, 浣跨敤jainsip 鑷甫鐨剆dp瑙f瀽鏂瑰紡
                 String contentString = new String(request.getRawContent());
 
-                // jainSip涓嶆敮鎸亂=瀛楁锛� 绉婚櫎浠ヨВ鏋愩��
-                int ssrcIndex = contentString.indexOf("y=");
-                // 妫�鏌ユ槸鍚︽湁y瀛楁
-                String ssrcDefault = "0000000000";
-                String ssrc;
-                SessionDescription sdp;
-                if (ssrcIndex >= 0) {
-                    //ssrc瑙勫畾闀垮害涓�10涓瓧鑺傦紝涓嶅彇浣欎笅闀垮害浠ラ伩鍏嶅悗缁繕鏈夆�渇=鈥濆瓧娈�
-                    ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
-                    String substring = contentString.substring(0, contentString.indexOf("y="));
-                    sdp = SdpFactory.getInstance().createSessionDescription(substring);
-                } else {
-                    ssrc = ssrcDefault;
-                    sdp = SdpFactory.getInstance().createSessionDescription(contentString);
-                }
+                Gb28181Sdp gb28181Sdp = SipUtils.parseSDP(contentString);
+                SessionDescription sdp = gb28181Sdp.getBaseSdb();
                 String sessionName = sdp.getSessionName().getValue();
 
                 Long startTime = null;
@@ -314,9 +303,9 @@
                     return;
                 }
                 String username = sdp.getOrigin().getUsername();
-                String addressStr = sdp.getOrigin().getAddress();
+                String addressStr = sdp.getConnection().getAddress();
 
-                logger.info("[涓婄骇鐐规挱]鐢ㄦ埛锛歿}锛� 閫氶亾锛歿}, 鍦板潃锛歿}:{}锛� ssrc锛歿}", username, channelId, addressStr, port, ssrc);
+
                 Device device = null;
                 // 閫氳繃 channel 鍜� gbStream 鏄惁涓簄ull 鍊煎垽鏂潵婧愭槸鐩存挱娴佸悎閫傚浗鏍�
                 if (channel != null) {
@@ -340,9 +329,28 @@
                         }
                         return;
                     }
+
+                    String ssrc;
+                    if (gb28181Sdp.getSsrc() == null) {
+                        // 涓婄骇骞冲彴鐐规挱鏃朵笉浣跨敤涓婄骇骞冲彴鎸囧畾鐨剆src锛屼娇鐢ㄨ嚜瀹氫箟鐨剆src锛屽弬鑰冨浗鏍囨枃妗�-鐐规挱澶栧煙璁惧濯掍綋娴丼SRC澶勭悊鏂瑰紡
+                        ssrc = "Play".equalsIgnoreCase(sessionName) ? ssrcFactory.getPlaySsrc(mediaServerItem.getId()) : ssrcFactory.getPlayBackSsrc(mediaServerItem.getId());
+                        logger.warn("[涓婄骇Invite] {} 骞冲彴锛歿}锛� 閫氶亾锛歿}, 缂哄皯 ssrc锛岃ˉ鍏呬负锛� {}", sessionName, username, channelId, ssrc);
+                    }else {
+                        ssrc = gb28181Sdp.getSsrc();
+                    }
+                    String streamTypeStr = null;
+                    if (mediaTransmissionTCP) {
+                        if (tcpActive) {
+                            streamTypeStr = "TCP-ACTIVE";
+                        }else {
+                            streamTypeStr = "TCP-PASSIVE";
+                        }
+                    }else {
+                        streamTypeStr = "UDP";
+                    }
+                    logger.info("[涓婄骇Invite] {}, 骞冲彴锛歿}锛� 閫氶亾锛歿}, 鏀舵祦鍦板潃锛歿}:{}锛屾敹娴佹柟寮忥細{}, ssrc锛歿}", sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc);
                     SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
-                            device.getDeviceId(), channelId,
-                            mediaTransmissionTCP);
+                            device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                     if (tcpActive != null) {
                         sendRtpItem.setTcpActive(tcpActive);
@@ -406,12 +414,8 @@
                             }, 60 * 1000);
                             responseSdpAck(request, content.toString(), platform);
 
-                        } catch (SipException e) {
-                            e.printStackTrace();
-                        } catch (InvalidArgumentException e) {
-                            e.printStackTrace();
-                        } catch (ParseException e) {
-                            e.printStackTrace();
+                        } catch (SipException | InvalidArgumentException | ParseException e) {
+                            logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍥炲SdpAck", e);
                         }
                     };
                     SipSubscribe.Event errorEvent = ((event) -> {
@@ -420,13 +424,13 @@
                             Response response = getMessageFactory().createResponse(event.statusCode, evt.getRequest());
                             sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response);
                         } catch (ParseException | SipException  e) {
-                            e.printStackTrace();
+                            logger.error("鏈鐞嗙殑寮傚父 ", e);
                         }
                     });
                     sendRtpItem.setApp("rtp");
                     if ("Playback".equalsIgnoreCase(sessionName)) {
                         sendRtpItem.setPlayType(InviteStreamType.PLAYBACK);
-                        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, device.isSsrcCheck(), true);
+                        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
                         sendRtpItem.setStreamId(ssrcInfo.getStream());
                         // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
                         redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -451,29 +455,25 @@
                                 });
                     } else {
                         sendRtpItem.setPlayType(InviteStreamType.PLAY);
-                        SsrcTransaction playTransaction = sessionManager.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
-                        if (playTransaction != null) {
-                            Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream());
+//                        SsrcTransaction playTransaction = sessionManager.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
+                        StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
+                        if (streamInfo != null) {
+                            Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, streamInfo.getApp(), streamInfo.getStream());
                             if (!streamReady) {
-                                boolean hasRtpServer = mediaServerService.checkRtpServer(mediaServerItem, "rtp", playTransaction.getStream());
-                                if (hasRtpServer) {
-                                    logger.info("[涓婄骇鐐规挱]宸茬粡寮�鍚痳tpServer浣嗘槸灏氭湭鏀跺埌娴侊紝寮�鍚洃鍚祦鐨勫埌鏉�");
-                                    HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", playTransaction.getStream(), true, "rtsp", mediaServerItem.getId());
-                                    zlmHttpHookSubscribe.addSubscribe(hookSubscribe, hookEvent);
-                                }else {
-                                    playTransaction = null;
-                                }
+                                redisCatchStorage.stopPlay(streamInfo);
+                                storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
+                                streamInfo = null;
                             }
                         }
-                        if (playTransaction == null) {
+                        if (streamInfo == null) {
                             String streamId = null;
                             if (mediaServerItem.isRtpEnable()) {
                                 streamId = String.format("%s_%s", device.getDeviceId(), channelId);
                             }
-                            SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false);
+                            SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam());
                             logger.info(JSONObject.toJSONString(ssrcInfo));
                             sendRtpItem.setStreamId(ssrcInfo.getStream());
-                            sendRtpItem.setSsrc(ssrc.equals(ssrcDefault) ? ssrcInfo.getSsrc() : ssrc);
+                            sendRtpItem.setSsrc(ssrc);
 
                             // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
                             redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -483,7 +483,9 @@
                                 redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
                             });
                         } else {
-                            sendRtpItem.setStreamId(playTransaction.getStream());
+                            // 褰撳墠绯荤粺浣滀负涓嬬骇骞冲彴浣跨敤锛屽綋涓婄骇骞冲彴鐐规挱鏃朵笉鎼哄甫ssrc鏃讹紝骞朵笖璁惧鍦ㄥ綋鍓嶇郴缁熶腑宸茬粡鐐规挱浜嗐�傝繖涓椂鍊欓渶瑕侀噸鏂扮粰鐢熸垚涓�涓猻src锛屼笉浣跨敤榛樿鐨�"0000000000"銆�
+                            sendRtpItem.setSsrc(ssrc);
+                            sendRtpItem.setStreamId(streamInfo.getStream());
                             // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
                             redisCatchStorage.updateSendRTPSever(sendRtpItem);
                             JSONObject jsonObject = new JSONObject();
@@ -493,6 +495,15 @@
                         }
                     }
                 } else if (gbStream != null) {
+
+                    String ssrc;
+                    if (gb28181Sdp.getSsrc() == null) {
+                        // 涓婄骇骞冲彴鐐规挱鏃朵笉浣跨敤涓婄骇骞冲彴鎸囧畾鐨剆src锛屼娇鐢ㄨ嚜瀹氫箟鐨剆src锛屽弬鑰冨浗鏍囨枃妗�-鐐规挱澶栧煙璁惧濯掍綋娴丼SRC澶勭悊鏂瑰紡
+                        ssrc = "Play".equalsIgnoreCase(sessionName) ? ssrcFactory.getPlaySsrc(mediaServerItem.getId()) : ssrcFactory.getPlayBackSsrc(mediaServerItem.getId());
+                    }else {
+                        ssrc = gb28181Sdp.getSsrc();
+                    }
+
                     if("push".equals(gbStream.getStreamType())) {
                         if (streamPushItem != null && streamPushItem.isPushIng()) {
                             // 鎺ㄦ祦鐘舵��
@@ -522,7 +533,7 @@
         } catch (SdpParseException e) {
             logger.error("sdp瑙f瀽閿欒", e);
         } catch (SdpException e) {
-            e.printStackTrace();
+            logger.error("鏈鐞嗙殑寮傚父 ", e);
         }
     }
 
@@ -537,8 +548,7 @@
             if (streamReady) {
                 // 鑷钩鍙板唴瀹�
                 SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
-                        gbStream.getApp(), gbStream.getStream(), channelId,
-                        mediaTransmissionTCP);
+                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                 if (sendRtpItem == null) {
                     logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�");
@@ -577,8 +587,7 @@
             if (streamReady) {
                 // 鑷钩鍙板唴瀹�
                 SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
-                        gbStream.getApp(), gbStream.getStream(), channelId,
-                        mediaTransmissionTCP);
+                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                 if (sendRtpItem == null) {
                     logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�");
@@ -679,11 +688,11 @@
                     mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
                     responseAck(request, Response.REQUEST_TIMEOUT); // 瓒呮椂
                 } catch (SipException e) {
-                    e.printStackTrace();
+                    logger.error("鏈鐞嗙殑寮傚父 ", e);
                 } catch (InvalidArgumentException e) {
-                    e.printStackTrace();
+                    logger.error("鏈鐞嗙殑寮傚父 ", e);
                 } catch (ParseException e) {
-                    e.printStackTrace();
+                    logger.error("鏈鐞嗙殑寮傚父 ", e);
                 }
             }, userSetting.getPlatformPlayTimeout());
             // 娣诲姞鐩戝惉
@@ -695,18 +704,18 @@
                 dynamicTask.stop(callIdHeader.getCallId());
                 if (serverId.equals(userSetting.getServerId())) {
                     SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
-                            app, stream, channelId, mediaTransmissionTCP);
+                            app, stream, channelId, mediaTransmissionTCP, platform.isRtcp());
 
                     if (sendRtpItem == null) {
                         logger.warn("涓婄骇鐐规椂鍒涘缓sendRTPItem澶辫触锛屽彲鑳芥槸鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�");
                         try {
                             responseAck(request, Response.BUSY_HERE);
                         } catch (SipException e) {
-                            e.printStackTrace();
+                            logger.error("鏈鐞嗙殑寮傚父 ", e);
                         } catch (InvalidArgumentException e) {
-                            e.printStackTrace();
+                            logger.error("鏈鐞嗙殑寮傚父 ", e);
                         } catch (ParseException e) {
-                            e.printStackTrace();
+                            logger.error("鏈鐞嗙殑寮傚父 ", e);
                         }
                         return;
                     }
@@ -757,18 +766,18 @@
         // 鍙戦�乺edis娑堟伅
         redisGbPlayMsgListener.sendMsg(streamPushItem.getServerId(), streamPushItem.getMediaServerId(),
                 streamPushItem.getApp(), streamPushItem.getStream(), addressStr, port, ssrc, requesterId,
-                channelId, mediaTransmissionTCP, null, responseSendItemMsg -> {
+                channelId, mediaTransmissionTCP, platform.isRtcp(),null, responseSendItemMsg -> {
                     SendRtpItem sendRtpItem = responseSendItemMsg.getSendRtpItem();
                     if (sendRtpItem == null || responseSendItemMsg.getMediaServerItem() == null) {
                         logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�");
                         try {
                             responseAck(request, Response.BUSY_HERE);
                         } catch (SipException e) {
-                            e.printStackTrace();
+                            logger.error("鏈鐞嗙殑寮傚父 ", e);
                         } catch (InvalidArgumentException e) {
-                            e.printStackTrace();
+                            logger.error("鏈鐞嗙殑寮傚父 ", e);
                         } catch (ParseException e) {
-                            e.printStackTrace();
+                            logger.error("鏈鐞嗙殑寮傚父 ", e);
                         }
                         return;
                     }
@@ -821,7 +830,13 @@
         content.append("s=Play\r\n");
         content.append("c=IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
         content.append("t=0 0\r\n");
-        content.append("m=video " + sendRtpItem.getLocalPort() + " RTP/AVP 96\r\n");
+        // 闈炰弗鏍兼ā寮忕鍙d笉缁熶竴, 澧炲姞鍏煎鎬э紝淇敼涓轰竴涓笉涓�0鐨勭鍙�
+        int localPort = sendRtpItem.getLocalPort();
+        if(localPort == 0)
+        {
+            localPort = new Random().nextInt(65535) + 1;
+        }
+        content.append("m=video " + localPort + " RTP/AVP 96\r\n");
         content.append("a=sendonly\r\n");
         content.append("a=rtpmap:96 PS/90000\r\n");
         if (sendRtpItem.isTcp()) {
@@ -838,11 +853,11 @@
         try {
             return responseSdpAck(request, content.toString(), platform);
         } catch (SipException e) {
-            e.printStackTrace();
+            logger.error("鏈鐞嗙殑寮傚父 ", e);
         } catch (InvalidArgumentException e) {
-            e.printStackTrace();
+            logger.error("鏈鐞嗙殑寮傚父 ", e);
         } catch (ParseException e) {
-            e.printStackTrace();
+            logger.error("鏈鐞嗙殑寮傚父 ", e);
         }
         return null;
     }
@@ -915,7 +930,7 @@
                     return;
                 }
                 String username = sdp.getOrigin().getUsername();
-                String addressStr = sdp.getOrigin().getAddress();
+                String addressStr = sdp.getConnection().getAddress();
                 logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}", username, addressStr, port, ssrc);
             } catch (SdpException e) {
                 logger.error("[SDP瑙f瀽寮傚父]", e);

--
Gitblit v1.8.0