From 100252a253263321873e79d43dff94e19defe353 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 22 二月 2023 18:06:52 +0800
Subject: [PATCH] 完善语音对讲
---
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 289 ++++++++++++++++++++++++++++++++++++++-------------------
1 files changed, 194 insertions(+), 95 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..8770d0e 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.cmd.ISIPCommanderForPlatform;
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.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;
@@ -267,18 +274,17 @@
logger.info("[ZLM HOOK] 娴佹敞閿�, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
}
-
+ 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 +303,151 @@
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()) {
- // 璁剧疆鎷夋祦浠ｇ悊涓婄嚎/绂荤嚎
- 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 (!"rtp".equals(param.getApp())) {
- String type = OriginType.values()[param.getOriginType()].getType();
- MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
+ 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()) {
+ // 璁剧疆鎷夋祦浠ｇ悊涓婄嚎/绂荤嚎
+ 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, 60, mediaInfo, param.getApp(), param.getStream(), (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 ("talk".equals(param.getApp())){
+ // 璇煶瀵硅鎺ㄦ祦 stream闇�瑕佹弧瓒虫牸寮廳eviceId_channelId
+ if (param.isRegist() && 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) {
+ DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+ if (deviceChannel != null) {
+ if (audioBroadcastManager.exit(deviceId, channelId)) {
+ // 鐩存帴鎺ㄦ祦
+ SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);
+ if (sendRtpItem == null) {
+ // TODO 鍙兘鏁版嵁閿欒锛岄噸鏂板紑鍚闊抽�氶亾
+ }else {
+ MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+ logger.info("rtp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc());
+ Map<String, Object> sendParam = new HashMap<>(12);
+ sendParam.put("vhost","__defaultVhost__");
+ sendParam.put("app",sendRtpItem.getApp());
+ sendParam.put("stream",sendRtpItem.getStreamId());
+ sendParam.put("ssrc", sendRtpItem.getSsrc());
+ sendParam.put("src_port", sendRtpItem.getLocalPort());
+ sendParam.put("pt", sendRtpItem.getPt());
+ sendParam.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
+ sendParam.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
- 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(), 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) {
+ JSONObject jsonObject;
+ if (sendRtpItem.isTcpActive()) {
+ jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaServerItem, sendParam);
+ } else {
+ sendParam.put("is_udp", sendRtpItem.isTcp() ? "0" : "1");
+ sendParam.put("dst_url", sendRtpItem.getIp());
+ sendParam.put("dst_port", sendRtpItem.getPort());
+ jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaServerItem, sendParam);
+ }
+ if (jsonObject != null && jsonObject.getInteger("code") == 0) {
+ logger.info("[璇煶瀵硅] 鑷姩鎺ㄦ祦鎴愬姛, device: {}, channel: {}", deviceId, channelId);
+ }
+ }
+ }else {
+ // 寮�鍚闊冲璁查�氶亾
+ MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
+ playService.talk(mediaServerItem, device, channelId, (mediaServer, jsonObject)->{
+ System.out.println("寮�濮嬫帹娴�");
+ }, eventResult -> {
+ System.out.println(eventResult.msg);
+ }, ()->{
+ System.out.println("瓒呮椂");
+ });
+ }
+
+ }
+ }
+ }
+ }
+
+ }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());
@@ -404,42 +501,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.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());
+ }
+ }
redisCatchStorage.stopPlay(streamInfoForPlayCatch);
storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());
--
Gitblit v1.8.0