From 764d04b497356ba6bcbb75fd42b51eca750f7223 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 29 五月 2024 15:02:51 +0800
Subject: [PATCH] 调整上级观看消息的发送

---
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java |  165 ++++++++++++++++++++++++++++++++----------------------
 1 files changed, 97 insertions(+), 68 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 98ae86f..d4df0ae 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
@@ -1,38 +1,44 @@
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
 
-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.conf.exception.ControllerException;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
+import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
 import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
 import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
 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.media.zlm.ZLMRTPServerFactory;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
-import com.genersoft.iot.vmp.service.IMediaServerService;
+import com.genersoft.iot.vmp.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
+import com.genersoft.iot.vmp.service.IDeviceService;
+import com.genersoft.iot.vmp.service.IPlayService;
+import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-import javax.sip.Dialog;
-import javax.sip.DialogState;
 import javax.sip.RequestEvent;
 import javax.sip.address.SipURI;
+import javax.sip.header.CallIdHeader;
 import javax.sip.header.FromHeader;
 import javax.sip.header.HeaderAddress;
 import javax.sip.header.ToHeader;
-import java.util.HashMap;
-import java.util.Map;
 
 /**
  * SIP鍛戒护绫诲瀷锛� ACK璇锋眰
+ * @author lin
  */
 @Component
 public class AckRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor {
 
-	private Logger logger = LoggerFactory.getLogger(AckRequestProcessor.class);
-	private String method = "ACK";
+	private final Logger logger = LoggerFactory.getLogger(AckRequestProcessor.class);
+	private final String method = "ACK";
 
 	@Autowired
 	private SIPProcessorObserver sipProcessorObserver;
@@ -45,79 +51,102 @@
 
 	@Autowired
     private IRedisCatchStorage redisCatchStorage;
+	@Autowired
+    private IRedisRpcService redisRpcService;
 
 	@Autowired
-	private ZLMRTPServerFactory zlmrtpServerFactory;
+    private UserSetting userSetting;
+
+	@Autowired
+	private IVideoManagerStorage storager;
+
+	@Autowired
+	private IDeviceService deviceService;
 
 	@Autowired
 	private IMediaServerService mediaServerService;
 
+	@Autowired
+	private DynamicTask dynamicTask;
+
+	@Autowired
+	private IPlayService playService;
+
 
 	/**   
 	 * 澶勭悊  ACK璇锋眰
-	 * 
-	 * @param evt
 	 */
 	@Override
 	public void process(RequestEvent evt) {
-		Dialog dialog = evt.getDialog();
-		if (dialog == null) return;
-		if (dialog.getState()== DialogState.CONFIRMED) {
-			String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
-			String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
-			SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, channelId);
-			String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
-			String deviceId = sendRtpItem.getDeviceId();
-			StreamInfo streamInfo = null;
-			if (deviceId == null) {
-				streamInfo = new StreamInfo();
-				streamInfo.setApp(sendRtpItem.getApp());
-				streamInfo.setStream(sendRtpItem.getStreamId());
-			}else {
-				streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
-				sendRtpItem.setStreamId(streamInfo.getStream());
-				streamInfo.setApp("rtp");
-			}
+		CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
+		dynamicTask.stop(callIdHeader.getCallId());
+		String fromUserId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
+		String toUserId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
+		logger.info("[鏀跺埌ACK]锛� 鏉ヨ嚜->{}", fromUserId);
+		SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId());
+		if (sendRtpItem == null) {
+			logger.warn("[鏀跺埌ACK]锛氭湭鎵惧埌鏉ヨ嚜{}锛宑allId: {}", fromUserId, callIdHeader.getCallId());
+			return;
+		}
+		// tcp涓诲姩鏃讹紝姝ゆ椂鏄骇鑱斾笅绾у钩鍙帮紝鍦ㄥ洖澶�200ok鏃讹紝鏈湴宸茬粡璇锋眰zlm寮�鍚洃鍚紝璺宠繃涓嬮潰姝ラ
+		if (sendRtpItem.isTcpActive()) {
+			logger.info("鏀跺埌ACK锛宺tp/{} TCP涓诲姩鏂瑰紡鍚庣画澶勭悊", sendRtpItem.getStream());
+			return;
+		}
+		MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+		logger.info("鏀跺埌ACK锛宺tp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}, 鍗忚:{}",
+				sendRtpItem.getStream(),
+				sendRtpItem.getIp(),
+				sendRtpItem.getPort(),
+				sendRtpItem.getSsrc(),
+				sendRtpItem.isTcp()?(sendRtpItem.isTcpActive()?"TCP涓诲姩":"TCP琚姩"):"UDP"
+		);
+		ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(fromUserId);
 
-			redisCatchStorage.updateSendRTPSever(sendRtpItem);
-			logger.info(platformGbId);
-			logger.info(channelId);
-			Map<String, Object> param = new HashMap<>();
-			param.put("vhost","__defaultVhost__");
-			param.put("app",streamInfo.getApp());
-			param.put("stream",streamInfo.getStream());
-			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());
-			// 璁惧鎺ㄦ祦鏌ヨ锛屾垚鍔熷悗鎵嶈兘杞帹
-			boolean rtpPushed = false;
-			long startTime = System.currentTimeMillis();
-			while (!rtpPushed) {
-				try {
-					if (System.currentTimeMillis() - startTime < 30 * 1000) {
-						MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-						if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStream())) {
-							rtpPushed = true;
-							logger.info("宸茶幏鍙栬澶囨帹娴乕{}/{}]锛屽紑濮嬪悜涓婄骇鎺ㄦ祦[{}:{}]",
-									streamInfo.getApp() ,streamInfo.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort());
-							zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
-						} else {
-							logger.info("绛夊緟璁惧鎺ㄦ祦[{}/{}].......",
-									streamInfo.getApp() ,streamInfo.getStream());
-							Thread.sleep(1000);
-							continue;
-						}
-					} else {
-						rtpPushed = true;
-						logger.info("璁惧鎺ㄦ祦[{}/{}]瓒呮椂锛岀粓姝㈠悜涓婄骇鎺ㄦ祦",
-								streamInfo.getApp() ,streamInfo.getStream());
-					}
-				} catch (InterruptedException e) {
-					e.printStackTrace();
+		if (parentPlatform != null) {
+			if (!userSetting.getServerId().equals(sendRtpItem.getServerId())) {
+				WVPResult wvpResult = redisRpcService.startSendRtp(sendRtpItem.getRedisKey(), sendRtpItem);
+				if (wvpResult.getCode() == 0) {
+					redisCatchStorage.sendPlatformStartPlayMsg(sendRtpItem, parentPlatform);
 				}
+			} else {
+				try {
+					if (sendRtpItem.isTcpActive()) {
+						mediaServerService.startSendRtpPassive(mediaInfo,sendRtpItem, null);
+					} else {
+						mediaServerService.startSendRtp(mediaInfo, sendRtpItem);
+					}
+					redisCatchStorage.sendPlatformStartPlayMsg(sendRtpItem, parentPlatform);
+				}catch (ControllerException e) {
+					logger.error("RTP鎺ㄦ祦澶辫触: {}", e.getMessage());
+					playService.startSendRtpStreamFailHand(sendRtpItem, parentPlatform, callIdHeader);
+				}
+			}
+		}else {
+			Device device = deviceService.getDevice(fromUserId);
+			if (device == null) {
+				logger.warn("[鏀跺埌ACK]锛氭潵鑷獅}锛岀洰鏍囦负({})鐨勬帹娴佷俊鎭负鎵惧埌娴佷綋鏈嶅姟[{}]淇℃伅",fromUserId, toUserId, sendRtpItem.getMediaServerId());
+				return;
+			}
+			// 璁剧疆涓烘敹鍒癆CK鍚庡彂閫佽闊崇殑璁惧宸茬粡鍦ㄥ彂閫�200OK寮�濮嬪彂娴佷簡
+			if (!device.isBroadcastPushAfterAck()) {
+				return;
+			}
+			if (mediaInfo == null) {
+				logger.warn("[鏀跺埌ACK]锛氭潵鑷獅}锛岀洰鏍囦负({})鐨勬帹娴佷俊鎭负鎵惧埌娴佷綋鏈嶅姟[{}]淇℃伅",fromUserId, toUserId, sendRtpItem.getMediaServerId());
+				return;
+			}
+			try {
+				if (sendRtpItem.isTcpActive()) {
+					mediaServerService.startSendRtpPassive(mediaInfo, sendRtpItem, null);
+				} else {
+					mediaServerService.startSendRtp(mediaInfo, sendRtpItem);
+				}
+			}catch (ControllerException e) {
+				logger.error("RTP鎺ㄦ祦澶辫触: {}", e.getMessage());
+				playService.startSendRtpStreamFailHand(sendRtpItem, null, callIdHeader);
 			}
 		}
 	}
+
 }

--
Gitblit v1.8.0