From c3ce2bc5d069ba57309567aee4ae418bc0cf75ed Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 17 三月 2023 13:22:05 +0800
Subject: [PATCH] 优化异常处理以及级联设备状态查询

---
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java |  144 +++++++++++++++++++++++++++++-------------------
 1 files changed, 87 insertions(+), 57 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 a32fadc..1a6358b 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,21 +1,21 @@
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
 
-import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson2.JSONObject;
 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.SsrcConfig;
 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
 import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
 import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
-import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
 import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
-import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
 import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
 import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
+import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
 import com.genersoft.iot.vmp.media.zlm.dto.*;
 import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.IPlayService;
@@ -39,12 +39,14 @@
 import org.springframework.stereotype.Component;
 
 import javax.sdp.*;
-import javax.sip.*;
+import javax.sip.InvalidArgumentException;
+import javax.sip.RequestEvent;
+import javax.sip.SipException;
 import javax.sip.header.CallIdHeader;
-import javax.sip.message.Request;
 import javax.sip.message.Response;
 import java.text.ParseException;
 import java.time.Instant;
+import java.util.Random;
 import java.util.Vector;
 
 /**
@@ -157,11 +159,6 @@
                 StreamProxyItem proxyByAppAndStream =null;
                 // 涓嶆槸閫氶亾鍙兘鏄洿鎾祦
                 if (channel != null && gbStream == null) {
-//                    if (channel.getStatus() == 0) {
-//                        logger.info("閫氶亾绂荤嚎锛岃繑鍥�400");
-//                        responseAck(request, Response.BAD_REQUEST, "channel [" + channel.getChannelId() + "] offline");
-//                        return;
-//                    }
                     // 閫氶亾瀛樺湪锛屽彂100锛孴RYING
                     try {
                         responseAck(request, Response.TRYING);
@@ -232,7 +229,7 @@
                     }
                     return;
                 } else {
-                    logger.info("閫氶亾涓嶅瓨鍦紝杩斿洖404");
+                    logger.info("閫氶亾涓嶅瓨鍦紝杩斿洖404: {}", channelId);
                     try {
                         // 閫氶亾涓嶅瓨鍦紝鍙�404锛岃祫婧愪笉瀛樺湪
                         responseAck(request, Response.NOT_FOUND);
@@ -318,7 +315,7 @@
                     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;
@@ -345,8 +342,7 @@
                         return;
                     }
                     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);
@@ -385,7 +381,12 @@
                         } else {
                             content.append("t=0 0\r\n");
                         }
-                        content.append("m=video " + sendRtpItem.getLocalPort() + " RTP/AVP 96\r\n");
+                        int localPort = sendRtpItem.getLocalPort();
+                        if (localPort == 0) {
+                            // 闈炰弗鏍兼ā寮忕鍙d笉缁熶竴, 澧炲姞鍏煎鎬э紝淇敼涓轰竴涓笉涓�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");
                         content.append("y=" + sendRtpItem.getSsrc() + "\r\n");
@@ -405,21 +406,17 @@
                             }, 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) -> {
                         // 鏈煡閿欒銆傜洿鎺ヨ浆鍙戣澶囩偣鎾殑閿欒
                         try {
                             Response response = getMessageFactory().createResponse(event.statusCode, evt.getRequest());
-                            sipSender.transmitRequest(response);
+                            sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response);
                         } catch (ParseException | SipException  e) {
-                            e.printStackTrace();
+                            logger.error("鏈鐞嗙殑寮傚父 ", e);
                         }
                     });
                     sendRtpItem.setApp("rtp");
@@ -476,10 +473,11 @@
 
                             // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
                             redisCatchStorage.updateSendRTPSever(sendRtpItem);
+                            MediaServerItem finalMediaServerItem = mediaServerItem;
                             playService.play(mediaServerItem, ssrcInfo, device, channelId, hookEvent, errorEvent, (code, msg) -> {
                                 logger.info("[涓婄骇鐐规挱]瓒呮椂, 鐢ㄦ埛锛歿}锛� 閫氶亾锛歿}", username, channelId);
                                 redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
-                            }, null);
+                            });
                         } else {
                             sendRtpItem.setStreamId(playTransaction.getStream());
                             // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
@@ -491,6 +489,15 @@
                         }
                     }
                 } else if (gbStream != null) {
+                    if(ssrc.equals(ssrcDefault))
+                    {
+                        SsrcConfig ssrcConfig = mediaServerItem.getSsrcConfig();
+                        if(ssrcConfig != null)
+                        {
+                            ssrc = ssrcConfig.getPlaySsrc();
+                            ssrcConfig.releaseSsrc(ssrc);
+                        }
+                    }
                     if("push".equals(gbStream.getStreamType())) {
                         if (streamPushItem != null && streamPushItem.isPushIng()) {
                             // 鎺ㄦ祦鐘舵��
@@ -502,21 +509,17 @@
                                     mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
                         }
                     }else if ("proxy".equals(gbStream.getStreamType())){
-                        if(null != proxyByAppAndStream &&proxyByAppAndStream.isStatus()){
-                            pushProxyStream(evt, request, gbStream,  platform, callIdHeader, mediaServerItem, port, tcpActive,
-                                    mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
-                        }else{
-                            //寮�鍚唬鐞嗘媺娴�
-                            boolean start1 = streamProxyService.start(gbStream.getApp(), gbStream.getStream());
-                            if(start1) {
+                        if (null != proxyByAppAndStream) {
+                            if(proxyByAppAndStream.isStatus()){
                                 pushProxyStream(evt, request, gbStream,  platform, callIdHeader, mediaServerItem, port, tcpActive,
                                         mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
                             }else{
-                                //澶辫触鍚庨�氱煡
+                                //寮�鍚唬鐞嗘媺娴�
                                 notifyStreamOnline(evt, request,gbStream, null, platform, callIdHeader, mediaServerItem, port, tcpActive,
                                         mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
                             }
                         }
+
 
                     }
                 }
@@ -524,7 +527,7 @@
         } catch (SdpParseException e) {
             logger.error("sdp瑙f瀽閿欒", e);
         } catch (SdpException e) {
-            e.printStackTrace();
+            logger.error("鏈鐞嗙殑寮傚父 ", e);
         }
     }
 
@@ -539,8 +542,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祫婧愪笉瓒�");
@@ -579,8 +581,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祫婧愪笉瓒�");
@@ -629,15 +630,38 @@
         if ("proxy".equals(gbStream.getStreamType())) {
             // TODO 鎺у埗鍚敤浠ヤ娇璁惧涓婄嚎
             logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍚敤娴佸悗寮�濮嬫帹娴�", gbStream.getApp(), gbStream.getStream());
-            try {
-                responseAck(request, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline");
-            } catch (SipException | InvalidArgumentException | ParseException e) {
-                logger.error("[鍛戒护鍙戦�佸け璐 invite 閫氶亾鏈帹娴�: {}", e.getMessage());
+            // 鐩戝惉娴佷笂绾�
+            HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(gbStream.getApp(), gbStream.getStream(), true, "rtsp", mediaServerItem.getId());
+            zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, responseJSON) -> {
+                String app = responseJSON.getString("app");
+                String stream = responseJSON.getString("stream");
+                logger.info("[涓婄骇鐐规挱]鎷夋祦浠g悊宸茬粡灏辩华锛� {}/{}", app, stream);
+                dynamicTask.stop(callIdHeader.getCallId());
+                pushProxyStream(evt, request, gbStream,  platform, callIdHeader, mediaServerItem, port, tcpActive,
+                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+            });
+            dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
+                logger.info("[ app={}, stream={} ] 绛夊緟鎷夋祦浠g悊娴佽秴鏃�", gbStream.getApp(), gbStream.getStream());
+                zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
+            }, userSetting.getPlatformPlayTimeout());
+            boolean start = streamProxyService.start(gbStream.getApp(), gbStream.getStream());
+            if (!start) {
+                try {
+                    responseAck(request, Response.BUSY_HERE, "channel [" + gbStream.getGbId() + "] offline");
+                } catch (SipException | InvalidArgumentException | ParseException e) {
+                    logger.error("[鍛戒护鍙戦�佸け璐 invite 閫氶亾鏈帹娴�: {}", e.getMessage());
+                }
+                zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
+                dynamicTask.stop(callIdHeader.getCallId());
             }
+
+
+
         } else if ("push".equals(gbStream.getStreamType())) {
             if (!platform.isStartOfflinePush()) {
                 // 骞冲彴璁剧疆涓叧闂簡鎷夎捣绂荤嚎鐨勬帹娴佸垯鐩存帴鍥炲
                 try {
+                    logger.info("[涓婄骇鐐规挱] 澶辫触锛屾帹娴佽澶囨湭鎺ㄦ祦锛宑hannel: {}, app: {}, stream: {}", gbStream.getGbId(), gbStream.getApp(), gbStream.getStream());
                     responseAck(request, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing");
                 } catch (SipException | InvalidArgumentException | ParseException e) {
                     logger.error("[鍛戒护鍙戦�佸け璐 invite 閫氶亾鏈帹娴�: {}", e.getMessage());
@@ -658,11 +682,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());
             // 娣诲姞鐩戝惉
@@ -674,18 +698,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;
                     }
@@ -736,18 +760,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;
                     }
@@ -800,7 +824,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()) {
@@ -817,11 +847,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;
     }
@@ -894,7 +924,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