From f6320c7dd3dfd25e98bb63f41d79a33af5009a93 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期一, 17 十月 2022 17:02:16 +0800 Subject: [PATCH] 合并主线 --- src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 325 +++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 files changed, 311 insertions(+), 14 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java index 9ce76ae..e24fa63 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java @@ -9,12 +9,16 @@ import javax.sip.ResponseEvent; import javax.sip.SipException; +import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.exception.ServiceException; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; -import com.genersoft.iot.vmp.gb28181.bean.*; +import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; +import com.genersoft.iot.vmp.gb28181.utils.SipUtils; +import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; import com.genersoft.iot.vmp.service.IDeviceService; +import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult; import com.genersoft.iot.vmp.utils.redis.RedisUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import org.slf4j.Logger; @@ -24,6 +28,8 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; +import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; import org.springframework.web.context.request.async.DeferredResult; import com.alibaba.fastjson.JSON; @@ -31,13 +37,15 @@ import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.DynamicTask; +import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; +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.impl.SIPCommander; -import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform; +import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.utils.DateUtil; @@ -55,9 +63,8 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; +import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.AudioBroadcastEvent; import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult; - -import gov.nist.javax.sip.stack.SIPDialog; @SuppressWarnings(value = {"rawtypes", "unchecked"}) @Service @@ -72,10 +79,19 @@ private SIPCommander cmder; @Autowired - private SIPCommanderFroPlatform sipCommanderFroPlatform; + private AudioBroadcastManager audioBroadcastManager; + + @Autowired + private IDeviceService deviceService; + + @Autowired + private ISIPCommanderForPlatform sipCommanderFroPlatform; @Autowired private IRedisCatchStorage redisCatchStorage; + + @Autowired + private ZLMRTPServerFactory zlmrtpServerFactory; @Autowired private DeferredResultHolder resultHolder; @@ -95,12 +111,11 @@ @Autowired private VideoStreamSessionManager streamSession; - - @Autowired - private IDeviceService deviceService; - @Autowired private UserSetting userSetting; + + @Autowired + private SipConfig sipConfig; @Autowired private DynamicTask dynamicTask; @@ -112,6 +127,7 @@ @Qualifier("taskExecutor") @Autowired private ThreadPoolTaskExecutor taskExecutor; + @Override public PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, @@ -127,7 +143,7 @@ String uuid = UUID.randomUUID().toString(); msg.setId(uuid); playResult.setUuid(uuid); - DeferredResult<WVPResult<String>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); + DeferredResult<WVPResult<StreamInfo>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); playResult.setResult(result); // 褰曞儚鏌ヨ浠hannelId浣滀负deviceId鏌ヨ resultHolder.put(key, uuid, result); @@ -205,7 +221,6 @@ redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); streamInfo = null; - } } if (streamInfo == null) { @@ -246,6 +261,192 @@ return playResult; } + @Override + public void talk(MediaServerItem mediaServerItem, Device device, String channelId, + ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, + Runnable timeoutCallback) { + String streamId = null; + if (mediaServerItem.isRtpEnable()) { + streamId = String.format("%s_%s", device.getDeviceId(), channelId); + } + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); + logger.info("[瀵硅寮�濮媇 deviceId: {}, channelId: {},鏀舵祦绔彛锛� {}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); + // 瓒呮椂澶勭悊 + String timeOutTaskKey = UUID.randomUUID().toString(); + SSRCInfo finalSsrcInfo = ssrcInfo; + System.out.println("璁剧疆瓒呮椂浠诲姟锛� " + timeOutTaskKey); + dynamicTask.startDelay(timeOutTaskKey, () -> { + + logger.info("[瀵硅瓒呮椂] 鏀舵祦瓒呮椂 deviceId: {}, channelId: {}锛岀鍙o細{}, SSRC: {}", device.getDeviceId(), channelId, finalSsrcInfo.getPort(), finalSsrcInfo.getSsrc()); + timeoutCallback.run(); + // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧� + try { + cmder.streamByeCmd(device, channelId, finalSsrcInfo.getStream(), null); + } catch (InvalidArgumentException | ParseException | SipException e) { + logger.error("[瀵硅瓒呮椂]锛� 鍙戦�丅YE澶辫触 {}", e.getMessage()); + } catch (SsrcTransactionNotFoundException e) { + timeoutCallback.run(); + mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); + mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream()); + streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); + } + }, userSetting.getPlayTimeout()); + final String ssrc = ssrcInfo.getSsrc(); + final String stream = ssrcInfo.getStream(); + //绔彛鑾峰彇澶辫触鐨剆srcInfo 娌℃湁蹇呰鍙戦�佺偣鎾寚浠� + if (ssrcInfo.getPort() <= 0) { + logger.info("[瀵硅] 绔彛鍒嗛厤寮傚父锛宒eviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo); + return; + } + + String callId = SipUtils.getNewCallId(); + boolean pushing = false; + // 鏌ョ湅璁惧鏄惁宸茬粡鍦ㄦ帹娴� +// MediaItem mediaItem = zlmrtpServerFactory.getMediaInfo(mediaServerItem, "rtp",ssrcInfo.getStream()); +// if (mediaItem != null) { +// SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, +// mediaItem.getOriginSock().getPeer_ip(), mediaItem.getOriginSock().getPeer_port(), ssrcInfo.getSsrc(), device.getDeviceId(), +// device.getDeviceId(), channelId, +// false); +// +// sendRtpItem.setTcpActive(false); +// sendRtpItem.setCallId(callId); +// sendRtpItem.setPlayType(InviteStreamType.TALK); +// sendRtpItem.setStatus(1); +// sendRtpItem.setIp(mediaItem.getOriginSock().getPeer_ip()); +// sendRtpItem.setPort(mediaItem.getOriginSock().getPeer_port()); +// sendRtpItem.setTcpActive(false); +// sendRtpItem.setStreamId(ssrcInfo.getStream()); +// sendRtpItem.setApp("1000"); +// sendRtpItem.setStreamId("1000"); +// sendRtpItem.setSsrc(ssrc); +// sendRtpItem.setOnlyAudio(true); +// redisCatchStorage.updateSendRTPSever(sendRtpItem); +// +// 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("dst_url", sendRtpItem.getIp()); +// param.put("dst_port", sendRtpItem.getPort()); +// param.put("src_port", sendRtpItem.getLocalPort()); +// param.put("pt", sendRtpItem.getPt()); +// param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0"); +// param.put("is_udp", sendRtpItem.isTcp() ? "0" : "1"); +// param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0"); +// JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaServerItem, param); +// System.out.println(2222); +// System.out.println(jsonObject); +// }else { + try { + cmder.talkStreamCmd(mediaServerItem, ssrcInfo, device, channelId, callId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { + logger.info("[瀵硅] 娴佸凡鐢熸垚锛� 寮�濮嬫帹娴侊細 " + response.toJSONString()); + dynamicTask.stop(timeOutTaskKey); + // TODO 鏆備笉鍋氬鐞� + }, (MediaServerItem mediaServerItemInuse, JSONObject json) -> { + logger.info("[瀵硅] 璁惧寮�濮嬫帹娴侊細 " + json.toJSONString()); + dynamicTask.stop(timeOutTaskKey); + // 鑾峰彇杩滅▼IP绔彛 浣滀负鍥炲璇煶娴佺殑鍦板潃 + String ip = json.getString("ip"); + Integer port = json.getInteger("port"); + logger.info("[璁惧寮�濮嬫帹娴乚{}/{}, 鏉ヨ嚜ip锛歿}, 绔彛锛歿}", device.getDeviceId(), channelId, ip, port); + // 鏌ョ湅骞冲彴鎺ㄦ祦鏄惁灏辩华 +// Boolean ready = zlmrtpServerFactory.isStreamReady(mediaServerItemInuse, "talk", stream); +// if (!ready) { +// try { +// cmder.streamByeCmd(device, channelId, finalSsrcInfo.getStream(), null); +// } catch (InvalidArgumentException | ParseException | SipException e) { +// logger.error("[瀵硅瓒呮椂]锛� 鍙戦�丅YE澶辫触 {}", e.getMessage()); +// } catch (SsrcTransactionNotFoundException e) { +// timeoutCallback.run(); +// mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); +// mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream()); +// streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); +// } +// }else { +// try { +// Thread.sleep(1000); +// } catch (InterruptedException e) { +// throw new RuntimeException(e); +// } + SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, ip, port, ssrcInfo.getSsrc(), device.getDeviceId(), + device.getDeviceId(), channelId, + false); + + +// if (sendRtpItem.getLocalPort() == 0) { +// logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); +// try { +// cmder.streamByeCmd(device, channelId, finalSsrcInfo.getStream(), null); +// } catch (InvalidArgumentException | ParseException | SipException e) { +// logger.error("[瀵硅瓒呮椂]锛� 鍙戦�丅YE澶辫触 {}", e.getMessage()); +// } catch (SsrcTransactionNotFoundException e) { +// timeoutCallback.run(); +// mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); +// mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream()); +// streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); +// } +// return; +// } + sendRtpItem.setTcpActive(false); + sendRtpItem.setCallId(callId); + sendRtpItem.setPlayType(InviteStreamType.TALK); + sendRtpItem.setStatus(1); + sendRtpItem.setIp(ip); + sendRtpItem.setPort(port); + sendRtpItem.setTcpActive(false); + sendRtpItem.setApp("1000"); + sendRtpItem.setStreamId("1000"); + sendRtpItem.setSsrc(ssrc); + sendRtpItem.setOnlyAudio(true); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + + 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("dst_url", sendRtpItem.getIp()); + param.put("dst_port", sendRtpItem.getPort()); + param.put("src_port", sendRtpItem.getLocalPort()); + param.put("pt", sendRtpItem.getPt()); + param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0"); + param.put("is_udp", sendRtpItem.isTcp() ? "0" : "1"); + param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0"); + JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaServerItemInuse, param); + System.out.println(11111); + System.out.println(sendRtpItem.getIp() + ":" + sendRtpItem.getPort()); +// System.out.println(jsonObject); +// } + + }, (event) -> { + + }, (event) -> { + dynamicTask.stop(timeOutTaskKey); + mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream()); + // 閲婃斁ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); + + streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); + errorEvent.response(event); + }); + } catch (InvalidArgumentException | SipException | ParseException e) { + + logger.error("[鍛戒护鍙戦�佸け璐 瀵硅娑堟伅: {}", e.getMessage()); + dynamicTask.stop(timeOutTaskKey); + mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream()); + // 閲婃斁ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); + + streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); + SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); + eventResult.msg = "鍛戒护鍙戦�佸け璐�"; + errorEvent.response(eventResult); + } +// } + + } @Override public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, @@ -405,12 +606,11 @@ if (device == null) { return null; } - String mediaServerId = device.getMediaServerId(); MediaServerItem mediaServerItem; - if (mediaServerId == null || "".equals(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) { + if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) { mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(); } else { - mediaServerItem = mediaServerService.getOne(mediaServerId); + mediaServerItem = mediaServerService.getOne(device.getMediaServerId()); } if (mediaServerItem == null) { logger.warn("鐐规挱鏃舵湭鎵惧埌鍙娇鐢ㄧ殑ZLM..."); @@ -774,6 +974,103 @@ } @Override + public AudioBroadcastResult audioBroadcast(Device device, String channelId) { + if (device == null || channelId == null) { + return null; + } + logger.info("[璇煶鍠婅瘽] device锛� {}, channel: {}", device.getDeviceId(), channelId); + DeviceChannel deviceChannel = storager.queryChannel(device.getDeviceId(), channelId); + if (deviceChannel == null) { + logger.warn("寮�鍚闊冲箍鎾殑鏃跺�欐湭鎵惧埌閫氶亾锛� {}", channelId); + return null; + } + MediaServerItem mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(); + String app = "broadcast"; + // TODO 浠巗ip user agent涓垽鏂槸浠�涔堝搧鐗岃澶囷紝澶у崕榛樿浣跨敤talk妯″紡锛屽叾浠栦娇鐢╞roadcast妯″紡 +// String app = "talk"; + String stream = device.getDeviceId() + "_" + channelId; + StreamInfo broadcast = mediaService.getStreamInfoByAppAndStream(mediaServerItem, "broadcast", stream, null, null, null, false); + AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult(); + audioBroadcastResult.setApp(app); + audioBroadcastResult.setStream(stream); + audioBroadcastResult.setStreamInfo(mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, stream, null, null, null,false)); + audioBroadcastResult.setCodec("G.711"); + return audioBroadcastResult; + } + + @Override + public void audioBroadcastCmd(Device device, String channelId, int timeout, AudioBroadcastEvent event) throws InvalidArgumentException, ParseException, SipException { + if (device == null || channelId == null) { + return; + } + logger.info("[璇煶鍠婅瘽] device锛� {}, channel: {}", device.getDeviceId(), channelId); + DeviceChannel deviceChannel = storager.queryChannel(device.getDeviceId(), channelId); + if (deviceChannel == null) { + logger.warn("寮�鍚闊冲箍鎾殑鏃跺�欐湭鎵惧埌閫氶亾锛� {}", channelId); + event.call("寮�鍚闊冲箍鎾殑鏃跺�欐湭鎵惧埌閫氶亾"); + return; + } + // 鏌ヨ閫氶亾浣跨敤鐘舵�� + if (audioBroadcastManager.exit(device.getDeviceId(), channelId)) { + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null); + if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) { + // 鏌ヨ娴佹槸鍚﹀瓨鍦紝涓嶅瓨鍦ㄥ垯璁や负鏄紓甯哥姸鎬� + MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStreamId()); + if (streamReady) { + logger.warn("璇煶骞挎挱宸茬粡寮�鍚細 {}", channelId); + event.call("璇煶骞挎挱宸茬粡寮�鍚�"); + return; + } else { + audioBroadcastManager.del(deviceChannel.getDeviceId(), channelId); + redisCatchStorage.deleteSendRTPServer(device.getDeviceId(), channelId, sendRtpItem.getCallId(), sendRtpItem.getStreamId()); + } + } + } + + // 鍙戦�侀�氱煡 + cmder.audioBroadcastCmd(device, channelId, eventResultForOk -> { + // 鍙戦�佹垚鍔� + AudioBroadcastCatch audioBroadcastCatch = new AudioBroadcastCatch(device.getDeviceId(), channelId, AudioBroadcastCatchStatus.Ready); + audioBroadcastManager.add(audioBroadcastCatch); + }, eventResultForError -> { + // 鍙戦�佸け璐� + logger.error("璇煶骞挎挱鍙戦�佸け璐ワ細 {}:{}", channelId, eventResultForError.msg); + event.call("璇煶骞挎挱鍙戦�佸け璐�"); + stopAudioBroadcast(device.getDeviceId(), channelId); + }); + } + + + + @Override + public void stopAudioBroadcast(String deviceId, String channelId) { + AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(deviceId, channelId); + if (audioBroadcastCatch != null) { + + Device device = deviceService.queryDevice(deviceId); + if (device == null) { + return; + } +// if (audioBroadcastCatch.getStatus() == AudioBroadcastCatchStatus.Ok) { +// cmder.streamByeCmd(device, audioBroadcastCatch.getChannelId(), null, audioBroadcastCatch.getSipTransactionInfo().getCallId()); +// } + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null); + if (sendRtpItem != null) { + redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null); + MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + Map<String, Object> param = new HashMap<>(); + param.put("vhost", "__defaultVhost__"); + param.put("app", sendRtpItem.getApp()); + param.put("stream", sendRtpItem.getStreamId()); + zlmresTfulUtils.stopSendRtp(mediaInfo, param); + } + + audioBroadcastManager.del(deviceId, channelId); + } + } + + @Override public void zlmServerOnline(String mediaServerId) { // TODO 鏌ユ壘涔嬪墠鐨勭偣鎾紝娴佸鏋滀笉瀛樺湪鍒欑粰涓嬬骇鍙戦�乥ye // MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); -- Gitblit v1.8.0