From 82adc0cb23f3ee47322e78889cdaba57e9309000 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 21 三月 2023 15:55:24 +0800
Subject: [PATCH] 完善语音对讲级联

---
 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java |  291 +++++++++++++++++++++++++++++++++++++--------------------
 1 files changed, 189 insertions(+), 102 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 a795e77..984a377 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
@@ -7,11 +7,12 @@
 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.callback.DeferredResultHolder;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
-import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
 import com.genersoft.iot.vmp.media.zlm.dto.HookType;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
@@ -56,8 +57,14 @@
     @Autowired
     private SIPCommander cmder;
 
-    @Autowired
-    private SIPCommanderFroPlatform commanderFroPlatform;
+	@Autowired
+	private ISIPCommanderForPlatform commanderFroPlatform;
+
+	@Autowired
+	private AudioBroadcastManager audioBroadcastManager;
+
+	@Autowired
+	private ZLMRTPServerFactory zlmrtpServerFactory;
 
     @Autowired
     private IPlayService playService;
@@ -242,6 +249,7 @@
             String channelId = ssrcTransactionForAll.get(0).getChannelId();
             DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
             if (deviceChannel != null) {
+
                 result.setEnable_audio(deviceChannel.isHasAudio());
             }
             // 濡傛灉鏄綍鍍忎笅杞藉氨璁剧疆瑙嗛闂撮殧鍗佺
@@ -250,6 +258,11 @@
                 result.setEnable_audio(true);
                 result.setEnable_mp4(true);
             }
+            // 濡傛灉鏄痶alk瀵硅锛屽垯榛樿鑾峰彇澹伴煶
+            if (ssrcTransactionForAll.get(0).getType() == VideoStreamSessionManager.SessionType.talk) {
+                result.setEnable_audio(true);
+            }
+
         }
         return result;
     }
@@ -267,18 +280,20 @@
             logger.info("[ZLM HOOK] 娴佹敞閿�, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
         }
 
-
+		JSONObject ret = new JSONObject();
+		ret.put("code", 0);
+		ret.put("msg", "success");
+        MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());
         JSONObject json = (JSONObject) JSON.toJSON(param);
         taskExecutor.execute(() -> {
             ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json);
             if (subscribe != null) {
-                MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());
+
                 if (mediaInfo != null) {
                     subscribe.response(mediaInfo, json);
                 }
             }
             // 娴佹秷澶辩Щ闄edis play
-            List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();
             if (param.isRegist()) {
                 if (param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
                         || param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
@@ -297,60 +312,119 @@
                 redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());
             }
 
-            if ("rtsp".equals(param.getSchema())) {
-                if (param.isRegist()) {
-                    mediaServerService.addCount(param.getMediaServerId());
-                } else {
-                    mediaServerService.removeCount(param.getMediaServerId());
-                }
-                if (param.getOriginType() == OriginType.PULL.ordinal()
-                        || param.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {
-                    // 璁剧疆鎷夋祦浠g悊涓婄嚎/绂荤嚎
-                    streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
-                }
-                if ("rtp".equals(param.getApp()) && !param.isRegist()) {
-                    StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());
-                    if (streamInfo != null) {
-                        redisCatchStorage.stopPlay(streamInfo);
-                        storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
-                    } else {
-                        streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);
-                        if (streamInfo != null) {
-                            redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
-                                    streamInfo.getStream(), null);
+		if ("rtsp".equals(param.getSchema())){
+			logger.info("娴佸彉鍖栵細娉ㄥ唽->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream());
+			if (param.isRegist()) {
+				mediaServerService.addCount(param.getMediaServerId());
+			}else {
+				mediaServerService.removeCount(param.getMediaServerId());
+			}
+			if (param.getOriginType() == OriginType.PULL.ordinal()
+					|| param.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {
+				// 璁剧疆鎷夋祦浠g悊涓婄嚎/绂荤嚎
+				streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
+			}
+			if ("rtp".equals(param.getApp()) && !param.isRegist() ) {
+				StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());
+				if (streamInfo!=null){
+					redisCatchStorage.stopPlay(streamInfo);
+					storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
+				}else{
+					streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);
+					if (streamInfo != null) {
+						redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
+								streamInfo.getStream(), null);
+					}
+				}
+			}else if ("broadcast".equals(param.getApp())){
+				// 璇煶瀵硅鎺ㄦ祦  stream闇�瑕佹弧瓒虫牸寮廳eviceId_channelId
+                if (param.getStream().indexOf("_") > 0) {
+                    String[] streamArray = param.getStream().split("_");
+                    if (streamArray.length == 2) {
+                        String deviceId = streamArray[0];
+                        String channelId = streamArray[1];
+                        Device device = deviceService.getDevice(deviceId);
+                        if (device != null) {
+                            if (param.isRegist()) {
+                                if (audioBroadcastManager.exit(deviceId, channelId)) {
+                                    playService.stopAudioBroadcast(deviceId, channelId);
+                                }
+                                // 寮�鍚闊冲璁查�氶亾
+                                try {
+                                    playService.audioBroadcastCmd(device, channelId, mediaInfo, param.getApp(), param.getStream(), 60, false, (msg)->{
+                                        logger.info("[璇煶瀵硅] 閫氶亾寤虹珛鎴愬姛, device: {}, channel: {}", deviceId, channelId);
+                                    });
+                                } catch (InvalidArgumentException | ParseException | SipException e) {
+                                    logger.error("[鍛戒护鍙戦�佸け璐 璇煶瀵硅: {}", e.getMessage());
+                                }
+                            }else {
+                                // 娴佹敞閿�
+                                playService.stopAudioBroadcast(deviceId, channelId);
+                            }
+                        } else{
+                            logger.info("[璇煶瀵硅] 鏈壘鍒拌澶囷細{}", deviceId);
                         }
                     }
-                } else {
-                    if (!"rtp".equals(param.getApp())) {
-                        String type = OriginType.values()[param.getOriginType()].getType();
-                        MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
-
-                        if (mediaServerItem != null) {
+                }
+			}else if ("talk".equals(param.getApp())){
+				// 璇煶瀵硅鎺ㄦ祦  stream闇�瑕佹弧瓒虫牸寮廳eviceId_channelId
+                if (param.getStream().indexOf("_") > 0) {
+                    String[] streamArray = param.getStream().split("_");
+                    if (streamArray.length == 2) {
+                        String deviceId = streamArray[0];
+                        String channelId = streamArray[1];
+                        Device device = deviceService.getDevice(deviceId);
+                        if (device != null) {
                             if (param.isRegist()) {
-                                StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
-                                String callId = null;
-                                if (streamAuthorityInfo != null) {
-                                    callId = streamAuthorityInfo.getCallId();
+                                if (audioBroadcastManager.exit(deviceId, channelId)) {
+                                    playService.stopAudioBroadcast(deviceId, channelId);
                                 }
-                                StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,
-                                        param.getApp(), param.getStream(), tracks, callId);
-                                param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));
-                                redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);
-                                if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
-                                        || param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
-                                        || param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
-                                    param.setSeverId(userSetting.getServerId());
-                                    zlmMediaListManager.addPush(param);
-                                }
-                            } else {
-                                // 鍏煎娴佹敞閿�鏃剁被鍨嬩粠redis璁板綍鑾峰彇
-                                OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(param.getApp(), param.getStream(), param.getMediaServerId());
-                                if (onStreamChangedHookParam != null) {
-                                    type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();
-                                    redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());
-                                }
-                                GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
-                                if (gbStream != null) {
+                                // 寮�鍚闊冲璁查�氶亾
+                                playService.talkCmd(device, channelId, mediaInfo, param.getStream(), (msg)->{
+                                    logger.info("[璇煶瀵硅] 閫氶亾寤虹珛鎴愬姛, device: {}, channel: {}", deviceId, channelId);
+                                });
+                            }else {
+                                // 娴佹敞閿�
+                                playService.stopTalk(device, channelId, param.isRegist());
+                            }
+                        } else{
+                            logger.info("[璇煶瀵硅] 鏈壘鍒拌澶囷細{}", deviceId);
+                        }
+                    }
+                }
+
+			}else{
+				if (!"rtp".equals(param.getApp())){
+					String type = OriginType.values()[param.getOriginType()].getType();
+					MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
+
+						if (mediaServerItem != null){
+							if (param.isRegist()) {
+								StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
+								String callId = null;
+								if (streamAuthorityInfo != null) {
+									callId = streamAuthorityInfo.getCallId();
+								}
+								StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,
+										param.getApp(), param.getStream(), param.getTracks(), callId);
+								param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));
+								redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);
+								if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
+										|| param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
+										|| param.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
+									param.setSeverId(userSetting.getServerId());
+									zlmMediaListManager.addPush(param);
+								}
+							}else {
+								// 鍏煎娴佹敞閿�鏃剁被鍨嬩粠redis璁板綍鑾峰彇
+								OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(
+										param.getApp(), param.getStream(), param.getMediaServerId());
+								if (onStreamChangedHookParam != null) {
+									type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();
+									redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());
+								}
+								GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
+								if (gbStream != null) {
 //									eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
                                 }
                                 zlmMediaListManager.removeMedia(param.getApp(), param.getStream());
@@ -377,16 +451,21 @@
                                 ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
                                 Device device = deviceService.getDevice(platformId);
 
-                                try {
+
                                     if (platform != null) {
-                                        commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
+                                        try {
+                                            commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
+                                        } catch (SipException | InvalidArgumentException | ParseException e) {
+                                            logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
+                                        }
                                     } else {
-                                        cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
+                                        try {
+                                            cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
+                                        } catch (SipException | InvalidArgumentException | ParseException |
+                                                 SsrcTransactionNotFoundException e) {
+                                            logger.error("[鍛戒护鍙戦�佸け璐 鍙戦�丅YE: {}", e.getMessage());
+                                        }
                                     }
-                                } catch (SipException | InvalidArgumentException | ParseException |
-                                         SsrcTransactionNotFoundException e) {
-                                    logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
-                                }
                             }
                         }
                     }
@@ -404,42 +483,44 @@
     @PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8")
     public JSONObject onStreamNoneReader(@RequestBody OnStreamNoneReaderHookParam param) {
 
-        logger.info("[ZLM HOOK]娴佹棤浜鸿鐪嬶細{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
-        JSONObject ret = new JSONObject();
-        ret.put("code", 0);
-        // 鍥芥爣绫诲瀷鐨勬祦
-        if ("rtp".equals(param.getApp())) {
-            ret.put("close", userSetting.getStreamOnDemand());
-            // 鍥芥爣娴侊紝 鐐规挱/褰曞儚鍥炴斁/褰曞儚涓嬭浇
-            StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream());
-            // 鐐规挱
-            if (streamInfoForPlayCatch != null) {
-                // 鏀跺埌鏃犱汉瑙傜湅璇存槑娴佷篃娌℃湁鍦ㄥ線涓婄骇鎺ㄩ��
-                if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {
-                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(streamInfoForPlayCatch.getChannelId());
-                    if (sendRtpItems.size() > 0) {
-                        for (SendRtpItem sendRtpItem : sendRtpItems) {
-                            ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
-                            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());
-                        }
-                    }
-                }
-                Device device = deviceService.getDevice(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());
-                    }
-                }
+		logger.info("[ZLM HOOK]娴佹棤浜鸿鐪嬶細{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
+		JSONObject ret = new JSONObject();
+		ret.put("code", 0);
+		// 鍥芥爣绫诲瀷鐨勬祦
+		if ("rtp".equals(param.getApp())){
+			ret.put("close", userSetting.getStreamOnDemand());
+			// 鍥芥爣娴侊紝 鐐规挱/褰曞儚鍥炴斁/褰曞儚涓嬭浇
+			StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream());
+			// 鐐规挱
+			if (streamInfoForPlayCatch != null) {
+				// 鏀跺埌鏃犱汉瑙傜湅璇存槑娴佷篃娌℃湁鍦ㄥ線涓婄骇鎺ㄩ��
+				if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {
+					List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(streamInfoForPlayCatch.getChannelId());
+					if (sendRtpItems.size() > 0) {
+						for (SendRtpItem sendRtpItem : sendRtpItems) {
+							ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+							if (parentPlatform == null) {
+								continue;
+							}
+							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.getStream());
+						}
+					}
+				}
+				Device device = deviceService.getDevice(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());
@@ -456,8 +537,7 @@
                         try {
                             cmder.streamByeCmd(device, streamInfoForPlayBackCatch.getChannelId(),
                                     streamInfoForPlayBackCatch.getStream(), null);
-                        } catch (InvalidArgumentException | ParseException | SipException |
-                                 SsrcTransactionNotFoundException e) {
+                        } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
                             logger.error("[鏃犱汉瑙傜湅]鍥炴斁锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
                         }
                     }
@@ -473,6 +553,13 @@
                 ret.put("close", false);
                 return ret;
             }
+            SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);
+            if ("talk".equals(sendRtpItem.getApp())){
+                ret.put("close", false);
+                return ret;
+            }
+        }else if ("talk".equals(param.getApp()) || "broadcast".equals(param.getApp())){
+            ret.put("close", false);
         } else {
             // 闈炲浗鏍囨祦 鎺ㄦ祦/鎷夋祦浠g悊
             // 鎷夋祦浠g悊
@@ -634,7 +721,7 @@
                         logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
                     }
                     redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
-                            sendRtpItem.getCallId(), sendRtpItem.getStreamId());
+                            sendRtpItem.getCallId(), sendRtpItem.getStream());
                 }
             }
         });

--
Gitblit v1.8.0