package com.genersoft.iot.vmp.service.impl; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.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.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.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.session.SsrcConfig; 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.AssistRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IDeviceService; 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.service.bean.*; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.redis.RedisUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.ErrorHook; import com.genersoft.iot.vmp.vmanager.bean.StreamContent; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPResponse; 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.util.ObjectUtils; import org.springframework.web.context.request.async.DeferredResult; import javax.sip.InvalidArgumentException; import javax.sip.ResponseEvent; import javax.sip.SipException; import java.math.BigDecimal; import java.math.RoundingMode; import java.text.ParseException; import java.util.EventObject; import java.util.List; import java.util.UUID; @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 void play(MediaServerItem mediaServerItem, String deviceId, String channelId, int isUsePs, ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, Runnable timeoutCallback) { String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId; WVPResult wvpResult = new WVPResult(); RequestMessage msg = new RequestMessage(); Device device = storager.queryVideoDevice(deviceId); StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); msg.setKey(key); if (mediaServerItem == null) { logger.info("无法找到zlm服务"); // 释放异步请求 wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("点播失败， 无法找到zlm服务"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); redisCatchStorage.stopPlay(streamInfo); return; } if (streamInfo != null) { logger.info("可以找到对应的stream"); String streamId = streamInfo.getStream(); if (streamId == null) { wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("点播失败， redis缓存streamId等于null"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); redisCatchStorage.stopPlay(streamInfo); resultHolder.invokeAllResult(msg); return; } String mediaServerId = streamInfo.getMediaServerId(); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId); // 判断rtpserver是否为null if (rtpInfo == null) { logger.info("rtpInfo为null"); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("点播失败，请重试 rtpInfo为null"); msg.setData(wvpResult); redisCatchStorage.stopPlay(streamInfo); resultHolder.invokeAllResult(msg); return; } if (rtpInfo.getInteger("code") == 0) { if (rtpInfo.getBoolean("exist")) { int localPort = rtpInfo.getInteger("local_port"); if (localPort == 0) { logger.warn("[点播]，点播时发现rtpServer存在，但是尚未开始推流"); // 此时说明rtpServer已经创建但是流还没有推上来 wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("点播已经在进行中，请稍候重试"); msg.setData(wvpResult); redisCatchStorage.stopPlay(streamInfo); resultHolder.invokeAllResult(msg); return; } else { 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, JSON.parseObject(JSON.toJSONString(streamInfo))); } } } else { redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); streamInfo = null; } } else { logger.warn("获取rtpinfo失败"); //zlm连接失败 redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); streamInfo = null; } } if (streamInfo == null) { String streamId = null; if (mediaServerItem.isRtpEnable()) { logger.info("test ----- {}",device.getDeviceId()); logger.info("test ----- {}",channelId); streamId = String.format("%s_%s", device.getDeviceId(), channelId); } SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); logger.info(JSONObject.toJSONString(ssrcInfo)); if (ssrcInfo == null) { wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("开启收流失败"); msg.setData(wvpResult); resultHolder.invokeAllResult(msg); return; } play(mediaServerItem, ssrcInfo, device, channelId, isUsePs, (mediaServerItemInUse, response) -> { if (hookEvent != null) { hookEvent.response(mediaServerItem, response); } }, event -> { // sip error错误 logger.warn("sip 错误,点播失败"); 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点播超时 logger.warn("sip 点播超时"); wvpResult.setCode(ErrorCode.ERROR100.getCode()); if (code == 0) { wvpResult.setMsg("点播超时，请稍候重试"); } else if (code == 1) { wvpResult.setMsg("收流超时，请稍候重试"); } msg.setData(wvpResult); // 回复之前所有的点播请求 resultHolder.invokeAllResult(msg); }); } } @Override public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, int isUsePs, ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, InviteTimeOutCallback timeoutCallback) { logger.info("[点播开始] deviceId: {}, channelId: {},收流端口： {}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); // 超时处理 String timeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(timeOutTaskKey, () -> { // 执行超时任务时查询是否已经成功，成功了则不执行超时任务，防止超时任务取消失败的情况 if (redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId) == null) { logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {}，端口：{}, SSRC: {}", device.getDeviceId(), channelId, ssrcInfo.getPort(), ssrcInfo.getSsrc()); // 点播超时回复BYE 同时释放ssrc以及此次点播的资源 try { cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { logger.error("[点播超时]， 发送BYE失败 {}", e.getMessage()); } finally { timeoutCallback.run(1, "收流超时"); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); logger.info("点播超时, 关闭rtp通道"); mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); // 取消订阅消息监听 HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId()); subscribe.removeSubscribe(hookSubscribe); } } }, userSetting.getPlayTimeout()); RequestMessage msg = new RequestMessage(); msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + device.getDeviceId() + channelId); //端口获取失败的ssrcInfo 没有必要发送点播指令 if (ssrcInfo.getPort() <= 0) { logger.info("[点播端口分配异常]，deviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo); dynamicTask.stop(timeOutTaskKey); // 释放ssrc mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "点播端口分配异常")); resultHolder.invokeAllResult(msg); return; } try { cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, isUsePs, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { logger.info("收到订阅消息： " + response.toJSONString()); logger.info("停止超时任务： " + timeOutTaskKey); dynamicTask.stop(timeOutTaskKey); // hook响应 onPublishHandlerForPlay(msg, mediaServerItemInuse, response, device.getDeviceId(), channelId, ssrcInfo.getSsrc()); hookEvent.response(mediaServerItemInuse, response); logger.info("[点播成功] deviceId: {}, channelId: {}", device.getDeviceId(), channelId); String streamUrl = String.format("http://127.0.0.1:%s/%s/%s.live.flv", mediaServerItemInuse.getHttpPort(), "rtp", ssrcInfo.getStream()); String path = "snap"; String fileName = device.getDeviceId() + "_" + channelId + ".jpg"; // 请求截图 logger.info("[请求截图]: " + fileName); zlmresTfulUtils.getSnap(mediaServerItemInuse, streamUrl, 15, 1, path, fileName); }, (okEvent) -> { ResponseEvent responseEvent = (ResponseEvent) okEvent.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()); okEvent.msg = "下级自定义了ssrc,但是此ssrc不可用"; okEvent.statusCode = 400; errorEvent.response(okEvent); 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(timeOutTaskKey); // hook响应 onPublishHandlerForPlay(msg, mediaServerItemInUse, response, device.getDeviceId(), channelId, ssrcInfo.getSsrc()); hookEvent.response(mediaServerItemInUse, response); }); } // 关闭rtp server mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream()); // 重新开启ssrc server mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false, ssrcInfo.getPort()); } } } , (errEvent) -> { dynamicTask.stop(timeOutTaskKey); logger.info("[未知原因] 关闭推流 closeRTPServer"); mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream()); // 释放ssrc mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); errorEvent.response(errEvent); }); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 点播消息: {}", e.getMessage()); dynamicTask.stop(timeOutTaskKey); mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream()); // 释放ssrc mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); // 有什么大病一般的代码 // SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); // eventResult.msg = "命令发送失败"; EventObject eventObject = new EventObject("命令发送失败"); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(eventObject); errorEvent.response(eventResult); // // RequestMessage msg = new RequestMessage(); // msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + device.getDeviceId() + channelId); // msg.setData(WVPResult.fail(ErrorCode.ERR_Invite_fail, "点播命令发送失败: " + e.getMessage())); // resultHolder.invokeAllResult(msg); } } /** * 从缓存中获取流信息 * * @param deviceId * @param channelId * @return */ private StreamInfo getStreamInfo(String deviceId, String channelId, WVPResult wvpResult) { StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); // 两种情况, 一种是已经在推流中了. 一种是还没有开始推流 if (streamInfo == null) { // 未开始推流, 开启新推流 return null; } // 能够读取到视频缓存 String streamId = streamInfo.getStream(); // 应该是不可能出现的异常情况, if (streamId == null) { logger.warn("[读取推流信息] 失败， redis缓存异常 streamId 为空"); // 说明缓存的流信息有问题 // 移除redis 缓存 redisCatchStorage.stopPlay(streamInfo); redisCatchStorage.deleteDeviceStream(deviceId, channelId); storager.stopPlay(deviceId, channelId); return null; } String mediaServerId = streamInfo.getMediaServerId(); // 获取视频推流通道对应的流媒体信息 MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, streamId); if (rtpInfo == null) { logger.warn("[读取推流信息] 失败， rtpInfo为null"); redisCatchStorage.stopPlay(streamInfo); redisCatchStorage.deleteDeviceStream(deviceId, channelId); storager.stopPlay(deviceId, channelId); return null; } if (rtpInfo.getInteger("code") != 0) { redisCatchStorage.stopPlay(streamInfo); redisCatchStorage.deleteDeviceStream(deviceId, channelId); storager.stopPlay(deviceId, channelId); return null; } if (!rtpInfo.getBoolean("exist")) { // 说明流已经停止 redisCatchStorage.stopPlay(streamInfo); redisCatchStorage.deleteDeviceStream(deviceId, channelId); storager.stopPlay(deviceId, channelId); return null; } int localPort = rtpInfo.getInteger("local_port"); if (localPort == 0) { logger.warn("[读取推流信息] 失败， rtpServer存在，但是尚未开始推流"); // 此时说明rtpServer已经创建但是流还没有推上来, 所以稍微等待一下重试即可 // 使用 wvpResult 返回对应信息提示 wvpResult.setCode(ErrorCode.ERROR_Retry.getCode()); wvpResult.setMsg("点播已经在进行中，请稍后进行重试"); } return streamInfo; } // 直接抛弃原有的播放 函数 创建新地播放函数, 重新梳理播放逻辑 public void startPlay(RequestMessage requestMsg, MediaServerItem mediaServerItem, Device device, String channelId, int isUsePs) { // 获取缓存的流信息 WVPResult wvpResult = new WVPResult(); StreamInfo streamInfo = getStreamInfo(device.getDeviceId(), channelId, wvpResult); if (streamInfo != null) { logger.info("已经启用视频流 {}", wvpResult.toString()); // 说明已经在推流中了 if (wvpResult.getCode() != ErrorCode.ERROR_Retry.getCode()) { // 在获取流信息没有更改 wvpResult 时将返回值设置为成功 logger.warn("{} 已经在推流, 并且无异常", device.getDeviceId(), channelId); // 数据类型转换, 懒得改前端了 wvpResult = WVPResult.success( new StreamContent(streamInfo) ); } resultHolder.invokeResult(requestMsg.setData(wvpResult)); return; } // 说明还没有开始推流 // if (!mediaServerItem.isRtpEnable()) { // logger.warn("设备应用的流媒体未启用"); // resultHolder.invokeAllResult( // requestMsg.setData(WVPResult.fail(ErrorCode.ERROR100, "设备应用的流媒体未启用")) // ); // return; // } // 生成streamId String streamId = String.format("%s_%s", device.getDeviceId(), channelId); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); if (ssrcInfo == null) { logger.warn("开启收流失败"); resultHolder.invokeAllResult( requestMsg.setData(WVPResult.fail(ErrorCode.ERROR100, "开启收流失败")) ); return; } // 尝试启用收流 if (ssrcInfo.getPort() <= 0) { logger.info("[点播端口分配异常]，deviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo); resultHolder.invokeAllResult( requestMsg.setData(WVPResult.fail(ErrorCode.ERROR100, "点播端口分配异常, 请重试")) ); return; } executePlayLive(requestMsg, mediaServerItem, ssrcInfo, device, channelId, isUsePs); } /** * 执行推流相关操作 发送命令 * * @param requestMsg * @param mediaServerItem * @param device * @param channelId * @param isUsePs */ private void executePlayLive(RequestMessage requestMsg, MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, int isUsePs) { logger.info("[支持点播推流] 收流端口： {}, 收流模式：{}, SSRC: {}, SSRC校验：{}", ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); String stream = ssrcInfo.getStream(); logger.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort()); // 构建errorHook 回调函数 ErrorHook errorHook = wvpResult -> { logger.info("收到点播命令回调 {}", wvpResult); if (wvpResult.getCode() != ErrorCode.SUCCESS.getCode()) { logger.error("[发送点播命令] 失败 {}", wvpResult.getMsg()); requestMsg.setData(wvpResult); stopPlay(device, channelId); // resultHolder.invokeAllResult(requestMsg); } // 监听zlm的流改变事件, 在流改变时直接回复 http 信息 addHookSubscribeForStreamChange(requestMsg, mediaServerItem, device, channelId, ssrcInfo); }; try { // 发送命令 cmder.sendPlayCmd(mediaServerItem, ssrcInfo, device, channelId, isUsePs, errorHook); } catch (Exception e) { e.printStackTrace(); logger.error("[发送点播命令] 失败 {}", e.getMessage()); // 未知的错误, 打印错误堆栈 requestMsg.setData(WVPResult.fail(ErrorCode.ERROR100, "发送点播命令失败")); resultHolder.invokeAllResult(requestMsg); } } // 停止推流 public void stopPlay(Device device, String channelId) { WVPResult wvpResult = new WVPResult(); StreamInfo streamInfo = getStreamInfo(device.getDeviceId(), channelId, wvpResult); // 获取 流媒体 // 发送bye给设备 ErrorHook errorHook = wvpResult1 -> { logger.info("关闭推流回调 {}", wvpResult1); if (wvpResult1.getCode() != ErrorCode.SUCCESS.getCode()) { logger.error("[关闭推流回调] 失败 {}", wvpResult1.getMsg()); return; } // if (streamInfo != null) { logger.info("找到正在推流的视频流 尝试停止 {}", wvpResult.toString()); MediaServerItem mediaServerItem = mediaServerService.getOne( streamInfo.getMediaServerId() ); // 获取ssrc logger.info("移除并关闭推流通道"); redisCatchStorage.stopPlay(streamInfo); redisCatchStorage.deleteDeviceStream(device.getDeviceId(), channelId); mediaServerService.releaseSsrc(streamInfo.getMediaServerId(), streamInfo.getSsrc()); mediaServerService.closeRTPServer(streamInfo.getMediaServerId(), streamInfo.getStream()); } else { logger.info("在没有服务端缓存的情况下关闭推流通道"); } }; try { // 发送命令 cmder.stopPlayCmd(device, channelId, errorHook); } catch (Exception e) { e.printStackTrace(); logger.error("[关闭推流] 失败 {}", e.getMessage()); // 未知的错误, 打印错误堆栈 } } /** * 截取视频快照 * * @param mediaServerItemInUse * @param deviceId * @param channelId * @param ssrcInfo */ public void screenshot(MediaServerItem mediaServerItemInUse, String deviceId, String channelId, SSRCInfo ssrcInfo) { String streamUrl = String.format("http://127.0.0.1:%s/%s/%s.live.flv", mediaServerItemInUse.getHttpPort(), "rtp", ssrcInfo.getStream()); String path = "snap"; String fileName = deviceId + "_" + channelId + ".jpg"; // 请求截图 logger.info("[尝试截取视频快照]: " + fileName); zlmresTfulUtils.getSnap(mediaServerItemInUse, streamUrl, 15, 1, path, fileName); } public void openBroadcast(MediaServerItem mediaServerItem, Device device, int waitTime, BroadcastCallback callback) { logger.warn("[语音广播] 开语音广播 新"); JSONObject errJson = new JSONObject(); try { cmder.audioBroadcastCmd(device); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 发送broadcast中 errorMsg: {}", e.getMessage()); errJson.put("msg", "[命令发送失败] 无法发送broadcast消息"); callback.run(2, errJson, null); } logger.warn("等待设备返回invite"); HookSubscribeForKey broadcastForInviteHook = GBHookSubscribeFactory.on_broadcast_invite(device.getDeviceId()); // 创建计时器,计时结束未收到invite则自动进行失败处理 String timeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(timeOutTaskKey, () -> { // todo 发送 bye 通知给设备? logger.warn("invite超时"); errJson.put("msg", "等待设备语音invite信息超时"); callback.run(1, errJson, null); }, waitTime); GBHookSubscribe.addInviteSubscribe( broadcastForInviteHook, (int code, JSONObject json, SIPRequest request) -> { // invite信息返回 logger.info("[语音广播] 接收到设备invite信息___订阅事件触发 JSONDATA: {}", json.toJSONString()); // 取消计时器 dynamicTask.stop(timeOutTaskKey); callback.run(0, json, request); }); } ; private void onPublishHandlerForPlay( RequestMessage requestMsg, MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String ssrc) { StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId); if (streamInfo != null) { streamInfo.setBack(false); streamInfo.setSsrc(ssrc); DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); if (deviceChannel != null) { deviceChannel.setStreamId(streamInfo.getStream()); storager.startPlay(deviceId, channelId, streamInfo.getStream()); } redisCatchStorage.startPlay(streamInfo); // 流信息转换 requestMsg.setData(WVPResult.success(new StreamContent(streamInfo))); resultHolder.invokeAllResult(requestMsg); } else { logger.warn("设备预览API调用失败！"); requestMsg.setData(WVPResult.fail(ErrorCode.ERROR100, "设备预览API调用失败！")); resultHolder.invokeAllResult(requestMsg); } } private void onPublishHandlerForPlayback(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String ssrc, PlayBackCallback playBackCallback) { StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId); PlayBackResult playBackResult = new PlayBackResult<>(); if (streamInfo != null) { streamInfo.setBack(true); streamInfo.setSsrc(ssrc); DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); if (deviceChannel != null) { deviceChannel.setStreamId(streamInfo.getStream()); storager.startPlay(deviceId, channelId, streamInfo.getStream()); } redisCatchStorage.startPlay(streamInfo); playBackResult.setCode(ErrorCode.SUCCESS.getCode()); playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); playBackResult.setData(streamInfo); playBackCallback.call(playBackResult); } else { logger.warn("录像回放调用失败！"); playBackResult.setCode(ErrorCode.ERROR100.getCode()); playBackResult.setMsg("录像回放调用失败！"); playBackCallback.call(playBackResult); } } @Override public MediaServerItem getNewMediaServerItem(Device device) { if (device == null) { return null; } MediaServerItem mediaServerItem; if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) { mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(null); } else { // 尝试获取device配置的zlm服务 mediaServerItem = mediaServerService.getOne(device.getMediaServerId()); // 如果默认zlm无法找到则随机分配一个zlm if(mediaServerItem == null){ logger.warn("无法找到设备默认流媒体服务,即将使用默认流媒体服务"); mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(null); } } if (mediaServerItem == null) { logger.warn("点播时未找到可使用的ZLM..."); } return mediaServerItem; } @Override public MediaServerItem getNewMediaServerItemHasAssist(Device device) { if (device == null) { return null; } MediaServerItem mediaServerItem; if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) { mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(true); } else { mediaServerItem = mediaServerService.getOne(device.getMediaServerId()); } if (mediaServerItem == null) { logger.warn("[获取可用的ZLM节点]未找到可使用的ZLM..."); } return mediaServerItem; } @Override public void playBack(String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback inviteStreamCallback, PlayBackCallback callback) { Device device = storager.queryVideoDevice(deviceId); if (device == null) { return; } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, device.isSsrcCheck(), true); playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback); } @Override public void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback) { if (mediaServerItem == null || ssrcInfo == null) { return; } Device device = storager.queryVideoDevice(deviceId); if (device == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "设备： " + deviceId + "不存在"); } 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("回放超时"); try { cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException e) { logger.error("[录像流]回放超时 发送BYE失败 {}", e.getMessage()); } catch (SsrcTransactionNotFoundException e) { // 点播超时回复BYE 同时释放ssrc以及此次点播的资源 logger.info("回放超时") mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream()); streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); } // 回复之前所有的点播请求 playBackCallback.call(playBackResult); }, userSetting.getPlayTimeout()); SipSubscribe.Event errorEvent = event -> { dynamicTask.stop(playBackTimeOutTaskKey); playBackResult.setCode(ErrorCode.ERROR100.getCode()); playBackResult.setMsg(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg)); 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()); playBackResult.setCode(ErrorCode.SUCCESS.getCode()); playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); playBackResult.setData(streamInfo); 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响应 onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, ssrcInfo.getSsrc(), playBackCallback); hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream())); }); } // 关闭rtp server mediaServerService.closeRTPServer(mediaServerItem.getId(), 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); } } @Override public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback) { Device device = storager.queryVideoDevice(deviceId); if (device == null) { return; } MediaServerItem newMediaServerItem = getNewMediaServerItemHasAssist(device); if (newMediaServerItem == null) { PlayBackResult downloadResult = new PlayBackResult<>(); downloadResult.setCode(ErrorCode.ERROR100.getCode()); downloadResult.setMsg("未找到assist服务"); playBackCallback.call(downloadResult); return; } SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, device.isSsrcCheck(), true); download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, infoCallBack, playBackCallback); } @Override public void 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; } Device device = storager.queryVideoDevice(deviceId); if (device == null) { throw new ControllerException(ErrorCode.ERROR400.getCode(), "设备：" + deviceId + "不存在"); } PlayBackResult downloadResult = new PlayBackResult<>(); String downLoadTimeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(downLoadTimeOutTaskKey, () -> { logger.warn(String.format("录像下载请求超时，deviceId：%s ，channelId：%s", deviceId, channelId)); 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) { logger.info("下载失败 SsrcTransactionNotFoundException"); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream()); streamSession.remove(deviceId, channelId, ssrcInfo.getStream()); } }, userSetting.getPlayTimeout()); SipSubscribe.Event errorEvent = event -> { dynamicTask.stop(downLoadTimeOutTaskKey); downloadResult.setCode(ErrorCode.ERROR100.getCode()); downloadResult.setMsg(String.format("录像下载失败， 错误码： %s, %s", event.statusCode, event.msg)); downloadResult.setEvent(event); hookCallBack.call(downloadResult); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); }; InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> { logger.info("收到订阅消息： " + inviteStreamInfo.getCallId()); dynamicTask.stop(downLoadTimeOutTaskKey); StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId); streamInfo.setStartTime(startTime); streamInfo.setEndTime(endTime); redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId()); downloadResult.setCode(ErrorCode.SUCCESS.getCode()); downloadResult.setMsg(ErrorCode.SUCCESS.getMsg()); downloadResult.setData(streamInfo); downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem()); downloadResult.setResponse(inviteStreamInfo.getResponse()); hookCallBack.call(downloadResult); }; try { cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack, hookEvent, errorEvent, 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(downLoadTimeOutTaskKey); // hook响应 onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, ssrcInfo.getSsrc(), hookCallBack); hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream())); }); } logger.info("下载完成"); // 关闭rtp server mediaServerService.closeRTPServer(mediaServerItem.getId(), ssrcInfo.getStream()); // 重新开启ssrc server mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort()); } } } }); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 录像下载: {}", e.getMessage()); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); eventResult.msg = "命令发送失败"; errorEvent.response(eventResult); } } @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) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "连接Assist服务失败"); } if (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 json, String deviceId, String channelId) { String streamId = json.getString("stream"); JSONArray tracks = json.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.getDevice(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); } public void broadcast( MediaServerItem mediaServerItem, Device device, BroadcastItem broadcastItem, int waitTime, BroadcastCallback callback ){ // 初始化waitTime值为15秒 if (waitTime == 0) { waitTime = 15 * 1000; } logger.info("[语言广播] 开始返回invite200,并创建zlm服务 app:{} stream:{} ", broadcastItem.getApp(), broadcastItem.getStream()); JSONObject errJson = new JSONObject(); try { String steamId = "relay_" + broadcastItem.getStream(); String ssrcStr = broadcastItem.getSsrc(); // 更新 broadcastItem.setRecv_stream(steamId); ssrcStr = ssrcStr.replaceAll("\\s*|\r|\n|\t", ""); SSRCInfo ssrcInfo = mediaServerService.startSendRtpServer(mediaServerItem, broadcastItem.getApp(), broadcastItem.getRecv_stream(), broadcastItem.getStream(), broadcastItem.getIpcIp(), String.valueOf(broadcastItem.getIpcAudioPort()), ssrcStr, waitTime ); broadcastItem.setSsrc(ssrcInfo.getSsrc()); if (ssrcInfo == null) { logger.error("[zlm控制异常] 创建媒体流失败"); errJson.put("msg", "[zlm控制异常] 创建媒体流失败"); callback.run(2, errJson, null); return; } // 回复invite 200 信息至设备 cmder.sendBoradcastInviteCmd(broadcastItem.getRequest(), mediaServerItem, ssrcInfo, device, null, null, null, null, null); callback.run(0, null, null); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[下发audio拉流invite失败]", e); logger.error("[zlm控制异常] 创建媒体流失败"); errJson.put("msg", "[zlm控制异常] 创建媒体流失败"); callback.run(2, errJson, null); // SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null)); // eventResult.msg = "命令发送失败"; // errorEvent.response(eventResult); } } ; private void addHookSubscribeForStreamChange( RequestMessage requestMsg, MediaServerItem mediaServerItem, Device device, String channelId, SSRCInfo ssrcInfo) { String stream = ssrcInfo.getStream(); HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId()); subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> { // zlm 事件触发. 流改变事件 logger.info("[ZLM HOOK] 收到ZLM流 编号信息： {}", json.toJSONString()); logger.info("[点播成功] deviceId: {}, channelId: {}", device.getDeviceId(), channelId); // 处理结果 onPublishHandlerForPlay(requestMsg, mediaServerItemInUse, json, device.getDeviceId(), channelId, ssrcInfo.getSsrc()); // 请求截图 screenshot(mediaServerItemInUse, device.getDeviceId(), channelId, ssrcInfo); }); } }