PlayServiceImpl.java 56 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006
  1. package com.genersoft.iot.vmp.service.impl;
  2. import com.alibaba.fastjson2.JSON;
  3. import com.alibaba.fastjson2.JSONArray;
  4. import com.alibaba.fastjson2.JSONObject;
  5. import com.genersoft.iot.vmp.common.StreamInfo;
  6. import com.genersoft.iot.vmp.conf.DynamicTask;
  7. import com.genersoft.iot.vmp.conf.UserSetting;
  8. import com.genersoft.iot.vmp.conf.exception.ControllerException;
  9. import com.genersoft.iot.vmp.conf.exception.ServiceException;
  10. import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
  11. import com.genersoft.iot.vmp.gb28181.bean.*;
  12. import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
  13. import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
  14. import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
  15. import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
  16. import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
  17. import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
  18. import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
  19. import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
  20. import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
  21. import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
  22. import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
  23. import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
  24. import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
  25. import com.genersoft.iot.vmp.service.IDeviceService;
  26. import com.genersoft.iot.vmp.service.IMediaServerService;
  27. import com.genersoft.iot.vmp.service.IMediaService;
  28. import com.genersoft.iot.vmp.service.IPlayService;
  29. import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback;
  30. import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
  31. import com.genersoft.iot.vmp.service.bean.PlayBackResult;
  32. import com.genersoft.iot.vmp.service.bean.SSRCInfo;
  33. import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
  34. import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
  35. import com.genersoft.iot.vmp.utils.DateUtil;
  36. import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
  37. import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
  38. import org.slf4j.Logger;
  39. import org.slf4j.LoggerFactory;
  40. import org.springframework.beans.factory.annotation.Autowired;
  41. import org.springframework.data.redis.core.RedisTemplate;
  42. import org.springframework.stereotype.Service;
  43. import org.springframework.util.ObjectUtils;
  44. import javax.sdp.*;
  45. import javax.sip.InvalidArgumentException;
  46. import javax.sip.ResponseEvent;
  47. import javax.sip.SipException;
  48. import java.math.BigDecimal;
  49. import java.math.RoundingMode;
  50. import java.text.ParseException;
  51. import java.util.List;
  52. import java.util.UUID;
  53. import java.util.Vector;
  54. @SuppressWarnings(value = {"rawtypes", "unchecked"})
  55. @Service
  56. public class PlayServiceImpl implements IPlayService {
  57. private final static Logger logger = LoggerFactory.getLogger(PlayServiceImpl.class);
  58. @Autowired
  59. private IVideoManagerStorage storager;
  60. @Autowired
  61. private SIPCommander cmder;
  62. @Autowired
  63. private SIPCommanderFroPlatform sipCommanderFroPlatform;
  64. @Autowired
  65. private IRedisCatchStorage redisCatchStorage;
  66. @Autowired
  67. private DeferredResultHolder resultHolder;
  68. @Autowired
  69. private ZLMRESTfulUtils zlmresTfulUtils;
  70. @Autowired
  71. private AssistRESTfulUtils assistRESTfulUtils;
  72. @Autowired
  73. private IMediaService mediaService;
  74. @Autowired
  75. private IMediaServerService mediaServerService;
  76. @Autowired
  77. private VideoStreamSessionManager streamSession;
  78. @Autowired
  79. private IDeviceService deviceService;
  80. @Autowired
  81. private UserSetting userSetting;
  82. @Autowired
  83. private DynamicTask dynamicTask;
  84. @Autowired
  85. private ZlmHttpHookSubscribe subscribe;
  86. @Autowired
  87. private SSRCFactory ssrcFactory;
  88. @Autowired
  89. private RedisTemplate<Object, Object> redisTemplate;
  90. @Override
  91. public void play(MediaServerItem mediaServerItem, String deviceId, String channelId,
  92. ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
  93. Runnable timeoutCallback) {
  94. if (mediaServerItem == null) {
  95. throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的zlm");
  96. }
  97. String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;
  98. RequestMessage msg = new RequestMessage();
  99. msg.setKey(key);
  100. Device device = redisCatchStorage.getDevice(deviceId);
  101. StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
  102. if (streamInfo != null) {
  103. String streamId = streamInfo.getStream();
  104. if (streamId == null) {
  105. WVPResult wvpResult = new WVPResult();
  106. wvpResult.setCode(ErrorCode.ERROR100.getCode());
  107. wvpResult.setMsg("点播失败, redis缓存streamId等于null");
  108. msg.setData(wvpResult);
  109. resultHolder.invokeAllResult(msg);
  110. return;
  111. }
  112. String mediaServerId = streamInfo.getMediaServerId();
  113. MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
  114. JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
  115. if (rtpInfo.getInteger("code") == 0) {
  116. if (rtpInfo.getBoolean("exist")) {
  117. int localPort = rtpInfo.getInteger("local_port");
  118. if (localPort == 0) {
  119. logger.warn("[点播],点播时发现rtpServer存在,但是尚未开始推流");
  120. // 此时说明rtpServer已经创建但是流还没有推上来
  121. WVPResult wvpResult = new WVPResult();
  122. wvpResult.setCode(ErrorCode.ERROR100.getCode());
  123. wvpResult.setMsg("点播已经在进行中,请稍候重试");
  124. msg.setData(wvpResult);
  125. resultHolder.invokeAllResult(msg);
  126. return;
  127. } else {
  128. WVPResult wvpResult = new WVPResult();
  129. wvpResult.setCode(ErrorCode.SUCCESS.getCode());
  130. wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
  131. wvpResult.setData(streamInfo);
  132. msg.setData(wvpResult);
  133. resultHolder.invokeAllResult(msg);
  134. if (hookEvent != null) {
  135. hookEvent.response(mediaServerItem, JSON.parseObject(JSON.toJSONString(streamInfo)));
  136. }
  137. }
  138. } else {
  139. redisCatchStorage.stopPlay(streamInfo);
  140. storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
  141. streamInfo = null;
  142. }
  143. } else {
  144. //zlm连接失败
  145. redisCatchStorage.stopPlay(streamInfo);
  146. storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
  147. streamInfo = null;
  148. }
  149. }
  150. if (streamInfo == null) {
  151. String streamId = null;
  152. if (mediaServerItem.isRtpEnable()) {
  153. streamId = String.format("%s_%s", device.getDeviceId(), channelId);
  154. }
  155. SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam());
  156. if (ssrcInfo == null) {
  157. WVPResult wvpResult = new WVPResult();
  158. wvpResult.setCode(ErrorCode.ERROR100.getCode());
  159. wvpResult.setMsg("开启收流失败");
  160. msg.setData(wvpResult);
  161. resultHolder.invokeAllResult(msg);
  162. return;
  163. }
  164. play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response) -> {
  165. if (hookEvent != null) {
  166. hookEvent.response(mediaServerItem, response);
  167. }
  168. }, event -> {
  169. // sip error错误
  170. WVPResult wvpResult = new WVPResult();
  171. wvpResult.setCode(ErrorCode.ERROR100.getCode());
  172. wvpResult.setMsg(String.format("点播失败, 错误码: %s, %s", event.statusCode, event.msg));
  173. msg.setData(wvpResult);
  174. resultHolder.invokeAllResult(msg);
  175. if (errorEvent != null) {
  176. errorEvent.response(event);
  177. }
  178. }, (code, msgStr) -> {
  179. // invite点播超时
  180. WVPResult wvpResult = new WVPResult();
  181. wvpResult.setCode(ErrorCode.ERROR100.getCode());
  182. if (code == 0) {
  183. wvpResult.setMsg("点播超时,请稍候重试");
  184. } else if (code == 1) {
  185. wvpResult.setMsg("收流超时,请稍候重试");
  186. }
  187. msg.setData(wvpResult);
  188. // 回复之前所有的点播请求
  189. resultHolder.invokeAllResult(msg);
  190. });
  191. }
  192. }
  193. @Override
  194. public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
  195. ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
  196. InviteTimeOutCallback timeoutCallback) {
  197. logger.info("\r\n" +
  198. "[点播开始] \r\n" +
  199. "deviceId : {}, \r\n" +
  200. "channelId : {},\r\n" +
  201. "收流端口 :{}, \r\n" +
  202. "收流模式 :{}, \r\n" +
  203. "SSRC : {}, \r\n" +
  204. "SSRC校验 :{} ", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
  205. // 超时处理
  206. String timeOutTaskKey = UUID.randomUUID().toString();
  207. dynamicTask.startDelay(timeOutTaskKey, () -> {
  208. // 执行超时任务时查询是否已经成功,成功了则不执行超时任务,防止超时任务取消失败的情况
  209. if (redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId) == null) {
  210. logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {},端口:{}, SSRC: {}", device.getDeviceId(), channelId, ssrcInfo.getPort(), ssrcInfo.getSsrc());
  211. // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
  212. try {
  213. cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null);
  214. } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
  215. logger.error("[点播超时], 发送BYE失败 {}", e.getMessage());
  216. } finally {
  217. timeoutCallback.run(1, "收流超时");
  218. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  219. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
  220. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  221. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
  222. // 取消订阅消息监听
  223. HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
  224. subscribe.removeSubscribe(hookSubscribe);
  225. }
  226. }
  227. }, userSetting.getPlayTimeout());
  228. //端口获取失败的ssrcInfo 没有必要发送点播指令
  229. if (ssrcInfo.getPort() <= 0) {
  230. logger.info("[点播端口分配异常],deviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, JSON.toJSONString(ssrcInfo));
  231. dynamicTask.stop(timeOutTaskKey);
  232. // 释放ssrc
  233. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  234. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  235. RequestMessage msg = new RequestMessage();
  236. msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + device.getDeviceId() + channelId);
  237. msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "点播端口分配异常"));
  238. resultHolder.invokeAllResult(msg);
  239. return;
  240. }
  241. try {
  242. cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
  243. logger.info("收到订阅消息: " + response.toJSONString());
  244. dynamicTask.stop(timeOutTaskKey);
  245. // hook响应
  246. onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId);
  247. hookEvent.response(mediaServerItemInuse, response);
  248. logger.info("[点播成功] deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
  249. String streamUrl;
  250. if (mediaServerItemInuse.getRtspPort() != 0) {
  251. streamUrl = String.format("rtsp://127.0.0.1:%s/%s/%s", mediaServerItemInuse.getRtspPort(), "rtp", ssrcInfo.getStream());
  252. }else {
  253. streamUrl = String.format("http://127.0.0.1:%s/%s/%s.live.mp4", mediaServerItemInuse.getHttpPort(), "rtp", ssrcInfo.getStream());
  254. }
  255. String path = "snap";
  256. String fileName = device.getDeviceId() + "_" + channelId + ".jpg";
  257. // 请求截图
  258. logger.info("[请求截图]: " + fileName);
  259. zlmresTfulUtils.getSnap(mediaServerItemInuse, streamUrl, 15, 1, path, fileName);
  260. }, (event) -> {
  261. ResponseEvent responseEvent = (ResponseEvent) event.event;
  262. String contentString = new String(responseEvent.getResponse().getRawContent());
  263. // 获取ssrc
  264. int ssrcIndex = contentString.indexOf("y=");
  265. // 检查是否有y字段
  266. if (ssrcIndex >= 0) {
  267. //ssrc规定长度为10字节,不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容
  268. String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12).trim();
  269. // 查询到ssrc不一致且开启了ssrc校验则需要针对处理
  270. if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
  271. if (device.getStreamMode().equalsIgnoreCase("TCP-ACTIVE")) {
  272. String substring = contentString.substring(0, contentString.indexOf("y="));
  273. try {
  274. SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
  275. int port = -1;
  276. Vector mediaDescriptions = sdp.getMediaDescriptions(true);
  277. for (Object description : mediaDescriptions) {
  278. MediaDescription mediaDescription = (MediaDescription) description;
  279. Media media = mediaDescription.getMedia();
  280. Vector mediaFormats = media.getMediaFormats(false);
  281. if (mediaFormats.contains("96")) {
  282. port = media.getMediaPort();
  283. break;
  284. }
  285. }
  286. logger.info("[点播-TCP主动连接对方] deviceId: {}, channelId: {}, 连接对方的地址:{}:{}, 收流模式:{}, SSRC: {}, SSRC校验:{}", device.getDeviceId(), channelId, sdp.getConnection().getAddress(), port, device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
  287. JSONObject jsonObject = zlmresTfulUtils.connectRtpServer(mediaServerItem, sdp.getConnection().getAddress(), port, ssrcInfo.getStream());
  288. logger.info("[点播-TCP主动连接对方] 结果: {}", jsonObject);
  289. } catch (SdpException e) {
  290. logger.error("[点播-TCP主动连接对方] deviceId: {}, channelId: {}, 解析200OK的SDP信息失败", device.getDeviceId(), channelId, e);
  291. }
  292. }
  293. return;
  294. }
  295. logger.info("[点播消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
  296. if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
  297. logger.info("[点播消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
  298. // 释放不被使用的ssrc
  299. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  300. // 单端口模式streamId也有变化,需要重新设置监听
  301. if (!mediaServerItem.isRtpEnable()) {
  302. // 添加订阅
  303. HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
  304. subscribe.removeSubscribe(hookSubscribe);
  305. hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
  306. subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
  307. logger.info("[ZLM HOOK] ssrc修正后收到订阅消息: " + response.toJSONString());
  308. dynamicTask.stop(timeOutTaskKey);
  309. // hook响应
  310. onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId);
  311. hookEvent.response(mediaServerItemInUse, response);
  312. });
  313. }
  314. // 关闭rtp server
  315. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream(), result->{
  316. if (result) {
  317. // 重新开启ssrc server
  318. mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false, ssrcInfo.getPort(), true, device.getStreamModeForParam());
  319. }else {
  320. try {
  321. logger.warn("[停止点播] {}/{}", device.getDeviceId(), channelId);
  322. cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null);
  323. } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
  324. logger.error("[命令发送失败] 停止点播, 发送BYE: {}", e.getMessage());
  325. throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage());
  326. }
  327. dynamicTask.stop(timeOutTaskKey);
  328. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  329. event.msg = "下级自定义了ssrc,重新设置收流信息失败";
  330. event.statusCode = 500;
  331. errorEvent.response(event);
  332. }
  333. });
  334. }
  335. }
  336. }, (event) -> {
  337. dynamicTask.stop(timeOutTaskKey);
  338. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
  339. // 释放ssrc
  340. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  341. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  342. errorEvent.response(event);
  343. });
  344. } catch (InvalidArgumentException | SipException | ParseException e) {
  345. logger.error("[命令发送失败] 点播消息: {}", e.getMessage());
  346. dynamicTask.stop(timeOutTaskKey);
  347. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
  348. // 释放ssrc
  349. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  350. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  351. SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult();
  352. eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent;
  353. eventResult.statusCode = -1;
  354. eventResult.msg = "命令发送失败";
  355. errorEvent.response(eventResult);
  356. }
  357. }
  358. @Override
  359. public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId) {
  360. StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
  361. RequestMessage msg = new RequestMessage();
  362. msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
  363. if (streamInfo != null) {
  364. DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
  365. if (deviceChannel != null) {
  366. deviceChannel.setStreamId(streamInfo.getStream());
  367. storager.startPlay(deviceId, channelId, streamInfo.getStream());
  368. }
  369. redisCatchStorage.startPlay(streamInfo);
  370. WVPResult wvpResult = new WVPResult();
  371. wvpResult.setCode(ErrorCode.SUCCESS.getCode());
  372. wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
  373. wvpResult.setData(streamInfo);
  374. msg.setData(wvpResult);
  375. resultHolder.invokeAllResult(msg);
  376. } else {
  377. logger.warn("设备预览API调用失败!");
  378. msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "设备预览API调用失败!"));
  379. resultHolder.invokeAllResult(msg);
  380. }
  381. }
  382. private void onPublishHandlerForPlayback(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, PlayBackCallback playBackCallback) {
  383. StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
  384. PlayBackResult<StreamInfo> playBackResult = new PlayBackResult<>();
  385. if (streamInfo != null) {
  386. DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
  387. if (deviceChannel != null) {
  388. deviceChannel.setStreamId(streamInfo.getStream());
  389. storager.startPlay(deviceId, channelId, streamInfo.getStream());
  390. }
  391. redisCatchStorage.startPlay(streamInfo);
  392. playBackResult.setCode(ErrorCode.SUCCESS.getCode());
  393. playBackResult.setMsg(ErrorCode.SUCCESS.getMsg());
  394. playBackResult.setData(streamInfo);
  395. playBackCallback.call(playBackResult);
  396. } else {
  397. logger.warn("录像回放调用失败!");
  398. playBackResult.setCode(ErrorCode.ERROR100.getCode());
  399. playBackResult.setMsg("录像回放调用失败!");
  400. playBackCallback.call(playBackResult);
  401. }
  402. }
  403. @Override
  404. public MediaServerItem getNewMediaServerItem(Device device) {
  405. if (device == null) {
  406. return null;
  407. }
  408. MediaServerItem mediaServerItem;
  409. if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) {
  410. mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(null);
  411. } else {
  412. mediaServerItem = mediaServerService.getOne(device.getMediaServerId());
  413. }
  414. if (mediaServerItem == null) {
  415. logger.warn("点播时未找到可使用的ZLM...");
  416. }
  417. return mediaServerItem;
  418. }
  419. @Override
  420. public MediaServerItem getNewMediaServerItemHasAssist(Device device) {
  421. if (device == null) {
  422. return null;
  423. }
  424. MediaServerItem mediaServerItem;
  425. if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) {
  426. mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(true);
  427. } else {
  428. mediaServerItem = mediaServerService.getOne(device.getMediaServerId());
  429. }
  430. if (mediaServerItem == null) {
  431. logger.warn("[获取可用的ZLM节点]未找到可使用的ZLM...");
  432. }
  433. return mediaServerItem;
  434. }
  435. @Override
  436. public void playBack(String deviceId, String channelId, String startTime,
  437. String endTime, InviteStreamCallback inviteStreamCallback,
  438. PlayBackCallback callback) {
  439. Device device = storager.queryVideoDevice(deviceId);
  440. if (device == null) {
  441. return;
  442. }
  443. MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
  444. SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
  445. playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback);
  446. }
  447. @Override
  448. public void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,
  449. String deviceId, String channelId, String startTime,
  450. String endTime, InviteStreamCallback infoCallBack,
  451. PlayBackCallback playBackCallback) {
  452. if (mediaServerItem == null || ssrcInfo == null) {
  453. return;
  454. }
  455. Device device = storager.queryVideoDevice(deviceId);
  456. if (device == null) {
  457. throw new ControllerException(ErrorCode.ERROR100.getCode(), "设备: " + deviceId + "不存在");
  458. }
  459. logger.info("[回放消息] deviceId: {}, channelId: {},收流端口: {}, 收流模式:{}, SSRC: {}, SSRC校验:{}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
  460. PlayBackResult<StreamInfo> playBackResult = new PlayBackResult<>();
  461. String playBackTimeOutTaskKey = UUID.randomUUID().toString();
  462. dynamicTask.startDelay(playBackTimeOutTaskKey, () -> {
  463. logger.warn(String.format("设备回放超时,deviceId:%s ,channelId:%s", deviceId, channelId));
  464. playBackResult.setCode(ErrorCode.ERROR100.getCode());
  465. playBackResult.setMsg("回放超时");
  466. try {
  467. cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null);
  468. } catch (InvalidArgumentException | ParseException | SipException e) {
  469. logger.error("[录像流]回放超时 发送BYE失败 {}", e.getMessage());
  470. } catch (SsrcTransactionNotFoundException e) {
  471. // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
  472. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  473. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
  474. streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
  475. }
  476. // 回复之前所有的点播请求
  477. playBackCallback.call(playBackResult);
  478. }, userSetting.getPlayTimeout());
  479. SipSubscribe.Event errorEvent = event -> {
  480. dynamicTask.stop(playBackTimeOutTaskKey);
  481. playBackResult.setCode(ErrorCode.ERROR100.getCode());
  482. playBackResult.setMsg(String.format("回放失败, 错误码: %s, %s", event.statusCode, event.msg));
  483. playBackResult.setEvent(event);
  484. playBackCallback.call(playBackResult);
  485. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  486. };
  487. InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> {
  488. logger.info("收到回放订阅消息: " + inviteStreamInfo.getResponse().toJSONString());
  489. dynamicTask.stop(playBackTimeOutTaskKey);
  490. StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
  491. if (streamInfo == null) {
  492. logger.warn("设备回放API调用失败!");
  493. playBackResult.setCode(ErrorCode.ERROR100.getCode());
  494. playBackResult.setMsg("设备回放API调用失败!");
  495. playBackCallback.call(playBackResult);
  496. return;
  497. }
  498. redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId());
  499. playBackResult.setCode(ErrorCode.SUCCESS.getCode());
  500. playBackResult.setMsg(ErrorCode.SUCCESS.getMsg());
  501. playBackResult.setData(streamInfo);
  502. playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
  503. playBackResult.setResponse(inviteStreamInfo.getResponse());
  504. playBackCallback.call(playBackResult);
  505. };
  506. try {
  507. cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack,
  508. hookEvent, eventResult -> {
  509. if (eventResult.type == SipSubscribe.EventResultType.response) {
  510. ResponseEvent responseEvent = (ResponseEvent) eventResult.event;
  511. String contentString = new String(responseEvent.getResponse().getRawContent());
  512. // 获取ssrc
  513. int ssrcIndex = contentString.indexOf("y=");
  514. // 检查是否有y字段
  515. if (ssrcIndex >= 0) {
  516. //ssrc规定长度为10字节,不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容
  517. String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
  518. // 查询到ssrc不一致且开启了ssrc校验则需要针对处理
  519. if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
  520. return;
  521. }
  522. logger.info("[回放消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
  523. if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
  524. logger.info("[回放消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
  525. // 释放不被使用的ssrc
  526. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  527. // 单端口模式streamId也有变化,需要重新设置监听
  528. if (!mediaServerItem.isRtpEnable()) {
  529. // 添加订阅
  530. HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
  531. subscribe.removeSubscribe(hookSubscribe);
  532. hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
  533. subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
  534. logger.info("[ZLM HOOK] ssrc修正后收到订阅消息: " + response.toJSONString());
  535. dynamicTask.stop(playBackTimeOutTaskKey);
  536. // hook响应
  537. onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, playBackCallback);
  538. hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
  539. });
  540. }
  541. // 关闭rtp server
  542. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream(), result->{
  543. if (result) {
  544. // 重新开启ssrc server
  545. mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort(), true, device.getStreamModeForParam());
  546. }else {
  547. try {
  548. logger.warn("[回放消息]停止 {}/{}", device.getDeviceId(), channelId);
  549. cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null);
  550. } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
  551. logger.error("[命令发送失败] 停止点播 停止, 发送BYE: {}", e.getMessage());
  552. throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage());
  553. }
  554. dynamicTask.stop(playBackTimeOutTaskKey);
  555. // 释放ssrc
  556. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  557. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  558. errorEvent.response(eventResult);
  559. eventResult.msg = "下级自定义了ssrc,重新设置收流信息失败";
  560. eventResult.statusCode = 500;
  561. errorEvent.response(eventResult);
  562. }
  563. });
  564. }
  565. }
  566. }
  567. }, errorEvent);
  568. } catch (InvalidArgumentException | SipException | ParseException e) {
  569. logger.error("[命令发送失败] 回放: {}", e.getMessage());
  570. SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult();
  571. eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent;
  572. eventResult.statusCode = -1;
  573. eventResult.msg = "命令发送失败";
  574. errorEvent.response(eventResult);
  575. }
  576. }
  577. @Override
  578. public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback) {
  579. Device device = storager.queryVideoDevice(deviceId);
  580. if (device == null) {
  581. return;
  582. }
  583. MediaServerItem newMediaServerItem = getNewMediaServerItemHasAssist(device);
  584. if (newMediaServerItem == null) {
  585. PlayBackResult<StreamInfo> downloadResult = new PlayBackResult<>();
  586. downloadResult.setCode(ErrorCode.ERROR100.getCode());
  587. downloadResult.setMsg("未找到assist服务");
  588. playBackCallback.call(downloadResult);
  589. return;
  590. }
  591. SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
  592. download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, infoCallBack, playBackCallback);
  593. }
  594. @Override
  595. public void download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) {
  596. if (mediaServerItem == null || ssrcInfo == null) {
  597. return;
  598. }
  599. Device device = storager.queryVideoDevice(deviceId);
  600. if (device == null) {
  601. throw new ControllerException(ErrorCode.ERROR400.getCode(), "设备:" + deviceId + "不存在");
  602. }
  603. PlayBackResult<StreamInfo> downloadResult = new PlayBackResult<>();
  604. logger.info("[录像下载] deviceId: {}, channelId: {},收流端口: {}, 收流模式:{}, SSRC: {}, SSRC校验:{}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
  605. String downLoadTimeOutTaskKey = UUID.randomUUID().toString();
  606. dynamicTask.startDelay(downLoadTimeOutTaskKey, () -> {
  607. logger.warn(String.format("录像下载请求超时,deviceId:%s ,channelId:%s", deviceId, channelId));
  608. downloadResult.setCode(ErrorCode.ERROR100.getCode());
  609. downloadResult.setMsg("录像下载请求超时");
  610. hookCallBack.call(downloadResult);
  611. // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
  612. try {
  613. cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null);
  614. } catch (InvalidArgumentException | ParseException | SipException e) {
  615. logger.error("[录像流]录像下载请求超时, 发送BYE失败 {}", e.getMessage());
  616. } catch (SsrcTransactionNotFoundException e) {
  617. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  618. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
  619. streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
  620. }
  621. }, userSetting.getPlayTimeout());
  622. SipSubscribe.Event errorEvent = event -> {
  623. dynamicTask.stop(downLoadTimeOutTaskKey);
  624. downloadResult.setCode(ErrorCode.ERROR100.getCode());
  625. downloadResult.setMsg(String.format("录像下载失败, 错误码: %s, %s", event.statusCode, event.msg));
  626. downloadResult.setEvent(event);
  627. hookCallBack.call(downloadResult);
  628. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  629. };
  630. InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> {
  631. logger.info("[录像下载]收到订阅消息: " + inviteStreamInfo.getCallId());
  632. dynamicTask.stop(downLoadTimeOutTaskKey);
  633. StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
  634. streamInfo.setStartTime(startTime);
  635. streamInfo.setEndTime(endTime);
  636. redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
  637. downloadResult.setCode(ErrorCode.SUCCESS.getCode());
  638. downloadResult.setMsg(ErrorCode.SUCCESS.getMsg());
  639. downloadResult.setData(streamInfo);
  640. downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
  641. downloadResult.setResponse(inviteStreamInfo.getResponse());
  642. hookCallBack.call(downloadResult);
  643. };
  644. try {
  645. cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack,
  646. hookEvent, errorEvent, eventResult ->
  647. {
  648. if (eventResult.type == SipSubscribe.EventResultType.response) {
  649. ResponseEvent responseEvent = (ResponseEvent) eventResult.event;
  650. String contentString = new String(responseEvent.getResponse().getRawContent());
  651. // 获取ssrc
  652. int ssrcIndex = contentString.indexOf("y=");
  653. // 检查是否有y字段
  654. if (ssrcIndex >= 0) {
  655. //ssrc规定长度为10字节,不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容
  656. String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
  657. // 查询到ssrc不一致且开启了ssrc校验则需要针对处理
  658. if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
  659. return;
  660. }
  661. logger.info("[录像下载] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
  662. if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
  663. logger.info("[录像下载] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
  664. // 释放不被使用的ssrc
  665. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  666. // 单端口模式streamId也有变化,需要重新设置监听
  667. if (!mediaServerItem.isRtpEnable()) {
  668. // 添加订阅
  669. HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
  670. subscribe.removeSubscribe(hookSubscribe);
  671. hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
  672. subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
  673. logger.info("[ZLM HOOK] ssrc修正后收到订阅消息: " + response.toJSONString());
  674. dynamicTask.stop(downLoadTimeOutTaskKey);
  675. // hook响应
  676. onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, hookCallBack);
  677. hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
  678. });
  679. }
  680. // 关闭rtp server
  681. mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream(), result->{
  682. if (result) {
  683. // 重新开启ssrc server
  684. mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort(), true, device.getStreamModeForParam());
  685. }else {
  686. try {
  687. logger.warn("[录像下载] 停止{}/{}", device.getDeviceId(), channelId);
  688. cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null, null);
  689. } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
  690. logger.error("[命令发送失败] 录像下载停止, 发送BYE: {}", e.getMessage());
  691. throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage());
  692. }
  693. dynamicTask.stop(downLoadTimeOutTaskKey);
  694. // 释放ssrc
  695. mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  696. streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  697. eventResult.msg = "下级自定义了ssrc,重新设置收流信息失败";
  698. eventResult.statusCode = 500;
  699. errorEvent.response(eventResult);
  700. }
  701. });
  702. }
  703. }
  704. }
  705. });
  706. } catch (InvalidArgumentException | SipException | ParseException e) {
  707. logger.error("[命令发送失败] 录像下载: {}", e.getMessage());
  708. SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult();
  709. eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent;
  710. eventResult.statusCode = -1;
  711. eventResult.msg = "命令发送失败";
  712. errorEvent.response(eventResult);
  713. }
  714. }
  715. @Override
  716. public StreamInfo getDownLoadInfo(String deviceId, String channelId, String stream) {
  717. StreamInfo streamInfo = redisCatchStorage.queryDownload(deviceId, channelId, stream, null);
  718. if (streamInfo != null) {
  719. if (streamInfo.getProgress() == 1) {
  720. return streamInfo;
  721. }
  722. // 获取当前已下载时长
  723. String mediaServerId = streamInfo.getMediaServerId();
  724. MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
  725. if (mediaServerItem == null) {
  726. logger.warn("查询录像信息时发现节点已离线");
  727. return null;
  728. }
  729. if (mediaServerItem.getRecordAssistPort() > 0) {
  730. JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, streamInfo.getApp(), streamInfo.getStream(), null);
  731. if (jsonObject == null) {
  732. throw new ControllerException(ErrorCode.ERROR100.getCode(), "连接Assist服务失败");
  733. }
  734. if (jsonObject.getInteger("code") == 0) {
  735. long duration = jsonObject.getLong("data");
  736. if (duration == 0) {
  737. streamInfo.setProgress(0);
  738. } else {
  739. String startTime = streamInfo.getStartTime();
  740. String endTime = streamInfo.getEndTime();
  741. long start = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime);
  742. long end = DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime);
  743. BigDecimal currentCount = new BigDecimal(duration / 1000);
  744. BigDecimal totalCount = new BigDecimal(end - start);
  745. BigDecimal divide = currentCount.divide(totalCount, 2, RoundingMode.HALF_UP);
  746. double process = divide.doubleValue();
  747. streamInfo.setProgress(process);
  748. }
  749. }
  750. }
  751. }
  752. return streamInfo;
  753. }
  754. @Override
  755. public void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String uuid) {
  756. RequestMessage msg = new RequestMessage();
  757. msg.setKey(DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId);
  758. msg.setId(uuid);
  759. StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
  760. if (streamInfo != null) {
  761. redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
  762. msg.setData(JSON.toJSONString(streamInfo));
  763. resultHolder.invokeResult(msg);
  764. } else {
  765. logger.warn("设备预览API调用失败!");
  766. msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "设备预览API调用失败!"));
  767. resultHolder.invokeResult(msg);
  768. }
  769. }
  770. public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId) {
  771. String streamId = resonse.getString("stream");
  772. JSONArray tracks = resonse.getJSONArray("tracks");
  773. StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, "rtp", streamId, tracks, null);
  774. streamInfo.setDeviceID(deviceId);
  775. streamInfo.setChannelId(channelId);
  776. return streamInfo;
  777. }
  778. @Override
  779. public void zlmServerOffline(String mediaServerId) {
  780. // 处理正在向上推流的上级平台
  781. List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(null);
  782. if (sendRtpItems.size() > 0) {
  783. for (SendRtpItem sendRtpItem : sendRtpItems) {
  784. if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
  785. ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
  786. try {
  787. sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
  788. } catch (SipException | InvalidArgumentException | ParseException e) {
  789. logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
  790. }
  791. }
  792. }
  793. }
  794. // 处理正在观看的国标设备
  795. List<SsrcTransaction> allSsrc = streamSession.getAllSsrc();
  796. if (allSsrc.size() > 0) {
  797. for (SsrcTransaction ssrcTransaction : allSsrc) {
  798. if (ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
  799. Device device = deviceService.getDevice(ssrcTransaction.getDeviceId());
  800. if (device == null) {
  801. continue;
  802. }
  803. try {
  804. cmder.streamByeCmd(device, ssrcTransaction.getChannelId(),
  805. ssrcTransaction.getStream(), null);
  806. } catch (InvalidArgumentException | ParseException | SipException |
  807. SsrcTransactionNotFoundException e) {
  808. logger.error("[zlm离线]为正在使用此zlm的设备, 发送BYE失败 {}", e.getMessage());
  809. }
  810. }
  811. }
  812. }
  813. }
  814. @Override
  815. public void zlmServerOnline(String mediaServerId) {
  816. // TODO 查找之前的点播,流如果不存在则给下级发送bye
  817. // MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
  818. // zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
  819. // Integer code = mediaList.getInteger("code");
  820. // if (code == 0) {
  821. // JSONArray data = mediaList.getJSONArray("data");
  822. // if (data == null || data.size() == 0) {
  823. // zlmServerOffline(mediaServerId);
  824. // }else {
  825. // Map<String, JSONObject> mediaListMap = new HashMap<>();
  826. // for (int i = 0; i < data.size(); i++) {
  827. // JSONObject json = data.getJSONObject(i);
  828. // String app = json.getString("app");
  829. // if ("rtp".equals(app)) {
  830. // String stream = json.getString("stream");
  831. // if (mediaListMap.get(stream) != null) {
  832. // continue;
  833. // }
  834. // mediaListMap.put(stream, json);
  835. // // 处理正在观看的国标设备
  836. // List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(null, null, null, stream);
  837. // if (ssrcTransactions.size() > 0) {
  838. // for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
  839. // if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
  840. // cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(),
  841. // ssrcTransaction.getStream(), null);
  842. // }
  843. // }
  844. // }
  845. // }
  846. // }
  847. // if (mediaListMap.size() > 0 ) {
  848. // // 处理正在向上推流的上级平台
  849. // List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(null);
  850. // if (sendRtpItems.size() > 0) {
  851. // for (SendRtpItem sendRtpItem : sendRtpItems) {
  852. // if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
  853. // if (mediaListMap.get(sendRtpItem.getStreamId()) == null) {
  854. // ParentPlatform platform = storager.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
  855. // sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
  856. // }
  857. // }
  858. // }
  859. // }
  860. // }
  861. // }
  862. // }
  863. // }));
  864. }
  865. @Override
  866. public void pauseRtp(String streamId) throws ServiceException, InvalidArgumentException, ParseException, SipException {
  867. String key = redisCatchStorage.queryPlaybackForKey(null, null, streamId, null);
  868. StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
  869. if (null == streamInfo) {
  870. logger.warn("streamId不存在!");
  871. throw new ServiceException("streamId不存在");
  872. }
  873. streamInfo.setPause(true);
  874. redisTemplate.opsForValue().set(key, streamInfo);
  875. MediaServerItem mediaServerItem = mediaServerService.getOne(streamInfo.getMediaServerId());
  876. if (null == mediaServerItem) {
  877. logger.warn("mediaServer 不存在!");
  878. throw new ServiceException("mediaServer不存在");
  879. }
  880. // zlm 暂停RTP超时检查
  881. JSONObject jsonObject = zlmresTfulUtils.pauseRtpCheck(mediaServerItem, streamId);
  882. if (jsonObject == null || jsonObject.getInteger("code") != 0) {
  883. throw new ServiceException("暂停RTP接收失败");
  884. }
  885. Device device = storager.queryVideoDevice(streamInfo.getDeviceID());
  886. cmder.playPauseCmd(device, streamInfo);
  887. }
  888. @Override
  889. public void resumeRtp(String streamId) throws ServiceException, InvalidArgumentException, ParseException, SipException {
  890. String key = redisCatchStorage.queryPlaybackForKey(null, null, streamId, null);
  891. StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
  892. if (null == streamInfo) {
  893. logger.warn("streamId不存在!");
  894. throw new ServiceException("streamId不存在");
  895. }
  896. streamInfo.setPause(false);
  897. redisTemplate.opsForValue().set(key, streamInfo);
  898. MediaServerItem mediaServerItem = mediaServerService.getOne(streamInfo.getMediaServerId());
  899. if (null == mediaServerItem) {
  900. logger.warn("mediaServer 不存在!");
  901. throw new ServiceException("mediaServer不存在");
  902. }
  903. // zlm 暂停RTP超时检查
  904. JSONObject jsonObject = zlmresTfulUtils.resumeRtpCheck(mediaServerItem, streamId);
  905. if (jsonObject == null || jsonObject.getInteger("code") != 0) {
  906. throw new ServiceException("继续RTP接收失败");
  907. }
  908. Device device = storager.queryVideoDevice(streamInfo.getDeviceID());
  909. cmder.playResumeCmd(device, streamInfo);
  910. }
  911. }