From 06bbe3fe01e5af9486c309693a975077df813f7c Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期四, 29 九月 2022 16:27:59 +0800
Subject: [PATCH] 添加第二种语音对讲实现

---
 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java |  227 ++++++++++++++++++++++++++++++++++++++++++++++++++++----
 1 files changed, 208 insertions(+), 19 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
index b789974..5194a9a 100644
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
@@ -1,5 +1,6 @@
 package com.genersoft.iot.vmp.media.zlm;
 
+import java.text.ParseException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -7,14 +8,19 @@
 import com.alibaba.fastjson.JSON;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
 import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
+import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
-import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
+import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
 import com.genersoft.iot.vmp.media.zlm.dto.*;
 import com.genersoft.iot.vmp.service.*;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
+import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -31,6 +37,8 @@
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
 
 import javax.servlet.http.HttpServletRequest;
+import javax.sip.InvalidArgumentException;
+import javax.sip.SipException;
 
 /**    
  * @description:閽堝 ZLMediaServer鐨刪ook浜嬩欢鐩戝惉
@@ -47,7 +55,13 @@
 	private SIPCommander cmder;
 
 	@Autowired
-	private SIPCommanderFroPlatform commanderFroPlatform;
+	private ISIPCommanderForPlatform commanderFroPlatform;
+
+	@Autowired
+	private AudioBroadcastManager audioBroadcastManager;
+
+	@Autowired
+	private ZLMRTPServerFactory zlmrtpServerFactory;
 
 	@Autowired
 	private IPlayService playService;
@@ -57,6 +71,9 @@
 
 	@Autowired
 	private IRedisCatchStorage redisCatchStorage;
+
+	@Autowired
+	private IDeviceService deviceService;
 
 	@Autowired
 	private IMediaServerService mediaServerService;
@@ -103,7 +120,7 @@
 	@PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
 	public JSONObject onServerKeepalive(@RequestBody JSONObject json){
 
-		logger.info("[ ZLM HOOK ] on_server_keepalive API璋冪敤锛屽弬鏁帮細" + json.toString());
+		logger.info("[ ZLM HOOK ]on_server_keepalive API璋冪敤锛屽弬鏁帮細" + json.toString());
 		String mediaServerId = json.getString("mediaServerId");
 		List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive);
 		if (subscribes != null  && subscribes.size() > 0) {
@@ -417,10 +434,11 @@
 		String schema = item.getSchema();
 		List<MediaItem.MediaTrack> tracks = item.getTracks();
 		boolean regist = item.isRegist();
-		if (item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
-				|| item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
-				|| item.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
-			if (regist) {
+		if (regist) {
+			if (item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
+					|| item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
+					|| item.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
+
 				StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
 				if (streamAuthorityInfo == null) {
 					streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(item);
@@ -429,9 +447,9 @@
 					streamAuthorityInfo.setOriginTypeStr(item.getOriginTypeStr());
 				}
 				redisCatchStorage.updateStreamAuthorityInfo(app, stream, streamAuthorityInfo);
-			}else {
-				redisCatchStorage.removeStreamAuthorityInfo(app, stream);
 			}
+		}else {
+			redisCatchStorage.removeStreamAuthorityInfo(app, stream);
 		}
 
 		if ("rtsp".equals(schema)){
@@ -451,17 +469,134 @@
 				if (streamInfo!=null){
 					redisCatchStorage.stopPlay(streamInfo);
 					storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
-					// 濡傛灉姝e湪缁欎笂绾ф帹閫侊紝鍒欏彂閫乥ye
-
 				}else{
 					streamInfo = redisCatchStorage.queryPlayback(null, null, stream, null);
 					if (streamInfo != null) {
 						redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
 								streamInfo.getStream(), null);
 					}
-					// 濡傛灉姝e湪缁欎笂绾ф帹閫侊紝鍒欏彂閫乥ye
 				}
-			}else {
+			}else if ("broadcast".equals(app)){
+				// 璇煶瀵硅鎺ㄦ祦  stream闇�瑕佹弧瓒虫牸寮廳eviceId_channelId
+				if (regist && stream.indexOf("_") > 0) {
+					String[] streamArray = stream.split("_");
+					if (streamArray.length == 2) {
+						String deviceId = streamArray[0];
+						String channelId = streamArray[1];
+						Device device = deviceService.queryDevice(deviceId);
+						if (device != null) {
+							DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+							if (deviceChannel != null) {
+								if (audioBroadcastManager.exit(deviceId, channelId)) {
+									// 鐩存帴鎺ㄦ祦
+									SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, stream, null);
+									if (sendRtpItem == null) {
+										// TODO 鍙兘鏁版嵁閿欒锛岄噸鏂板紑鍚闊抽�氶亾
+									}else {
+										String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
+										MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+										logger.info("rtp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc());
+										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());
+										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;
+										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 && jsonObject.getInteger("code") == 0) {
+											logger.info("[璇煶瀵硅] 鑷姩鎺ㄦ祦鎴愬姛, device: {}, channel: {}", deviceId, channelId);
+										}
+
+									}
+								}else {
+									// 寮�鍚闊冲璁查�氶亾
+									try {
+										playService.audioBroadcastCmd(device, channelId, 60, (msg)->{
+											logger.info("[璇煶瀵硅] 閫氶亾寤虹珛鎴愬姛, device: {}, channel: {}", deviceId, channelId);
+										});
+									} catch (InvalidArgumentException | ParseException | SipException e) {
+										logger.error("[鍛戒护鍙戦�佸け璐 璇煶瀵硅: {}", e.getMessage());
+									}
+								}
+
+							}
+						}
+					}
+				}
+
+			}else if ("talk".equals(app)){
+				// 璇煶瀵硅鎺ㄦ祦  stream闇�瑕佹弧瓒虫牸寮廳eviceId_channelId
+				if (regist && stream.indexOf("_") > 0) {
+					String[] streamArray = stream.split("_");
+					if (streamArray.length == 2) {
+						String deviceId = streamArray[0];
+						String channelId = streamArray[1];
+						Device device = deviceService.queryDevice(deviceId);
+						if (device != null) {
+							DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+							if (deviceChannel != null) {
+								if (audioBroadcastManager.exit(deviceId, channelId)) {
+									// 鐩存帴鎺ㄦ祦
+									SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, stream, null);
+									if (sendRtpItem == null) {
+										// TODO 鍙兘鏁版嵁閿欒锛岄噸鏂板紑鍚闊抽�氶亾
+									}else {
+										String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
+										MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+										logger.info("rtp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc());
+										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());
+										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;
+										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 && jsonObject.getInteger("code") == 0) {
+											logger.info("[璇煶瀵硅] 鑷姩鎺ㄦ祦鎴愬姛, device: {}, channel: {}", deviceId, channelId);
+										}
+									}
+								}else {
+									// 寮�鍚闊冲璁查�氶亾
+									MediaServerItem mediaServerForMinimumLoad = mediaServerService.getMediaServerForMinimumLoad();
+									playService.talk(mediaServerForMinimumLoad, device, channelId, (mediaServerItem, jsonObject)->{
+										System.out.println("寮�濮嬫帹娴�");
+									}, eventResult -> {
+										System.out.println(eventResult.msg);
+									}, ()->{
+										System.out.println("瓒呮椂");
+									});
+								}
+
+							}
+						}
+					}
+				}
+
+			}else{
 				if (!"rtp".equals(app)){
 					String type = OriginType.values()[item.getOriginType()].getType();
 					MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
@@ -509,6 +644,36 @@
 					}
 				}
 			}
+			if (!regist) {
+				List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(stream);
+				if (sendRtpItems.size() > 0) {
+					for (SendRtpItem sendRtpItem : sendRtpItems) {
+						if (sendRtpItem.getApp().equals(app)) {
+							String platformId = sendRtpItem.getPlatformId();
+							ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
+							Device device = deviceService.queryDevice(platformId);
+
+							try {
+								if (platform != null) {
+									commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
+								}else {
+									if (sendRtpItem.isOnlyAudio()) {
+										AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
+										if (device != null && audioBroadcastCatch != null) {
+//											cmder.streamByeCmd(device, sendRtpItem.getChannelId(), audioBroadcastCatch.getSipTransactionInfo(), null);
+										}
+									}else {
+										cmder.streamByeCmd(device, sendRtpItem.getChannelId(), stream, sendRtpItem.getCallId());
+									}
+
+								}
+							} catch (SipException | InvalidArgumentException | ParseException | SsrcTransactionNotFoundException e) {
+								logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
+							}
+						}
+					}
+				}
+			}
 		}
 
 		JSONObject ret = new JSONObject();
@@ -541,21 +706,41 @@
 					if (sendRtpItems.size() > 0) {
 						for (SendRtpItem sendRtpItem : sendRtpItems) {
 							ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
-							commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+							try {
+								commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+							} catch (SipException | InvalidArgumentException | ParseException e) {
+								logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
+							}
 							redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
 									sendRtpItem.getCallId(), sendRtpItem.getStreamId());
 						}
 					}
 				}
-				cmder.streamByeCmd(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId(),
-						streamInfoForPlayCatch.getStream(), null);
+				Device device = deviceService.queryDevice(streamInfoForPlayCatch.getDeviceID());
+				if (device != null) {
+					try {
+						cmder.streamByeCmd(device, streamInfoForPlayCatch.getChannelId(),
+								streamInfoForPlayCatch.getStream(), null);
+					} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
+						logger.error("[鏃犱汉瑙傜湅]鐐规挱锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
+					}
+				}
+
 				redisCatchStorage.stopPlay(streamInfoForPlayCatch);
 				storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());
 			}else{
 				StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, streamId, null);
 				if (streamInfoForPlayBackCatch != null) {
-					cmder.streamByeCmd(streamInfoForPlayBackCatch.getDeviceID(),
-							streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null);
+					Device device = deviceService.queryDevice(streamInfoForPlayCatch.getDeviceID());
+					if (device != null) {
+						try {
+							cmder.streamByeCmd(device,streamInfoForPlayBackCatch.getChannelId(),
+									streamInfoForPlayBackCatch.getStream(), null);
+						} catch (InvalidArgumentException | ParseException | SipException |
+								 SsrcTransactionNotFoundException e) {
+							logger.error("[鏃犱汉瑙傜湅]鍥炴斁锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
+						}
+					}
 					redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(),
 							streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null);
 				}else {
@@ -678,7 +863,11 @@
 		if (sendRtpItems.size() > 0) {
 			for (SendRtpItem sendRtpItem : sendRtpItems) {
 				ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
-				commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+				try {
+					commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+				} catch (SipException | InvalidArgumentException | ParseException e) {
+					logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
+				}
 				redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
 						sendRtpItem.getCallId(), sendRtpItem.getStreamId());
 			}

--
Gitblit v1.8.0