package com.genersoft.iot.vmp.service.impl; import java.math.BigDecimal; import java.math.RoundingMode; import java.text.ParseException; import java.util.*; import javax.sip.InvalidArgumentException; import javax.sip.ResponseEvent; import javax.sip.SipException; 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.GBEventSubscribe; import com.genersoft.iot.vmp.gb28181.GBHookSubscribeFactory; import com.genersoft.iot.vmp.gb28181.HookSubscribeForKey; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.bean.*; import com.genersoft.iot.vmp.utils.redis.RedisUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; import org.springframework.web.context.request.async.DeferredResult; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONArray; 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.UserSetting; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; 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.dto.HookSubscribeFactory; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaService; import com.genersoft.iot.vmp.service.IPlayService; 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.PlayResult; import gov.nist.javax.sip.stack.SIPDialog; @SuppressWarnings(value = {"rawtypes", "unchecked"}) @Service public class PlayServiceImpl implements IPlayService { private final static Logger logger = LoggerFactory.getLogger(PlayServiceImpl.class); @Autowired private IVideoManagerStorage storager; @Autowired private SIPCommander cmder; @Autowired private SIPCommanderFroPlatform sipCommanderFroPlatform; @Autowired private IRedisCatchStorage redisCatchStorage; @Autowired private DeferredResultHolder resultHolder; @Autowired private ZLMRESTfulUtils zlmresTfulUtils; @Autowired private AssistRESTfulUtils assistRESTfulUtils; @Autowired private IMediaService mediaService; @Autowired private IMediaServerService mediaServerService; @Autowired private VideoStreamSessionManager streamSession; @Autowired private IDeviceService deviceService; @Autowired private UserSetting userSetting; @Autowired private DynamicTask dynamicTask; @Autowired private ZlmHttpHookSubscribe subscribe; @Autowired private GBEventSubscribe GBHookSubscribe; @Qualifier("taskExecutor") @Autowired private ThreadPoolTaskExecutor taskExecutor; @Override public PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId,int isUsePs, ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, Runnable timeoutCallback) { if (mediaServerItem == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的zlm"); } PlayResult playResult = new PlayResult(); RequestMessage msg = new RequestMessage(); String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId; msg.setKey(key); String uuid = UUID.randomUUID().toString(); msg.setId(uuid); playResult.setUuid(uuid); DeferredResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); playResult.setResult(result); // 录像查询以channelId作为deviceId查询 resultHolder.put(key, uuid, result); Device device = redisCatchStorage.getDevice(deviceId); StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); playResult.setDevice(device); result.onCompletion(()->{ // 点播结束时调用截图接口 taskExecutor.execute(()->{ // TODO 应该在上流时调用更好,结束也可能是错误结束 String path = "snap"; String fileName = deviceId + "_" + channelId + ".jpg"; WVPResult wvpResult = (WVPResult)result.getResult(); if (Objects.requireNonNull(wvpResult).getCode() == 0) { StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData(); MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId()); String streamUrl = streamInfoForSuccess.getFmp4(); // 请求截图 logger.info("[请求截图]: " + fileName); zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName); } }); }); if (streamInfo != null) { String streamId = streamInfo.getStream(); if (streamId == null) { WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("点播失败, redis缓存streamId等于null"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); return playResult; } String mediaServerId = streamInfo.getMediaServerId(); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId); if(rtpInfo.getInteger("code") == 0){ if (rtpInfo.getBoolean("exist")) { int localPort = rtpInfo.getInteger("local_port"); if (localPort == 0) { logger.warn("[点播],点播时发现rtpServerC存在,但是尚未开始推流"); // 此时说明rtpServer已经创建但是流还没有推上来 WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("点播已经在进行中,请稍候重试"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); return playResult; }else { WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.SUCCESS.getCode()); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); wvpResult.setData(streamInfo); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); if (hookEvent != null) { hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo))); } } }else { redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); streamInfo = null; } }else { //zlm连接失败 redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); streamInfo = null; } } if (streamInfo == null) { 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(JSONObject.toJSONString(ssrcInfo)); play(mediaServerItem, ssrcInfo, device, channelId,isUsePs, (mediaServerItemInUse, response)->{ if (hookEvent != null) { hookEvent.response(mediaServerItem, response); } }, event -> { // sip error错误 WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg(String.format("点播失败, 错误码: %s, %s", event.statusCode, event.msg)); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); if (errorEvent != null) { errorEvent.response(event); } }, (code, msgStr)->{ // invite点播超时 WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); if (code == 0) { wvpResult.setMsg("点播超时,请稍候重试"); }else if (code == 1) { wvpResult.setMsg("收流超时,请稍候重试"); } msg.setData(wvpResult); // 回复之前所有的点播请求 resultHolder.invokeAllResult(msg); }, uuid); } return playResult; } @Override public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, int isUsePs, ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, InviteTimeOutCallback timeoutCallback, String uuid) { String streamId = null; if (mediaServerItem.isRtpEnable()) { streamId = String.format("%s_%s", device.getDeviceId(), channelId); } 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() ); // 超时处理 String timeOutTaskKey = UUID.randomUUID().toString(); SSRCInfo finalSsrcInfo = ssrcInfo; System.out.println("设置超时任务: " + timeOutTaskKey); dynamicTask.startDelay( timeOutTaskKey,()->{ logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {},端口:{}, SSRC: {}", device.getDeviceId(), channelId, finalSsrcInfo.getPort(), finalSsrcInfo.getSsrc()); timeoutCallback.run(1, "收流超时"); // 点播超时回复BYE 同时释放ssrc以及此次点播的资源 try { cmder.streamByeCmd(device, channelId, finalSsrcInfo.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException e) { logger.error("[点播超时], 发送BYE失败 {}", e.getMessage()); } catch (SsrcTransactionNotFoundException e) { timeoutCallback.run(0, "点播超时"); 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(); //端口获取失败的ssrcInfo 没有必要发送点播指令 if(ssrcInfo.getPort() <= 0){ logger.info("[点播端口分配异常],deviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo); return; } try { cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, isUsePs, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { logger.info("收到订阅消息: " + response.toJSONString()); System.out.println("停止超时任务: " + timeOutTaskKey); dynamicTask.stop(timeOutTaskKey); // hook响应 onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId, uuid); hookEvent.response(mediaServerItemInuse, response); logger.info("[点播成功] deviceId: {}, channelId: {}", device.getDeviceId(), channelId); }, (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字节,不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容 String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); // 查询到ssrc不一致且开启了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,但是此ssrc不可用"; event.statusCode = 400; errorEvent.response(event); return; } // 单端口模式streamId也有变化,需要重新设置监听 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()); // 重新开启ssrc 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); } } public PlayResult openBroadcast(MediaServerItem mediaServerItem,String deviceId,String audioStreamId,ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, Runnable timeoutCallback){ PlayResult playResult = new PlayResult(); RequestMessage msg = new RequestMessage(); String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + deviceId; // 使用invite 来进行延迟 msg.setKey(key); String uuid = UUID.randomUUID().toString(); msg.setId(uuid); playResult.setUuid(uuid); DeferredResult> result = new DeferredResult<>(13*1000l); playResult.setResult(result); resultHolder.put(key, uuid, result); // 检查zlm if (mediaServerItem == null) { logger.warn("[语音广播] 无法连接至ZLM服务器"); WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("无法连接至流媒体服务器"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); return playResult; } // 此段为录像查询... Device device = redisCatchStorage.getDevice(deviceId); StreamInfo streamInfo = redisCatchStorage.queryRecordByDevice(deviceId); playResult.setDevice(device); result.onCompletion(()->{ // 通道结束 logger.info("[语音广播] broadcast流程结束...."); // 下发invite信息给设备 }); HookSubscribeForKey broadcastForInviteHook = GBHookSubscribeFactory.on_broadcast_invite(deviceId); result.onTimeout(()->{ logger.warn("[广播超时] 与设备交互broadcast流程超时,未收到设备invite信息"); WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("[广播超时] 与设备交互broadcast流程超时,未收到设备invite信息"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); // 超时移除 GBHookSubscribe.removeSubscribe(broadcastForInviteHook); }); // 检查无法连接zlm的情况 if (streamInfo != null && false) { GBHookSubscribe.removeSubscribe(broadcastForInviteHook); String streamId = streamInfo.getStream(); if (streamId == null){ logger.warn("[语音广播] zlm 缓存的 streamId等于null"); WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("zlm 缓存的 streamId等于null"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); return playResult; } String mediaServerId = streamInfo.getMediaServerId(); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId); if(rtpInfo.getInteger("code") == 0){ if (rtpInfo.getBoolean("exist")) { int localPort = rtpInfo.getInteger("local_port"); if (localPort == 0) { logger.warn("[语音广播],点播时发现rtpServerC存在,但是尚未开始推流"); // 此时说明rtpServer已经创建但是流还没有推上来 WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("语音广播已经开始"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); return playResult; }else{ // 语音点播已经开始,返回推流信息给前端 WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.SUCCESS.getCode()); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); wvpResult.setData(streamInfo); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); if (hookEvent != null) { hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo))); } } }else{ logger.warn("[语音广播] 停止语音广播推流转发端口"); redisCatchStorage.stopBroadcast(streamInfo); storager.stopBroadcast(streamInfo.getDeviceID()); streamInfo = null; } }else{ logger.warn("[语音广播] 无法连接至zlm服务器"); redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); streamInfo = null; } } if(streamInfo == null){ // rtp语音通道创建完成,开始发送broadcast broadcastEventHandle(device, (int code,String tipMsg)->{ if(code == 1){ WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg(tipMsg); msg.setData(wvpResult); // 回复之前所有的点播请求 resultHolder.invokeAllResult(msg); } } ); // 开始创建 rtp/tcp 推流通道 // 注册subScript事件 GBHookSubscribe.addSubscribe(broadcastForInviteHook,(int code,JSONObject json)->{ logger.info("[语音广播] 接收到设备invite信息___订阅事件触发 JSONDATA: {}",json.toJSONString()); String streamId = null; SSRCInfo ssrcInfo; if (mediaServerItem.isRtpEnable()) { streamId = String.format("broadcast_%s", device.getDeviceId()); } logger.info("[语音广播] 尝试创建rtp语音推流通道"); ssrcInfo = mediaServerService.startSendRtpServer(mediaServerItem, streamId, audioStreamId, json.getString("addr"), json.getString("port"), json.getString("ssrc") ); logger.info(JSONObject.toJSONString(ssrcInfo)); resultHolder.invokeAllResult(msg); WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.SUCCESS.getCode()); wvpResult.setMsg("okokokokooo"); wvpResult.setData(ssrcInfo); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); // TODO: 2023/3/7 开始下发invite信息给设备 try { cmder.sendBoradcastInviteCmd(mediaServerItem, ssrcInfo, device, null, null, null, (_event) -> { }, (_errorEvent) -> { }); }catch(InvalidArgumentException | SipException | ParseException e){ logger.error("[下发audio拉流invite失败]",e); // SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); // eventResult.msg = "命令发送失败"; // errorEvent.response(eventResult); } }); } return playResult; } public void broadcastEventHandle( Device device, NodeCallBack nodeCallBack){ logger.info("[语音广播] 开始broadcast交互"); try { cmder.audioBroadcastCmd(device); nodeCallBack.run(0,"ok"); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 发送broadcast中 errorMsg: {}", e.getMessage()); nodeCallBack.run(1,"[命令发送失败] 无法发送broadcast消息"); } } @Override public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) { RequestMessage msg = new RequestMessage(); if (uuid != null) { msg.setId(uuid); } msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId); StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId); if (streamInfo != null) { DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); if (deviceChannel != null) { deviceChannel.setStreamId(streamInfo.getStream()); storager.startPlay(deviceId, channelId, streamInfo.getStream()); } redisCatchStorage.startPlay(streamInfo); WVPResult wvpResult = new WVPResult(); wvpResult.setCode(ErrorCode.SUCCESS.getCode()); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); wvpResult.setData(streamInfo); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); } else { logger.warn("设备预览API调用失败!"); msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "设备预览API调用失败!")); resultHolder.invokeAllResult(msg); } } @Override public MediaServerItem getNewMediaServerItem(Device device) { if (device == null) { return null; } String mediaServerId = device.getMediaServerId(); MediaServerItem mediaServerItem; if (mediaServerId == null) { mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(); }else { mediaServerItem = mediaServerService.getOne(mediaServerId); } if (mediaServerItem == null) { logger.warn("点播时未找到可使用的ZLM..."); } return mediaServerItem; } @Override public DeferredResult> playBack(String deviceId, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, PlayBackCallback callback) { Device device = storager.queryVideoDevice(deviceId); if (device == null) { return null; } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true, true); return playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback); } @Override public DeferredResult> playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback) { if (mediaServerItem == null || ssrcInfo == null) { return null; } String uuid = UUID.randomUUID().toString(); String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId; Device device = storager.queryVideoDevice(deviceId); if (device == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "设备: " + deviceId + "不存在"); } DeferredResult> result = new DeferredResult<>(30000L); resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid, result); RequestMessage requestMessage = new RequestMessage(); requestMessage.setId(uuid); requestMessage.setKey(key); PlayBackResult playBackResult = new PlayBackResult<>(); String playBackTimeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(playBackTimeOutTaskKey, ()->{ logger.warn(String.format("设备回放超时,deviceId:%s ,channelId:%s", deviceId, channelId)); playBackResult.setCode(ErrorCode.ERROR100.getCode()); playBackResult.setMsg("回放超时"); playBackResult.setData(requestMessage); try { cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException e) { logger.error("[录像流]回放超时 发送BYE失败 {}", e.getMessage()); } catch (SsrcTransactionNotFoundException e) { // 点播超时回复BYE 同时释放ssrc以及此次点播的资源 mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); } // 回复之前所有的点播请求 playBackCallback.call(playBackResult); result.setResult(WVPResult.fail(ErrorCode.ERROR100.getCode(), "回放超时")); resultHolder.exist(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid); }, userSetting.getPlayTimeout()); SipSubscribe.Event errorEvent = event -> { dynamicTask.stop(playBackTimeOutTaskKey); requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("回放失败, 错误码: %s, %s", event.statusCode, event.msg))); playBackResult.setCode(ErrorCode.ERROR100.getCode()); playBackResult.setMsg(String.format("回放失败, 错误码: %s, %s", event.statusCode, event.msg)); playBackResult.setData(requestMessage); playBackResult.setEvent(event); playBackCallback.call(playBackResult); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); }; InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> { logger.info("收到回放订阅消息: " + inviteStreamInfo.getResponse().toJSONString()); dynamicTask.stop(playBackTimeOutTaskKey); StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); if (streamInfo == null) { logger.warn("设备回放API调用失败!"); playBackResult.setCode(ErrorCode.ERROR100.getCode()); playBackResult.setMsg("设备回放API调用失败!"); playBackCallback.call(playBackResult); return; } redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId()); WVPResult success = WVPResult.success(streamInfo); requestMessage.setData(success); playBackResult.setCode(ErrorCode.SUCCESS.getCode()); playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); playBackResult.setData(requestMessage); playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem()); playBackResult.setResponse(inviteStreamInfo.getResponse()); playBackCallback.call(playBackResult); }; try { cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack, hookEvent, eventResult -> { if (eventResult.type == SipSubscribe.EventResultType.response) { ResponseEvent responseEvent = (ResponseEvent)eventResult.event; String contentString = new String(responseEvent.getResponse().getRawContent()); // 获取ssrc int ssrcIndex = contentString.indexOf("y="); // 检查是否有y字段 if (ssrcIndex >= 0) { //ssrc规定长度为10字节,不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容 String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); // 查询到ssrc不一致且开启了ssrc校验则需要针对处理 if (ssrcInfo.getSsrc().equals(ssrcInResponse)) { return; } logger.info("[回放消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse ); if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) { logger.info("[回放消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse); if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) { // ssrc 不可用 // 释放ssrc mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); eventResult.msg = "下级自定义了ssrc,但是此ssrc不可用"; eventResult.statusCode = 400; errorEvent.response(eventResult); return; } // 单端口模式streamId也有变化,需要重新设置监听 if (!mediaServerItem.isRtpEnable()) { // 添加订阅 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)->{ logger.info("[ZLM HOOK] ssrc修正后收到订阅消息: " + response.toJSONString()); dynamicTask.stop(playBackTimeOutTaskKey); // hook响应 onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid); hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream())); }); } // 关闭rtp server mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); // 重新开启ssrc server mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort()); } } } }, errorEvent); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 回放: {}", e.getMessage()); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); eventResult.msg = "命令发送失败"; errorEvent.response(eventResult); } return result; } @Override public DeferredResult> download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) { Device device = storager.queryVideoDevice(deviceId); if (device == null) { return null; } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true, true); return download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed,infoCallBack, hookCallBack); } @Override public DeferredResult> download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) { if (mediaServerItem == null || ssrcInfo == null) { return null; } String uuid = UUID.randomUUID().toString(); String key = DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId; DeferredResult> result = new DeferredResult<>(30000L); Device device = storager.queryVideoDevice(deviceId); if (device == null) { throw new ControllerException(ErrorCode.ERROR400.getCode(), "设备:" + deviceId + "不存在"); } resultHolder.put(key, uuid, result); RequestMessage requestMessage = new RequestMessage(); requestMessage.setId(uuid); requestMessage.setKey(key); WVPResult wvpResult = new WVPResult<>(); requestMessage.setData(wvpResult); PlayBackResult downloadResult = new PlayBackResult<>(); downloadResult.setData(requestMessage); String downLoadTimeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(downLoadTimeOutTaskKey, ()->{ logger.warn(String.format("录像下载请求超时,deviceId:%s ,channelId:%s", deviceId, channelId)); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("录像下载请求超时"); downloadResult.setCode(ErrorCode.ERROR100.getCode()); downloadResult.setMsg("录像下载请求超时"); hookCallBack.call(downloadResult); // 点播超时回复BYE 同时释放ssrc以及此次点播的资源 try { cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException e) { logger.error("[录像流]录像下载请求超时, 发送BYE失败 {}", e.getMessage()); } catch (SsrcTransactionNotFoundException e) { mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); } // 回复之前所有的点播请求 hookCallBack.call(downloadResult); }, userSetting.getPlayTimeout()); SipSubscribe.Event errorEvent = event -> { dynamicTask.stop(downLoadTimeOutTaskKey); downloadResult.setCode(ErrorCode.ERROR100.getCode()); downloadResult.setMsg(String.format("录像下载失败, 错误码: %s, %s", event.statusCode, event.msg)); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg(String.format("录像下载失败, 错误码: %s, %s", event.statusCode, event.msg)); downloadResult.setEvent(event); hookCallBack.call(downloadResult); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); }; try { cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack, inviteStreamInfo -> { logger.info("收到订阅消息: " + inviteStreamInfo.getResponse().toJSONString()); dynamicTask.stop(downLoadTimeOutTaskKey); StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); streamInfo.setStartTime(startTime); streamInfo.setEndTime(endTime); redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId()); wvpResult.setCode(ErrorCode.SUCCESS.getCode()); wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); wvpResult.setData(streamInfo); downloadResult.setCode(ErrorCode.SUCCESS.getCode()); downloadResult.setMsg(ErrorCode.SUCCESS.getMsg()); downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem()); downloadResult.setResponse(inviteStreamInfo.getResponse()); hookCallBack.call(downloadResult); }, errorEvent); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 录像下载: {}", e.getMessage()); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); eventResult.msg = "命令发送失败"; errorEvent.response(eventResult); } return result; } @Override public StreamInfo getDownLoadInfo(String deviceId, String channelId, String stream) { StreamInfo streamInfo = redisCatchStorage.queryDownload(deviceId, channelId, stream, null); if (streamInfo != null) { if (streamInfo.getProgress() == 1) { return streamInfo; } // 获取当前已下载时长 String mediaServerId = streamInfo.getMediaServerId(); MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); if (mediaServerItem == null) { logger.warn("查询录像信息时发现节点已离线"); return null; } if (mediaServerItem.getRecordAssistPort() > 0) { JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, streamInfo.getApp(), streamInfo.getStream(), null); if (jsonObject != null && jsonObject.getInteger("code") == 0) { long duration = jsonObject.getLong("data"); if (duration == 0) { streamInfo.setProgress(0); }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); double process = divide.doubleValue(); streamInfo.setProgress(process); } } } } return streamInfo; } @Override public void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String uuid) { RequestMessage msg = new RequestMessage(); msg.setKey(DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId); msg.setId(uuid); StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); if (streamInfo != null) { redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId()); msg.setData(JSON.toJSONString(streamInfo)); resultHolder.invokeResult(msg); } else { logger.warn("设备预览API调用失败!"); msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "设备预览API调用失败!")); resultHolder.invokeResult(msg); } } 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.setDeviceID(deviceId); streamInfo.setChannelId(channelId); return streamInfo; } @Override public void zlmServerOffline(String mediaServerId) { // 处理正在向上推流的上级平台 List sendRtpItems = redisCatchStorage.querySendRTPServer(null); if (sendRtpItems.size() > 0) { for (SendRtpItem sendRtpItem : sendRtpItems) { if (sendRtpItem.getMediaServerId().equals(mediaServerId)) { ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); try { sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId()); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage()); } } } } // 处理正在观看的国标设备 List allSsrc = streamSession.getAllSsrc(); if (allSsrc.size() > 0) { for (SsrcTransaction ssrcTransaction : allSsrc) { if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) { Device device = deviceService.queryDevice(ssrcTransaction.getDeviceId()); if (device == null) { continue; } try { cmder.streamByeCmd(device, ssrcTransaction.getChannelId(), ssrcTransaction.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { logger.error("[zlm离线]为正在使用此zlm的设备, 发送BYE失败 {}", e.getMessage()); } } } } } @Override public void zlmServerOnline(String mediaServerId) { // TODO 查找之前的点播,流如果不存在则给下级发送bye // MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); // zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{ // Integer code = mediaList.getInteger("code"); // if (code == 0) { // JSONArray data = mediaList.getJSONArray("data"); // if (data == null || data.size() == 0) { // zlmServerOffline(mediaServerId); // }else { // Map mediaListMap = new HashMap<>(); // for (int i = 0; i < data.size(); i++) { // JSONObject json = data.getJSONObject(i); // String app = json.getString("app"); // if ("rtp".equals(app)) { // String stream = json.getString("stream"); // if (mediaListMap.get(stream) != null) { // continue; // } // mediaListMap.put(stream, json); // // 处理正在观看的国标设备 // List ssrcTransactions = streamSession.getSsrcTransactionForAll(null, null, null, stream); // if (ssrcTransactions.size() > 0) { // for (SsrcTransaction ssrcTransaction : ssrcTransactions) { // if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) { // cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), // ssrcTransaction.getStream(), null); // } // } // } // } // } // if (mediaListMap.size() > 0 ) { // // 处理正在向上推流的上级平台 // List sendRtpItems = redisCatchStorage.querySendRTPServer(null); // if (sendRtpItems.size() > 0) { // for (SendRtpItem sendRtpItem : sendRtpItems) { // if (sendRtpItem.getMediaServerId().equals(mediaServerId)) { // if (mediaListMap.get(sendRtpItem.getStreamId()) == null) { // ParentPlatform platform = storager.queryPlatformByServerGBId(sendRtpItem.getPlatformId()); // sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId()); // } // } // } // } // } // } // } // })); } @Override public void pauseRtp(String streamId) throws ServiceException, InvalidArgumentException, ParseException, SipException { String key = redisCatchStorage.queryPlaybackForKey(null, null, streamId, null); StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null); if (null == streamInfo) { logger.warn("streamId不存在!"); throw new ServiceException("streamId不存在"); } streamInfo.setPause(true); RedisUtil.set(key, streamInfo); MediaServerItem mediaServerItem = mediaServerService.getOne(streamInfo.getMediaServerId()); if (null == mediaServerItem) { logger.warn("mediaServer 不存在!"); throw new ServiceException("mediaServer不存在"); } // zlm 暂停RTP超时检查 JSONObject jsonObject = zlmresTfulUtils.pauseRtpCheck(mediaServerItem, streamId); if (jsonObject == null || jsonObject.getInteger("code") != 0) { throw new ServiceException("暂停RTP接收失败"); } Device device = storager.queryVideoDevice(streamInfo.getDeviceID()); cmder.playPauseCmd(device, streamInfo); } @Override public void resumeRtp(String streamId) throws ServiceException, InvalidArgumentException, ParseException, SipException { String key = redisCatchStorage.queryPlaybackForKey(null, null, streamId, null); StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null); if (null == streamInfo) { logger.warn("streamId不存在!"); throw new ServiceException("streamId不存在"); } streamInfo.setPause(false); RedisUtil.set(key, streamInfo); MediaServerItem mediaServerItem = mediaServerService.getOne(streamInfo.getMediaServerId()); if (null == mediaServerItem) { logger.warn("mediaServer 不存在!"); throw new ServiceException("mediaServer不存在"); } // zlm 暂停RTP超时检查 JSONObject jsonObject = zlmresTfulUtils.resumeRtpCheck(mediaServerItem, streamId); if (jsonObject == null || jsonObject.getInteger("code") != 0) { throw new ServiceException("继续RTP接收失败"); } Device device = storager.queryVideoDevice(streamInfo.getDeviceID()); cmder.playResumeCmd(device, streamInfo); } }