From 3d68b56bb5ecbc3d34b2901c671dc7f1f0f82f74 Mon Sep 17 00:00:00 2001
From: pedoc <pedoc@qq.com>
Date: 星期一, 29 一月 2024 18:05:17 +0800
Subject: [PATCH] 修复未配置sipdomain时出现异常

---
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java |  154 +++++++++++++++++++++++++++++++++------------------
 1 files changed, 99 insertions(+), 55 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
old mode 100644
new mode 100755
index 635c3ca..5677008
--- 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,5 +1,7 @@
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
 
+import com.alibaba.fastjson2.JSON;
+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;
@@ -12,13 +14,11 @@
 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.ZLMMediaListManager;
-import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
+import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
 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;
-import com.genersoft.iot.vmp.service.IStreamProxyService;
-import com.genersoft.iot.vmp.service.IStreamPushService;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
+import com.genersoft.iot.vmp.service.*;
 import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
@@ -46,6 +46,8 @@
 import javax.sip.message.Response;
 import java.text.ParseException;
 import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Random;
 import java.util.Vector;
 
@@ -75,6 +77,9 @@
     private IRedisCatchStorage redisCatchStorage;
 
     @Autowired
+    private IInviteStreamService inviteStreamService;
+
+    @Autowired
     private SSRCFactory ssrcFactory;
 
     @Autowired
@@ -90,7 +95,7 @@
     private SIPSender sipSender;
 
     @Autowired
-    private ZLMRTPServerFactory zlmrtpServerFactory;
+    private ZLMServerFactory zlmServerFactory;
 
     @Autowired
     private IMediaServerService mediaServerService;
@@ -132,7 +137,7 @@
             String requesterId = SipUtils.getUserIdFromFromHeader(request);
             CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
             if (requesterId == null || channelId == null) {
-                logger.info("鏃犳硶浠嶧romHeader鐨凙ddress涓幏鍙栧埌骞冲彴id锛岃繑鍥�400");
+                logger.info("鏃犳硶浠庤姹備腑鑾峰彇鍒板钩鍙癷d锛岃繑鍥�400");
                 // 鍙傛暟涓嶅叏锛� 鍙�400锛岃姹傞敊璇�
                 try {
                     responseAck(request, Response.BAD_REQUEST);
@@ -142,6 +147,8 @@
                 return;
             }
 
+            logger.info("[INVITE] requesterId: {}, callId: {}, 鏉ヨ嚜锛歿}锛歿}",
+                    requesterId, callIdHeader.getCallId(), request.getRemoteAddress(), request.getRemotePort());
 
             // 鏌ヨ璇锋眰鏄惁鏉ヨ嚜涓婄骇骞冲彴\璁惧
             ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
@@ -152,7 +159,7 @@
                 // 鏌ヨ骞冲彴涓嬫槸鍚︽湁璇ラ�氶亾
                 DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
                 GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);
-                PlatformCatalog catalog = storager.getCatalog(channelId);
+                PlatformCatalog catalog = storager.getCatalog(requesterId, channelId);
 
                 MediaServerItem mediaServerItem = null;
                 StreamPushItem streamPushItem = null;
@@ -180,16 +187,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;
-                            }else {
-                                 // TODO 鍙兘婕忓洖澶嶆秷鎭�
+                            if (streamPushItem != null) {
+                                mediaServerItem = mediaServerService.getOne(streamPushItem.getMediaServerId());
+                            }
+                            if (mediaServerItem == null) {
+                                mediaServerItem = mediaServerService.getDefaultMediaServer();
                             }
                         }
                     } else {
@@ -349,7 +351,7 @@
                         streamTypeStr = "UDP";
                     }
                     logger.info("[涓婄骇Invite] {}, 骞冲彴锛歿}锛� 閫氶亾锛歿}, 鏀舵祦鍦板潃锛歿}:{}锛屾敹娴佹柟寮忥細{}, ssrc锛歿}", sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc);
-                    SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+                    SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                             device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                     if (tcpActive != null) {
@@ -394,11 +396,21 @@
                             // 闈炰弗鏍兼ā寮忕鍙d笉缁熶竴, 澧炲姞鍏煎鎬э紝淇敼涓轰竴涓笉涓�0鐨勭鍙�
                             localPort = new Random().nextInt(65535) + 1;
                         }
-                        content.append("m=video " + localPort + " RTP/AVP 96\r\n");
+                        if (sendRtpItem.isTcp()) {
+                            content.append("m=video " + localPort + " TCP/RTP/AVP 96\r\n");
+                            if (!sendRtpItem.isTcpActive()) {
+                                content.append("a=setup:active\r\n");
+                            } else {
+                                content.append("a=setup:passive\r\n");
+                            }
+                        }else {
+                            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");
                         content.append("f=\r\n");
+
 
                         try {
                             // 瓒呮椂鏈敹鍒癆ck搴旇鍥炲bye,褰撳墠绛夊緟鏃堕棿涓�10绉�
@@ -413,7 +425,33 @@
                                 }
                             }, 60 * 1000);
                             responseSdpAck(request, content.toString(), platform);
-
+                            // tcp涓诲姩妯″紡锛屽洖澶峴dp鍚庡紑鍚洃鍚�
+                            if (sendRtpItem.isTcpActive()) {
+                                MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+                                Map<String, Object> param = new HashMap<>(12);
+                                param.put("vhost","__defaultVhost__");
+                                param.put("app",sendRtpItem.getApp());
+                                param.put("stream",sendRtpItem.getStreamId());
+                                param.put("ssrc", sendRtpItem.getSsrc());
+                                if (!sendRtpItem.isTcpActive()) {
+                                    param.put("dst_url",sendRtpItem.getIp());
+                                    param.put("dst_port", sendRtpItem.getPort());
+                                }
+                                String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
+                                param.put("is_udp", is_Udp);
+                                param.put("src_port", localPort);
+                                param.put("pt", sendRtpItem.getPt());
+                                param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
+                                param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
+                                if (!sendRtpItem.isTcp()) {
+                                    // 寮�鍚痳tcp淇濇椿
+                                    param.put("udp_rtcp_timeout", sendRtpItem.isRtcp()? "1":"0");
+                                }
+                                JSONObject startSendRtpStreamResult = zlmServerFactory.startSendRtpStreamForPassive(mediaInfo, param);
+                                if (startSendRtpStreamResult != null) {
+                                    startSendRtpStreamHand(evt, sendRtpItem, null, startSendRtpStreamResult, param, callIdHeader);
+                                }
+                            }
                         } catch (SipException | InvalidArgumentException | ParseException e) {
                             logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍥炲SdpAck", e);
                         }
@@ -432,7 +470,10 @@
                     sendRtpItem.setApp("rtp");
                     if ("Playback".equalsIgnoreCase(sessionName)) {
                         sendRtpItem.setPlayType(InviteStreamType.PLAYBACK);
-                        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
+                        String startTimeStr = DateUtil.urlFormatter.format(start);
+                        String endTimeStr = DateUtil.urlFormatter.format(end);
+                        String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr;
+                        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
                         sendRtpItem.setStreamId(ssrcInfo.getStream());
                         // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
                         redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -449,13 +490,13 @@
                                         errorEvent.run(code, msg, data);
                                     }
                                 });
-                    }else if ("Download".equalsIgnoreCase(sessionName)) {
+                    } else if ("Download".equalsIgnoreCase(sessionName)) {
                         // 鑾峰彇鎸囧畾鐨勪笅杞介�熷害
                         Vector sdpMediaDescriptions = sdp.getMediaDescriptions(true);
                         MediaDescription mediaDescription = null;
                         String downloadSpeed = "1";
                         if (sdpMediaDescriptions.size() > 0) {
-                            mediaDescription = (MediaDescription)sdpMediaDescriptions.get(0);
+                            mediaDescription = (MediaDescription) sdpMediaDescriptions.get(0);
                         }
                         if (mediaDescription != null) {
                             downloadSpeed = mediaDescription.getAttribute("downloadspeed");
@@ -469,37 +510,34 @@
                         playService.download(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
                                 DateUtil.formatter.format(end), Integer.parseInt(downloadSpeed),
                                 (code, msg, data) -> {
-                                    if (code == InviteErrorCode.SUCCESS.getCode()){
+                                    if (code == InviteErrorCode.SUCCESS.getCode()) {
                                         hookEvent.run(code, msg, data);
-                                    }else if (code == InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode() || code == InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode()){
+                                    } else if (code == InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode() || code == InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode()) {
                                         logger.info("[褰曞儚涓嬭浇]瓒呮椂, 鐢ㄦ埛锛歿}锛� 閫氶亾锛歿}", username, channelId);
                                         redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
                                         errorEvent.run(code, msg, data);
-                                    }else {
+                                    } else {
                                         errorEvent.run(code, msg, data);
                                     }
                                 });
-                    }else {
+                    } else {
                         sendRtpItem.setPlayType(InviteStreamType.PLAY);
-                        String streamId = null;
-                        if (mediaServerItem.isRtpEnable()) {
-                            streamId = String.format("%s_%s", device.getDeviceId(), channelId);
-                        }else {
-                            streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
-                        }
+                        String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
                         sendRtpItem.setStreamId(streamId);
                         redisCatchStorage.updateSendRTPSever(sendRtpItem);
-                        playService.play(mediaServerItem, device.getDeviceId(), channelId, ((code, msg, data) -> {
-                            if (code == InviteErrorCode.SUCCESS.getCode()){
+                        SSRCInfo ssrcInfo = playService.play(mediaServerItem, device.getDeviceId(), channelId, ssrc, ((code, msg, data) -> {
+                            if (code == InviteErrorCode.SUCCESS.getCode()) {
                                 hookEvent.run(code, msg, data);
-                            }else if (code == InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode() || code == InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode()){
+                            } else if (code == InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode() || code == InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode()) {
                                 logger.info("[涓婄骇鐐规挱]瓒呮椂, 鐢ㄦ埛锛歿}锛� 閫氶亾锛歿}", username, channelId);
                                 redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
                                 errorEvent.run(code, msg, data);
-                            }else {
+                            } else {
                                 errorEvent.run(code, msg, data);
                             }
                         }));
+                        sendRtpItem.setSsrc(ssrcInfo.getSsrc());
+                        redisCatchStorage.updateSendRTPSever(sendRtpItem);
 
                     }
                 } else if (gbStream != null) {
@@ -545,6 +583,18 @@
         }
     }
 
+    private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform,
+                                        JSONObject jsonObject, Map<String, Object> param, CallIdHeader callIdHeader) {
+        if (jsonObject == null) {
+            logger.error("涓嬬骇TCP琚姩鍚姩鐩戝惉澶辫触: 璇锋鏌LM鏈嶅姟");
+        } else if (jsonObject.getInteger("code") == 0) {
+            logger.info("璋冪敤ZLM-TCP琚姩鎺ㄦ祦鎺ュ彛, 缁撴灉锛� {}",  jsonObject);
+            logger.info("鍚姩鐩戝惉TCP琚姩鎺ㄦ祦鎴愬姛[ {}/{} ]锛寋}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port"));
+        } else {
+            logger.error("鍚姩鐩戝惉TCP琚姩鎺ㄦ祦澶辫触: {}, 鍙傛暟锛歿}",jsonObject.getString("msg"), JSON.toJSONString(param));
+        }
+    }
+
     /**
      * 瀹夋帓鎺ㄦ祦
      */
@@ -552,10 +602,10 @@
                             CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
                             int port, Boolean tcpActive, boolean mediaTransmissionTCP,
                             String channelId, String addressStr, String ssrc, String requesterId) {
-            Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+            Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
             if (streamReady != null && streamReady) {
                 // 鑷钩鍙板唴瀹�
-                SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+                SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                         gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                 if (sendRtpItem == null) {
@@ -591,10 +641,10 @@
                             String channelId, String addressStr, String ssrc, String requesterId) {
         // 鎺ㄦ祦
         if (streamPushItem.isSelf()) {
-            Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+            Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
             if (streamReady != null && streamReady) {
                 // 鑷钩鍙板唴瀹�
-                SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+                SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                         gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                 if (sendRtpItem == null) {
@@ -646,10 +696,9 @@
             logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍚敤娴佸悗寮�濮嬫帹娴�", gbStream.getApp(), gbStream.getStream());
             // 鐩戝惉娴佷笂绾�
             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);
+            zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, hookParam) -> {
+                OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam;
+                logger.info("[涓婄骇鐐规挱]鎷夋祦浠g悊宸茬粡灏辩华锛� {}/{}", streamChangedHookParam.getApp(), streamChangedHookParam.getStream());
                 dynamicTask.stop(callIdHeader.getCallId());
                 pushProxyStream(evt, request, gbStream,  platform, callIdHeader, mediaServerItem, port, tcpActive,
                         mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
@@ -668,9 +717,6 @@
                 zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
                 dynamicTask.stop(callIdHeader.getCallId());
             }
-
-
-
         } else if ("push".equals(gbStream.getStreamType())) {
             if (!platform.isStartOfflinePush()) {
                 // 骞冲彴璁剧疆涓叧闂簡鎷夎捣绂荤嚎鐨勬帹娴佸垯鐩存帴鍥炲
@@ -693,13 +739,10 @@
             dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
                 logger.info("[ app={}, stream={} ] 绛夊緟璁惧寮�濮嬫帹娴佽秴鏃�", gbStream.getApp(), gbStream.getStream());
                 try {
+                    redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
                     mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
                     responseAck(request, Response.REQUEST_TIMEOUT); // 瓒呮椂
-                } catch (SipException e) {
-                    logger.error("鏈鐞嗙殑寮傚父 ", e);
-                } catch (InvalidArgumentException e) {
-                    logger.error("鏈鐞嗙殑寮傚父 ", e);
-                } catch (ParseException e) {
+                } catch (SipException | InvalidArgumentException | ParseException e) {
                     logger.error("鏈鐞嗙殑寮傚父 ", e);
                 }
             }, userSetting.getPlatformPlayTimeout());
@@ -710,8 +753,9 @@
             // 娣诲姞鍦ㄦ湰鏈轰笂绾跨殑閫氱煡
             mediaListManager.addChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream(), (app, stream, serverId) -> {
                 dynamicTask.stop(callIdHeader.getCallId());
+                redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
                 if (serverId.equals(userSetting.getServerId())) {
-                    SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
+                    SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
                             app, stream, channelId, mediaTransmissionTCP, platform.isRtcp());
 
                     if (sendRtpItem == null) {

--
Gitblit v1.8.0