From 9c555b56b70ed05c1b13f0a26098b702d7364839 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 18 五月 2022 18:14:46 +0800
Subject: [PATCH] 优化语音广播的TCP主动模式

---
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java |  104 +++++++++++++++++++++++++++++++++-
 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java                              |    7 ++
 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java                                  |    4 +
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java    |   22 +++++-
 4 files changed, 127 insertions(+), 10 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
index 153a08a..85ee647 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
@@ -107,29 +107,41 @@
 			SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId());
 			String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
 			MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-			logger.info("鏀跺埌ACK锛屽紑濮嬪悜涓婄骇鎺ㄦ祦 rtp/{}", sendRtpItem.getStreamId());
+			logger.info("[鏀跺埌ACK]锛屽紑濮嬩娇鐢▄}鍚戜笂绾ф帹娴� {}/{}->{}:{}({})", sendRtpItem.isTcp() ? "TCP" : "UDP",
+					sendRtpItem.getApp(), sendRtpItem.getStreamId(),
+					sendRtpItem.getIp() ,sendRtpItem.getPort(),
+					sendRtpItem.getSsrc());
 			Map<String, Object> param = new HashMap<>();
 			param.put("vhost","__defaultVhost__");
 			param.put("app",sendRtpItem.getApp());
 			param.put("stream",sendRtpItem.getStreamId());
 			param.put("ssrc", sendRtpItem.getSsrc());
-			param.put("dst_url",sendRtpItem.getIp());
-			param.put("dst_port", sendRtpItem.getPort());
-			param.put("is_udp", is_Udp);
 			param.put("src_port", sendRtpItem.getLocalPort());
 			param.put("pt", sendRtpItem.getPt());
 			param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
 			param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
-			JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+			JSONObject jsonObject;
+			if (sendRtpItem.isTcpActive()) {
+				jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
+			}else {
+				param.put("is_udp", is_Udp);
+				param.put("dst_url",sendRtpItem.getIp());
+				param.put("dst_port", sendRtpItem.getPort());
+				jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+			}
+
 			if (jsonObject == null) {
 				logger.error("RTP鎺ㄦ祦澶辫触: 璇锋鏌LM鏈嶅姟");
 			} else if (jsonObject.getInteger("code") == 0) {
+
 				if (sendRtpItem.isOnlyAudio()) {
 					AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
 					audioBroadcastCatch.setStatus(AudioBroadcastCatchStatus.Ok);
 					audioBroadcastCatch.setDialog((SIPDialog) evt.getDialog());
 					audioBroadcastCatch.setRequest((SIPRequest) evt.getRequest());
 					audioBroadcastManager.update(audioBroadcastCatch);
+					String waiteStreamTimeoutTaskKey = "waite-stream-" + audioBroadcastCatch.getDeviceId() + audioBroadcastCatch.getChannelId();
+					dynamicTask.stop(waiteStreamTimeoutTaskKey);
 				}
 				logger.info("RTP鎺ㄦ祦鎴愬姛[ {}/{} ]锛寋}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port"));
 			} else {
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 3d71c29..eabfd1a 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,5 +1,7 @@
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.SipConfig;
@@ -20,7 +22,9 @@
 import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
 import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
 import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
+import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
 import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItemLite;
 import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -89,6 +93,9 @@
 
 	@Autowired
 	private ZLMRTPServerFactory zlmrtpServerFactory;
+
+	@Autowired
+	private ZLMRESTfulUtils zlmresTfulUtils;
 
 	@Autowired
 	private IMediaServerService mediaServerService;
@@ -674,7 +681,19 @@
 			subscribeKey.put("mediaServerId", mediaServerItem.getId());
 			String finalSsrc = ssrc;
 			// 娴佸凡缁忓瓨鍦ㄦ椂鐩存帴鎺ㄦ祦
-			if (zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream)) {
+			JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtsp", stream);
+			JSONArray tracks = mediaInfo.getJSONArray("tracks");
+			Integer codecId = null;
+			if (tracks != null && tracks.size() > 0) {
+				for (int i = 0; i < tracks.size(); i++) {
+					MediaItem.MediaTrack track = JSON.toJavaObject((JSON)tracks.get(i),MediaItem.MediaTrack.class);
+					if (track.getCodecType() == 1) {
+						codecId = track.getCodecId();
+						break;
+					}
+				}
+			}
+			if ((mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"))) {
 				logger.info("鍙戠幇宸茬粡鍦ㄦ帹娴�");
 				sendRtpItem.setStatus(2);
 				redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -684,9 +703,40 @@
 				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=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
+				if (codecId == null) {
+					if (mediaTransmissionTCP) {
+						content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
+					}else {
+						content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
+					}
+
+					content.append("a=rtpmap:8 PCMA/8000\r\n");
+				}else {
+					if (codecId == 4) {
+						if (mediaTransmissionTCP) {
+							content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
+						}else {
+							content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
+						}
+						content.append("a=rtpmap:0 PCMU/8000\r\n");
+					}else {
+						if (mediaTransmissionTCP) {
+							content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
+						}else {
+							content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
+						}
+						content.append("a=rtpmap:8 PCMA/8000\r\n");
+					}
+				}
+				if (sendRtpItem.isTcp()) {
+					content.append("a=connection:new\r\n");
+					if (!sendRtpItem.isTcpActive()) {
+						content.append("a=setup:active\r\n");
+					}else {
+						content.append("a=setup:passive\r\n");
+					}
+				}
 				content.append("a=sendonly\r\n");
-				content.append("a=rtpmap:8 PCMA/8000\r\n");
 				content.append("y="+ finalSsrc + "\r\n");
 				content.append("f=v/////a/1/8/1\r\n");
 
@@ -727,9 +777,22 @@
 					}
 				}, 20*1000);
 
+				boolean finalMediaTransmissionTCP = mediaTransmissionTCP;
 				subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
 						(MediaServerItem mediaServerItemInUse, JSONObject json)->{
 					logger.info("鏀跺埌璇煶瀵硅鎺ㄦ祦");
+					MediaItem mediaItem = JSON.toJavaObject(json, MediaItem.class);
+					Integer audioCodecId = null;
+					if (mediaItem.getTracks() != null && mediaItem.getTracks().size() > 0) {
+						for (int i = 0; i < mediaItem.getTracks().size(); i++) {
+							MediaItem.MediaTrack mediaTrack = mediaItem.getTracks().get(i);
+							if (mediaTrack.getCodecType() == 1) {
+								audioCodecId = mediaTrack.getCodecId();
+								break;
+							}
+						}
+					}
+
 					try {
 						sendRtpItem.setStatus(2);
 						redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -739,9 +802,40 @@
 						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=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
+						if (audioCodecId == null) {
+							if (finalMediaTransmissionTCP) {
+								content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
+							}else {
+								content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
+							}
+
+							content.append("a=rtpmap:8 PCMA/8000\r\n");
+						}else {
+							if (audioCodecId == 4) {
+								if (finalMediaTransmissionTCP) {
+									content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
+								}else {
+									content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
+								}
+								content.append("a=rtpmap:0 PCMU/8000\r\n");
+							}else {
+								if (finalMediaTransmissionTCP) {
+									content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
+								}else {
+									content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
+								}
+								content.append("a=rtpmap:8 PCMA/8000\r\n");
+							}
+						}
 						content.append("a=sendonly\r\n");
-						content.append("a=rtpmap:8 PCMA/8000\r\n");
+						if (sendRtpItem.isTcp()) {
+							content.append("a=connection:new\r\n");
+							if (!sendRtpItem.isTcpActive()) {
+								content.append("a=setup:active\r\n");
+							}else {
+								content.append("a=setup:passive\r\n");
+							}
+						}
 						content.append("y="+ finalSsrc + "\r\n");
 						content.append("f=v/////a/1/8/1\r\n");
 
diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
index 4b2bf48..75f55cf 100644
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
@@ -253,6 +253,10 @@
         return sendPost(mediaServerItem, "startSendRtp",param, null);
     }
 
+    public JSONObject startSendRtpPassive(MediaServerItem mediaServerItem, Map<String, Object> param) {
+        return sendPost(mediaServerItem, "startSendRtpPassive",param, null);
+    }
+
     public JSONObject stopSendRtp(MediaServerItem mediaServerItem, Map<String, Object> param) {
         return sendPost(mediaServerItem, "stopSendRtp",param, null);
     }
diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
index 3a91f6a..34918ae 100644
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
@@ -250,6 +250,13 @@
     }
 
     /**
+     * 璋冪敤zlm RESTFUL API 鈥斺�� startSendRtpPassive
+     */
+    public JSONObject startSendRtpPassive(MediaServerItem mediaServerItem, Map<String, Object>param) {
+        return zlmresTfulUtils.startSendRtpPassive(mediaServerItem, param);
+    }
+
+    /**
      * 鏌ヨ寰呰浆鎺ㄧ殑娴佹槸鍚﹀氨缁�
      */
     public Boolean isRtpReady(MediaServerItem mediaServerItem, String streamId) {

--
Gitblit v1.8.0