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/service/impl/PlayServiceImpl.java | 386 ++++++++++++++++++++++++++++++++++++++++--------------- 1 files changed, 281 insertions(+), 105 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 d6c3d5f..68055a4 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 @@ -12,8 +12,10 @@ import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.conf.exception.ControllerException; 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.service.IDeviceService; +import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,7 +38,6 @@ 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; @@ -54,32 +55,9 @@ import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.utils.redis.RedisUtil; -import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult; -import com.genersoft.iot.vmp.utils.DateUtil; 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; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.http.HttpStatus; -import org.springframework.http.ResponseEntity; -import org.springframework.stereotype.Service; -import org.springframework.util.ResourceUtils; -import org.springframework.web.context.request.async.DeferredResult; - -import javax.sip.ResponseEvent; -import javax.sip.SipException; -import java.io.FileNotFoundException; -import java.math.BigDecimal; -import java.text.ParseException; -import java.math.RoundingMode; -import java.util.*; -import java.util.stream.Collectors; -import java.util.stream.Stream; @SuppressWarnings(value = {"rawtypes", "unchecked"}) @Service @@ -97,7 +75,10 @@ private AudioBroadcastManager audioBroadcastManager; @Autowired - private SIPCommanderFroPlatform sipCommanderFroPlatform; + private IDeviceService deviceService; + + @Autowired + private ISIPCommanderForPlatform sipCommanderFroPlatform; @Autowired private IRedisCatchStorage redisCatchStorage; @@ -123,10 +104,6 @@ @Autowired private VideoStreamSessionManager streamSession; - - @Autowired - private IDeviceService deviceService; - @Autowired private UserSetting userSetting; @@ -143,7 +120,6 @@ @Qualifier("taskExecutor") @Autowired private ThreadPoolTaskExecutor taskExecutor; - @Override @@ -169,15 +145,15 @@ StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); playResult.setDevice(device); - result.onCompletion(()->{ + result.onCompletion(() -> { // 鐐规挱缁撴潫鏃惰皟鐢ㄦ埅鍥炬帴鍙� - taskExecutor.execute(()->{ + taskExecutor.execute(() -> { // TODO 搴旇鍦ㄤ笂娴佹椂璋冪敤鏇村ソ锛岀粨鏉熶篃鍙兘鏄敊璇粨鏉� - String path = "snap"; - String fileName = deviceId + "_" + channelId + ".jpg"; - WVPResult wvpResult = (WVPResult)result.getResult(); + String path = "snap"; + String fileName = deviceId + "_" + channelId + ".jpg"; + WVPResult wvpResult = (WVPResult) result.getResult(); if (Objects.requireNonNull(wvpResult).getCode() == 0) { - StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData(); + StreamInfo streamInfoForSuccess = (StreamInfo) wvpResult.getData(); MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId()); String streamUrl = streamInfoForSuccess.getFmp4(); // 璇锋眰鎴浘 @@ -201,7 +177,7 @@ MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId); - if(rtpInfo.getInteger("code") == 0){ + if (rtpInfo.getInteger("code") == 0) { if (rtpInfo.getBoolean("exist")) { int localPort = rtpInfo.getInteger("local_port"); if (localPort == 0) { @@ -214,7 +190,7 @@ resultHolder.invokeAllResult(msg); return playResult; - }else { + } else { WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.SUCCESS.getCode()); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); @@ -227,12 +203,12 @@ } } - }else { + } else { redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); streamInfo = null; } - }else { + } else { //zlm杩炴帴澶辫触 redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); @@ -246,7 +222,7 @@ } SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); logger.info(JSONObject.toJSONString(ssrcInfo)); - play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response)->{ + play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response) -> { if (hookEvent != null) { hookEvent.response(mediaServerItem, response); } @@ -260,13 +236,13 @@ if (errorEvent != null) { errorEvent.response(event); } - }, (code, msgStr)->{ + }, (code, msgStr) -> { // invite鐐规挱瓒呮椂 WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); if (code == 0) { wvpResult.setMsg("鐐规挱瓒呮椂锛岃绋嶅�欓噸璇�"); - }else if (code == 1) { + } else if (code == 1) { wvpResult.setMsg("鏀舵祦瓒呮椂锛岃绋嶅�欓噸璇�"); } msg.setData(wvpResult); @@ -277,6 +253,186 @@ 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; + } + try { + String callId = SipUtils.getNewCallId(); + 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 { + SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, ip, port, ssrcInfo.getSsrc(), device.getDeviceId(), + device.getDeviceId(), channelId, + false); + + sendRtpItem.setTcpActive(false); + if (sendRtpItem == null || 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.setCallId(callId); + sendRtpItem.setPlayType(InviteStreamType.TALK); + sendRtpItem.setStatus(1); + sendRtpItem.setIp(ip); + sendRtpItem.setPort(port); + sendRtpItem.setTcpActive(false); + sendRtpItem.setStreamId(ssrcInfo.getStream()); + sendRtpItem.setApp("talk"); + sendRtpItem.setSsrc(ssrc); + 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("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 = zlmrtpServerFactory.startSendRtpStream(mediaServerItemInuse, param); + System.out.println(11111); + System.out.println(jsonObject); + } + + }, (event) -> { +// ResponseEvent responseEvent = (ResponseEvent) event.event; +// String contentString = new String(responseEvent.getResponse().getRawContent()); +// // 鑾峰彇ssrc +// int ssrcIndex = contentString.indexOf("y="); +// // 妫�鏌ユ槸鍚︽湁y瀛楁 +// if (ssrcIndex >= 0) { +// //ssrc瑙勫畾闀垮害涓�10瀛楄妭锛屼笉鍙栦綑涓嬮暱搴︿互閬垮厤鍚庣画杩樻湁鈥渇=鈥濆瓧娈� TODO 鍚庣画瀵逛笉瑙勮寖鐨勯潪10浣峴src鍏煎 +// String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); +// // 鏌ヨ鍒皊src涓嶄竴鑷翠笖寮�鍚簡ssrc鏍¢獙鍒欓渶瑕侀拡瀵瑰鐞� +// if (ssrc.equals(ssrcInResponse)) { +// return; +// } +// logger.info("[瀵硅娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse); +// if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { +// logger.info("[瀵硅娑堟伅] SSRC淇 {}->{}", ssrc, ssrcInResponse); +// +// if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { +// // ssrc 涓嶅彲鐢� +// // 閲婃斁ssrc +// mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc()); +// streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream()); +// event.msg = "涓嬬骇鑷畾涔変簡ssrc,浣嗘槸姝src涓嶅彲鐢�"; +// event.statusCode = 400; +// errorEvent.response(event); +// return; +// } +// +// // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚� +// if (!mediaServerItem.isRtpEnable()) { +// // 娣诲姞璁㈤槄 +// HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId()); +// subscribe.removeSubscribe(hookSubscribe); +// hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase()); +// subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> { +// logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); +// dynamicTask.stop(timeOutTaskKey); +// // hook鍝嶅簲 +// onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid); +// hookEvent.response(mediaServerItemInUse, response); +// }); +// } +// // 鍏抽棴rtp server +// mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream()); +// // 閲嶆柊寮�鍚痵src server +// mediaServerService.openRTPServer(mediaServerItem, finalSsrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false, finalSsrcInfo.getPort()); +// +// } +// } + }, (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 @@ -291,12 +447,12 @@ if (ssrcInfo == null) { 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() ); + 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,()->{ + dynamicTask.startDelay(timeOutTaskKey, () -> { logger.info("[鐐规挱瓒呮椂] 鏀舵祦瓒呮椂 deviceId: {}, channelId: {}锛岀鍙o細{}, SSRC: {}", device.getDeviceId(), channelId, finalSsrcInfo.getPort(), finalSsrcInfo.getSsrc()); timeoutCallback.run(1, "鏀舵祦瓒呮椂"); @@ -315,7 +471,7 @@ final String ssrc = ssrcInfo.getSsrc(); final String stream = ssrcInfo.getStream(); //绔彛鑾峰彇澶辫触鐨剆srcInfo 娌℃湁蹇呰鍙戦�佺偣鎾寚浠� - if(ssrcInfo.getPort() <= 0){ + if (ssrcInfo.getPort() <= 0) { logger.info("[鐐规挱绔彛鍒嗛厤寮傚父]锛宒eviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo); return; } @@ -330,7 +486,7 @@ logger.info("[鐐规挱鎴愬姛] deviceId: {}, channelId: {}", device.getDeviceId(), channelId); }, (event) -> { - ResponseEvent responseEvent = (ResponseEvent)event.event; + ResponseEvent responseEvent = (ResponseEvent) event.event; String contentString = new String(responseEvent.getResponse().getRawContent()); // 鑾峰彇ssrc int ssrcIndex = contentString.indexOf("y="); @@ -342,7 +498,7 @@ if (ssrc.equals(ssrcInResponse)) { return; } - logger.info("[鐐规挱娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse ); + logger.info("[鐐规挱娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse); if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { logger.info("[鐐规挱娑堟伅] SSRC淇 {}->{}", ssrc, ssrcInResponse); @@ -363,13 +519,13 @@ HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId()); subscribe.removeSubscribe(hookSubscribe); hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase()); - subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{ - logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); - dynamicTask.stop(timeOutTaskKey); - // hook鍝嶅簲 - onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid); - hookEvent.response(mediaServerItemInUse, response); - }); + subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> { + logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); + dynamicTask.stop(timeOutTaskKey); + // hook鍝嶅簲 + onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid); + hookEvent.response(mediaServerItemInUse, response); + }); } // 鍏抽棴rtp server mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream()); @@ -441,7 +597,7 @@ MediaServerItem mediaServerItem; if (mediaServerId == null) { mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(); - }else { + } else { mediaServerItem = mediaServerService.getOne(mediaServerId); } if (mediaServerItem == null) { @@ -452,8 +608,8 @@ @Override public DeferredResult<WVPResult<StreamInfo>> playBack(String deviceId, String channelId, String startTime, - String endTime,InviteStreamCallback inviteStreamCallback, - PlayBackCallback callback) { + String endTime, InviteStreamCallback inviteStreamCallback, + PlayBackCallback callback) { Device device = storager.queryVideoDevice(deviceId); if (device == null) { return null; @@ -466,9 +622,9 @@ @Override public DeferredResult<WVPResult<StreamInfo>> playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, - String deviceId, String channelId, String startTime, - String endTime, InviteStreamCallback infoCallBack, - PlayBackCallback playBackCallback) { + String deviceId, String channelId, String startTime, + String endTime, InviteStreamCallback infoCallBack, + PlayBackCallback playBackCallback) { if (mediaServerItem == null || ssrcInfo == null) { return null; } @@ -485,7 +641,7 @@ requestMessage.setKey(key); PlayBackResult<RequestMessage> playBackResult = new PlayBackResult<>(); String playBackTimeOutTaskKey = UUID.randomUUID().toString(); - dynamicTask.startDelay(playBackTimeOutTaskKey, ()->{ + dynamicTask.startDelay(playBackTimeOutTaskKey, () -> { logger.warn(String.format("璁惧鍥炴斁瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); playBackResult.setCode(ErrorCode.ERROR100.getCode()); playBackResult.setMsg("鍥炴斁瓒呮椂"); @@ -545,7 +701,7 @@ cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack, hookEvent, eventResult -> { if (eventResult.type == SipSubscribe.EventResultType.response) { - ResponseEvent responseEvent = (ResponseEvent)eventResult.event; + ResponseEvent responseEvent = (ResponseEvent) eventResult.event; String contentString = new String(responseEvent.getResponse().getRawContent()); // 鑾峰彇ssrc int ssrcIndex = contentString.indexOf("y="); @@ -557,7 +713,7 @@ if (ssrcInfo.getSsrc().equals(ssrcInResponse)) { return; } - logger.info("[鍥炴斁娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse ); + logger.info("[鍥炴斁娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse); if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { logger.info("[鍥炴斁娑堟伅] SSRC淇 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); @@ -578,7 +734,7 @@ HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId()); subscribe.removeSubscribe(hookSubscribe); hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase()); - subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{ + subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> { logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString()); dynamicTask.stop(playBackTimeOutTaskKey); // hook鍝嶅簲 @@ -614,7 +770,7 @@ MediaServerItem newMediaServerItem = getNewMediaServerItem(device); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true, true); - return download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed,infoCallBack, hookCallBack); + return download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, infoCallBack, hookCallBack); } @Override @@ -640,7 +796,7 @@ downloadResult.setData(requestMessage); String downLoadTimeOutTaskKey = UUID.randomUUID().toString(); - dynamicTask.startDelay(downLoadTimeOutTaskKey, ()->{ + dynamicTask.startDelay(downLoadTimeOutTaskKey, () -> { logger.warn(String.format("褰曞儚涓嬭浇璇锋眰瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId)); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("褰曞儚涓嬭浇璇锋眰瓒呮椂"); @@ -723,15 +879,15 @@ if (duration == 0) { streamInfo.setProgress(0); - }else { + } else { String startTime = streamInfo.getStartTime(); String endTime = streamInfo.getEndTime(); long start = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime); long end = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime); - BigDecimal currentCount = new BigDecimal(duration/1000); - BigDecimal totalCount = new BigDecimal(end-start); - BigDecimal divide = currentCount.divide(totalCount,2, RoundingMode.HALF_UP); + BigDecimal currentCount = new BigDecimal(duration / 1000); + BigDecimal totalCount = new BigDecimal(end - start); + BigDecimal divide = currentCount.divide(totalCount, 2, RoundingMode.HALF_UP); double process = divide.doubleValue(); streamInfo.setProgress(process); } @@ -762,7 +918,7 @@ public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId) { String streamId = resonse.getString("stream"); JSONArray tracks = resonse.getJSONArray("tracks"); - StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks, null); + StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, "rtp", streamId, tracks, null); streamInfo.setDeviceID(deviceId); streamInfo.setChannelId(channelId); return streamInfo; @@ -788,7 +944,7 @@ List<SsrcTransaction> allSsrc = streamSession.getAllSsrc(); if (allSsrc.size() > 0) { for (SsrcTransaction ssrcTransaction : allSsrc) { - if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) { + if (ssrcTransaction.getMediaServerId().equals(mediaServerId)) { Device device = deviceService.queryDevice(ssrcTransaction.getDeviceId()); if (device == null) { continue; @@ -806,10 +962,36 @@ } @Override - public void audioBroadcast(Device device, String channelId, int timeout, AudioBroadcastEvent event) { + 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); @@ -818,7 +1000,7 @@ } // 鏌ヨ閫氶亾浣跨敤鐘舵�� if (audioBroadcastManager.exit(device.getDeviceId(), channelId)) { - SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null); + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null); if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) { // 鏌ヨ娴佹槸鍚﹀瓨鍦紝涓嶅瓨鍦ㄥ垯璁や负鏄紓甯哥姸鎬� MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); @@ -827,8 +1009,8 @@ logger.warn("璇煶骞挎挱宸茬粡寮�鍚細 {}", channelId); event.call("璇煶骞挎挱宸茬粡寮�鍚�"); return; - }else { - audioBroadcastManager.del(deviceChannel.getDeviceId(),channelId); + } else { + audioBroadcastManager.del(deviceChannel.getDeviceId(), channelId); redisCatchStorage.deleteSendRTPServer(device.getDeviceId(), channelId, sendRtpItem.getCallId(), sendRtpItem.getStreamId()); } } @@ -847,39 +1029,33 @@ }); } + + @Override - public void stopAudioBroadcast(String deviceId, String channelId){ + public void stopAudioBroadcast(String deviceId, String channelId) { AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(deviceId, channelId); if (audioBroadcastCatch != null) { - try { - 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); - // 绔嬪埢缁撴潫璁惧鐨勬帹娴侊紝绛夊緟鑷缁撴潫澶參 - zlmresTfulUtils.closeStreams(mediaInfo, sendRtpItem.getApp(), sendRtpItem.getStreamId()); - } - if (audioBroadcastCatch.getStatus() == AudioBroadcastCatchStatus.Ok) { - cmder.streamByeCmd(audioBroadcastCatch.getDialog(), audioBroadcastCatch.getChannelId(), audioBroadcastCatch.getRequest(), null); - } - audioBroadcastManager.del(deviceId, channelId); - - } catch (SipException e) { - throw new RuntimeException(e); - } catch (ParseException e) { - throw new RuntimeException(e); - } catch (InvalidArgumentException e) { - throw new RuntimeException(e); + 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 -- Gitblit v1.8.0