From c1ac16bf620d09a94a0cfe184cbe1c9cb531a01b Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期六, 23 三月 2024 15:38:54 +0800
Subject: [PATCH] 优化媒体节点服务的代码结构
---
src/main/java/com/genersoft/iot/vmp/media/abl/ABLHttpHookListener.java | 906 +++++++++++++++++++++++++++++++
src/main/java/com/genersoft/iot/vmp/media/abl/ABLRESTfulUtils.java | 403 +++++++++++++
src/main/java/com/genersoft/iot/vmp/media/abl/event/HookAblServerKeepaliveEvent.java | 24
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 17
src/main/java/com/genersoft/iot/vmp/media/abl/bean/AblServerConfig.java | 4
src/main/java/com/genersoft/iot/vmp/media/abl/ABLMediaServerStatusManger.java | 305 ++++++++++
src/main/java/com/genersoft/iot/vmp/media/abl/event/HookAblServerStartEvent.java | 24
7 files changed, 1,671 insertions(+), 12 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/media/abl/ABLHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/abl/ABLHttpHookListener.java
new file mode 100755
index 0000000..026134a
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/media/abl/ABLHttpHookListener.java
@@ -0,0 +1,906 @@
+package com.genersoft.iot.vmp.media.abl;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.common.InviteInfo;
+import com.genersoft.iot.vmp.common.InviteSessionType;
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.common.VideoManagerConstants;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
+import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
+import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
+import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
+import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
+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.ISIPCommanderForPlatform;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
+import com.genersoft.iot.vmp.media.abl.event.HookAblServerKeepaliveEvent;
+import com.genersoft.iot.vmp.media.bean.MediaInfo;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
+import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
+import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
+import com.genersoft.iot.vmp.media.zlm.dto.*;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.*;
+import com.genersoft.iot.vmp.media.zlm.event.HookZlmServerStartEvent;
+import com.genersoft.iot.vmp.service.*;
+import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
+import com.genersoft.iot.vmp.service.bean.SSRCInfo;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.utils.DateUtil;
+import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
+import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo;
+import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo;
+import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
+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.context.ApplicationEventPublisher;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.ObjectUtils;
+import org.springframework.web.bind.annotation.*;
+import org.springframework.web.context.request.async.DeferredResult;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.sip.InvalidArgumentException;
+import javax.sip.SipException;
+import java.text.ParseException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * ABL鏈嶅姟Hook浜嬩欢
+ */
+@RestController
+@RequestMapping("/index/hook")
+public class ABLHttpHookListener {
+
+ private final static Logger logger = LoggerFactory.getLogger(ABLHttpHookListener.class);
+
+ @Autowired
+ private SIPCommander cmder;
+
+ @Autowired
+ private ISIPCommanderForPlatform commanderFroPlatform;
+
+ @Autowired
+ private AudioBroadcastManager audioBroadcastManager;
+
+ @Autowired
+ private IPlayService playService;
+
+ @Autowired
+ private IVideoManagerStorage storager;
+
+ @Autowired
+ private IRedisCatchStorage redisCatchStorage;
+
+ @Autowired
+ private IInviteStreamService inviteStreamService;
+
+ @Autowired
+ private IDeviceService deviceService;
+
+ @Autowired
+ private IMediaServerService mediaServerService;
+
+ @Autowired
+ private IStreamProxyService streamProxyService;
+
+ @Autowired
+ private DeferredResultHolder resultHolder;
+
+ @Autowired
+ private IMediaService mediaService;
+
+ @Autowired
+ private EventPublisher eventPublisher;
+
+ @Autowired
+ private ZLMMediaListManager zlmMediaListManager;
+
+ @Autowired
+ private ZlmHttpHookSubscribe subscribe;
+
+ @Autowired
+ private UserSetting userSetting;
+
+ @Autowired
+ private IUserService userService;
+
+ @Autowired
+ private ICloudRecordService cloudRecordService;
+
+ @Autowired
+ private VideoStreamSessionManager sessionManager;
+
+ @Autowired
+ private SSRCFactory ssrcFactory;
+
+ @Qualifier("taskExecutor")
+ @Autowired
+ private ThreadPoolTaskExecutor taskExecutor;
+
+ @Autowired
+ private RedisTemplate<Object, Object> redisTemplate;
+
+ @Autowired
+ private ApplicationEventPublisher applicationEventPublisher;
+
+ /**
+ * 鏈嶅姟鍣ㄥ畾鏃朵笂鎶ユ椂闂达紝涓婃姤闂撮殧鍙厤缃紝榛樿10s涓婃姤涓�娆�
+ */
+ @ResponseBody
+
+ @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
+ public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) {
+ try {
+ HookAblServerKeepaliveEvent event = new HookAblServerKeepaliveEvent(this);
+ MediaServer mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
+ if (mediaServerItem != null) {
+ event.setMediaServerItem(mediaServerItem);
+ applicationEventPublisher.publishEvent(event);
+ }
+ }catch (Exception e) {
+ logger.info("[ZLM-HOOK-蹇冭烦] 鍙戦�侀�氱煡澶辫触 ", e);
+ }
+ return HookResult.SUCCESS();
+ }
+
+ /**
+ * 鎾斁鍣ㄩ壌鏉冧簨浠讹紝rtsp/rtmp/http-flv/ws-flv/hls鐨勬挱鏀鹃兘灏嗚Е鍙戞閴存潈浜嬩欢銆�
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_play", produces = "application/json;charset=UTF-8")
+ public HookResult onPlay(@RequestBody OnPlayHookParam param) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("[ZLM HOOK] 鎾斁閴存潈锛歿}->{}", param.getMediaServerId(), param);
+ }
+ String mediaServerId = param.getMediaServerId();
+
+ taskExecutor.execute(() -> {
+ JSONObject json = (JSONObject) JSON.toJSON(param);
+ ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_play, json);
+ if (subscribe != null) {
+ MediaServer mediaInfo = mediaServerService.getOne(mediaServerId);
+ if (mediaInfo != null) {
+ subscribe.response(mediaInfo, param);
+ }
+ }
+ });
+ if (!"rtp".equals(param.getApp())) {
+ Map<String, String> paramMap = urlParamToMap(param.getParams());
+ StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
+ if (streamAuthorityInfo != null && streamAuthorityInfo.getCallId() != null && !streamAuthorityInfo.getCallId().equals(paramMap.get("callId"))) {
+ return new HookResult(401, "Unauthorized");
+ }
+ }
+
+ return HookResult.SUCCESS();
+ }
+
+ /**
+ * rtsp/rtmp/rtp鎺ㄦ祦閴存潈浜嬩欢銆�
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8")
+ public HookResultForOnPublish onPublish(@RequestBody OnPublishHookParam param) {
+
+ JSONObject json = (JSONObject) JSON.toJSON(param);
+
+ logger.info("[ABL HOOK]鎺ㄦ祦閴存潈锛歿}->{}", param.getMediaServerId(), param);
+ // TODO 鍔犲揩澶勭悊閫熷害
+
+ String mediaServerId = json.getString("mediaServerId");
+ MediaServer mediaInfo = mediaServerService.getOne(mediaServerId);
+ if (mediaInfo == null) {
+ return new HookResultForOnPublish(200, "success");
+ }
+ // 鎺ㄦ祦閴存潈鐨勫鐞�
+ if (!"rtp".equals(param.getApp())) {
+ StreamProxyItem stream = streamProxyService.getStreamProxyByAppAndStream(param.getApp(), param.getStream());
+ if (stream != null) {
+ HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();
+ result.setEnable_audio(stream.isEnableAudio());
+ result.setEnable_mp4(stream.isEnableMp4());
+ return result;
+ }
+ if (userSetting.getPushAuthority()) {
+ // 鎺ㄦ祦閴存潈
+ if (param.getParams() == null) {
+ logger.info("鎺ㄦ祦閴存潈澶辫触锛� 缂哄皯蹇呰鍙傛暟锛歴ign=md5(user琛ㄧ殑pushKey)");
+ return new HookResultForOnPublish(401, "Unauthorized");
+ }
+ Map<String, String> paramMap = urlParamToMap(param.getParams());
+ String sign = paramMap.get("sign");
+ if (sign == null) {
+ logger.info("鎺ㄦ祦閴存潈澶辫触锛� 缂哄皯蹇呰鍙傛暟锛歴ign=md5(user琛ㄧ殑pushKey)");
+ return new HookResultForOnPublish(401, "Unauthorized");
+ }
+ // 鎺ㄦ祦鑷畾涔夋挱鏀鹃壌鏉冪爜
+ String callId = paramMap.get("callId");
+ // 閴存潈閰嶇疆
+ boolean hasAuthority = userService.checkPushAuthority(callId, sign);
+ if (!hasAuthority) {
+ logger.info("鎺ㄦ祦閴存潈澶辫触锛� sign 鏃犳潈闄�: callId={}. sign={}", callId, sign);
+ return new HookResultForOnPublish(401, "Unauthorized");
+ }
+ StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
+ streamAuthorityInfo.setCallId(callId);
+ streamAuthorityInfo.setSign(sign);
+ // 閴存潈閫氳繃
+ redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
+ }
+ } else {
+ zlmMediaListManager.sendStreamEvent(param.getApp(), param.getStream(), param.getMediaServerId());
+ }
+
+
+ HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();
+ result.setEnable_audio(true);
+ taskExecutor.execute(() -> {
+ ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json);
+ if (subscribe != null) {
+ subscribe.response(mediaInfo, param);
+ }
+ });
+
+ // 鏄惁褰曞儚
+ if ("rtp".equals(param.getApp())) {
+ result.setEnable_mp4(userSetting.getRecordSip());
+ } else {
+ result.setEnable_mp4(userSetting.isRecordPushLive());
+ }
+ // 鍥芥爣娴�
+ if ("rtp".equals(param.getApp())) {
+
+ InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
+
+ // 鍗曠鍙ｆā寮忎笅淇敼娴� ID
+ if (!mediaInfo.isRtpEnable() && inviteInfo == null) {
+ String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));
+ inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
+ if (inviteInfo != null) {
+ result.setStream_replace(inviteInfo.getStream());
+ logger.info("[ABL HOOK]鎺ㄦ祦閴存潈 stream: {} 鏇挎崲涓� {}", param.getStream(), inviteInfo.getStream());
+ }
+ }
+
+ // 璁剧疆闊抽淇℃伅鍙婂綍鍒朵俊鎭�
+ List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, param.getStream());
+ if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
+
+ // 涓哄綍鍒跺浗鏍囨ā鎷熶竴涓壌鏉冧俊鎭�, 鏂逛究鍚庣画鍐欏叆褰曞儚鏂囦欢鏃朵娇鐢�
+ StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
+ streamAuthorityInfo.setApp(param.getApp());
+ streamAuthorityInfo.setStream(ssrcTransactionForAll.get(0).getStream());
+ streamAuthorityInfo.setCallId(ssrcTransactionForAll.get(0).getSipTransactionInfo().getCallId());
+
+ redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), ssrcTransactionForAll.get(0).getStream(), streamAuthorityInfo);
+
+ String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
+ String channelId = ssrcTransactionForAll.get(0).getChannelId();
+ DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+ if (deviceChannel != null) {
+ result.setEnable_audio(deviceChannel.isHasAudio());
+ }
+ // 濡傛灉鏄綍鍍忎笅杞藉氨璁剧疆瑙嗛闂撮殧鍗佺
+ if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {
+ // 鑾峰彇褰曞儚鐨勬�绘椂闀匡紝鐒跺悗璁剧疆涓鸿繖涓棰戠殑鏃堕暱
+ InviteInfo inviteInfoForDownload = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, deviceId, channelId, param.getStream());
+ if (inviteInfoForDownload != null && inviteInfoForDownload.getStreamInfo() != null) {
+ String startTime = inviteInfoForDownload.getStreamInfo().getStartTime();
+ String endTime = inviteInfoForDownload.getStreamInfo().getEndTime();
+ long difference = DateUtil.getDifference(startTime, endTime) / 1000;
+ result.setMp4_max_second((int) difference);
+ result.setEnable_mp4(true);
+ // 璁剧疆涓�2淇濊瘉寰楀埌鐨刴p4鐨勬椂闀挎槸姝ｅ父鐨�
+ result.setModify_stamp(2);
+ }
+ }
+ // 濡傛灉鏄痶alk瀵硅锛屽垯榛樿鑾峰彇澹伴煶
+ if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.TALK) {
+ result.setEnable_audio(true);
+ }
+ }
+ } else if (param.getApp().equals("broadcast")) {
+ result.setEnable_audio(true);
+ } else if (param.getApp().equals("talk")) {
+ result.setEnable_audio(true);
+ }
+ if (param.getApp().equalsIgnoreCase("rtp")) {
+ String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + param.getStream();
+ OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo) redisTemplate.opsForValue().get(receiveKey);
+
+ String receiveKeyForPS = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + param.getStream();
+ OtherPsSendInfo otherPsSendInfo = (OtherPsSendInfo) redisTemplate.opsForValue().get(receiveKeyForPS);
+ if (otherRtpSendInfo != null || otherPsSendInfo != null) {
+ result.setEnable_mp4(true);
+ }
+ }
+ logger.info("[ABL HOOK]鎺ㄦ祦閴存潈 鍝嶅簲锛歿}->{}->>>>{}", param.getMediaServerId(), param, result);
+ return result;
+ }
+
+
+ /**
+ * rtsp/rtmp娴佹敞鍐屾垨娉ㄩ攢鏃惰Е鍙戞浜嬩欢锛涙浜嬩欢瀵瑰洖澶嶄笉鏁忔劅銆�
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_stream_changed", produces = "application/json;charset=UTF-8")
+ public HookResult onStreamChanged(@RequestBody OnStreamChangedHookParam param) {
+
+ if (param.isRegist()) {
+ logger.info("[ABL HOOK] 娴佹敞鍐�, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
+ } else {
+ logger.info("[ABL HOOK] 娴佹敞閿�, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
+ }
+
+ JSONObject json = (JSONObject) JSON.toJSON(param);
+ taskExecutor.execute(() -> {
+ ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json);
+ MediaServer mediaInfo = mediaServerService.getOne(param.getMediaServerId());
+ if (mediaInfo == null) {
+ logger.info("[ABL HOOK] 娴佸彉鍖栨湭鎵惧埌ABL, {}", param.getMediaServerId());
+ return;
+ }
+ if (subscribe != null) {
+ subscribe.response(mediaInfo, param);
+ }
+
+ // TODO 閲嶆瀯姝ゅ閫昏緫
+ if (param.isRegist()) {
+ // 澶勭悊娴佹敞鍐岀殑閴存潈淇℃伅锛� 娴佹敞閿�杩欓噷涓嶅啀鍒犻櫎閴存潈淇℃伅锛屼笅娆℃潵浜嗘柊鐨勯壌鏉冧俊鎭細瀵瑰氨鐨勮繘琛岃鐩�
+ if (param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
+ || param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
+ || param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
+ StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
+ if (streamAuthorityInfo == null) {
+ streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
+ } else {
+ streamAuthorityInfo.setOriginType(param.getOriginType());
+ streamAuthorityInfo.setOriginTypeStr(param.getOriginTypeStr());
+ }
+ redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
+ }
+ }
+ if ("rtsp".equals(param.getSchema())) {
+ logger.info("娴佸彉鍖栵細娉ㄥ唽->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream());
+ if (param.isRegist()) {
+ mediaServerService.addCount(param.getMediaServerId());
+ } else {
+ mediaServerService.removeCount(param.getMediaServerId());
+ }
+
+ int updateStatusResult = streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
+ if (updateStatusResult > 0) {
+
+ }
+
+ if ("rtp".equals(param.getApp()) && !param.isRegist()) {
+ InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
+ if (inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY || inviteInfo.getType() == InviteSessionType.PLAYBACK)) {
+ inviteStreamService.removeInviteInfo(inviteInfo);
+ storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
+ }
+ } else if ("broadcast".equals(param.getApp())) {
+ // 璇煶瀵硅鎺ㄦ祦 stream闇�瑕佹弧瓒虫牸寮廳eviceId_channelId
+ if (param.getStream().indexOf("_") > 0) {
+ String[] streamArray = param.getStream().split("_");
+ if (streamArray.length == 2) {
+ String deviceId = streamArray[0];
+ String channelId = streamArray[1];
+ Device device = deviceService.getDevice(deviceId);
+ if (device != null) {
+ if (param.isRegist()) {
+ if (audioBroadcastManager.exit(deviceId, channelId)) {
+ playService.stopAudioBroadcast(deviceId, channelId);
+ }
+ // 寮�鍚闊冲璁查�氶亾
+ try {
+ playService.audioBroadcastCmd(device, channelId, mediaInfo, param.getApp(), param.getStream(), 60, false, (msg) -> {
+ logger.info("[璇煶瀵硅] 閫氶亾寤虹珛鎴愬姛, device: {}, channel: {}", deviceId, channelId);
+ });
+ } catch (InvalidArgumentException | ParseException | SipException e) {
+ logger.error("[鍛戒护鍙戦�佸け璐 璇煶瀵硅: {}", e.getMessage());
+ }
+ } else {
+ // 娴佹敞閿�
+ playService.stopAudioBroadcast(deviceId, channelId);
+ }
+ } else {
+ logger.info("[璇煶瀵硅] 鏈壘鍒拌澶囷細{}", deviceId);
+ }
+ }
+ }
+ } else if ("talk".equals(param.getApp())) {
+ // 璇煶瀵硅鎺ㄦ祦 stream闇�瑕佹弧瓒虫牸寮廳eviceId_channelId
+ if (param.getStream().indexOf("_") > 0) {
+ String[] streamArray = param.getStream().split("_");
+ if (streamArray.length == 2) {
+ String deviceId = streamArray[0];
+ String channelId = streamArray[1];
+ Device device = deviceService.getDevice(deviceId);
+ if (device != null) {
+ if (param.isRegist()) {
+ if (audioBroadcastManager.exit(deviceId, channelId)) {
+ playService.stopAudioBroadcast(deviceId, channelId);
+ }
+ // 寮�鍚闊冲璁查�氶亾
+ playService.talkCmd(device, channelId, mediaInfo, param.getStream(), (msg) -> {
+ logger.info("[璇煶瀵硅] 閫氶亾寤虹珛鎴愬姛, device: {}, channel: {}", deviceId, channelId);
+ });
+ } else {
+ // 娴佹敞閿�
+ playService.stopTalk(device, channelId, param.isRegist());
+ }
+ } else {
+ logger.info("[璇煶瀵硅] 鏈壘鍒拌澶囷細{}", deviceId);
+ }
+ }
+ }
+
+ } else {
+ if (!"rtp".equals(param.getApp())) {
+ String type = OriginType.values()[param.getOriginType()].getType();
+ if (param.isRegist()) {
+ StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(
+ param.getApp(), param.getStream());
+ String callId = null;
+ if (streamAuthorityInfo != null) {
+ callId = streamAuthorityInfo.getCallId();
+ }
+ StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaInfo,
+ param.getApp(), param.getStream(), MediaInfo.getInstance(param), callId);
+ param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));
+ redisCatchStorage.addStream(mediaInfo, type, param.getApp(), param.getStream(), param);
+ if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
+ || param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
+ || param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
+ param.setSeverId(userSetting.getServerId());
+ zlmMediaListManager.addPush(param);
+
+ // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+ redisCatchStorage.addPushListItem(param.getApp(), param.getStream(), param);
+ }
+ } else {
+ // 鍏煎娴佹敞閿�鏃剁被鍨嬩粠redis璁板綍鑾峰彇
+ OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(
+ param.getApp(), param.getStream(), param.getMediaServerId());
+ if (onStreamChangedHookParam != null) {
+ type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();
+ redisCatchStorage.removeStream(mediaInfo.getId(), type, param.getApp(), param.getStream());
+ if ("PUSH".equalsIgnoreCase(type)) {
+ // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+ redisCatchStorage.removePushListItem(param.getApp(), param.getStream(), param.getMediaServerId());
+ }
+ }
+ GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
+ if (gbStream != null) {
+// eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
+ }
+ zlmMediaListManager.removeMedia(param.getApp(), param.getStream());
+ }
+ GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
+ if (gbStream != null) {
+ if (userSetting.isUsePushingAsStatus()) {
+ eventPublisher.catalogEventPublishForStream(null, gbStream, param.isRegist() ? CatalogEvent.ON : CatalogEvent.OFF);
+ }
+ }
+ if (type != null) {
+ // 鍙戦�佹祦鍙樺寲redis娑堟伅
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("serverId", userSetting.getServerId());
+ jsonObject.put("app", param.getApp());
+ jsonObject.put("stream", param.getStream());
+ jsonObject.put("register", param.isRegist());
+ jsonObject.put("mediaServerId", param.getMediaServerId());
+ redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+ }
+ }
+ }
+ if (!param.isRegist()) {
+ List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(param.getStream());
+ if (!sendRtpItems.isEmpty()) {
+ for (SendRtpItem sendRtpItem : sendRtpItems) {
+ if (sendRtpItem != null && sendRtpItem.getApp().equals(param.getApp())) {
+ String platformId = sendRtpItem.getPlatformId();
+ ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
+ Device device = deviceService.getDevice(platformId);
+
+ try {
+ if (platform != null) {
+ commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
+ redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(),
+ sendRtpItem.getCallId(), sendRtpItem.getStream());
+ } else {
+ cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
+ if (sendRtpItem.getPlayType().equals(InviteStreamType.BROADCAST)
+ || sendRtpItem.getPlayType().equals(InviteStreamType.TALK)) {
+ AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
+ if (audioBroadcastCatch != null) {
+ // 鏉ヨ嚜涓婄骇骞冲彴鐨勫仠姝㈠璁�
+ logger.info("[鍋滄瀵硅] 鏉ヨ嚜涓婄骇锛屽钩鍙帮細{}, 閫氶亾锛歿}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
+ audioBroadcastManager.del(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
+ }
+ }
+ }
+ } catch (SipException | InvalidArgumentException | ParseException |
+ SsrcTransactionNotFoundException e) {
+ logger.error("[鍛戒护鍙戦�佸け璐 鍙戦�丅YE: {}", e.getMessage());
+ }
+ }
+ }
+ }
+ }
+ }
+ });
+ return HookResult.SUCCESS();
+ }
+
+ /**
+ * 娴佹棤浜鸿鐪嬫椂浜嬩欢锛岀敤鎴峰彲浠ラ�氳繃姝や簨浠堕�夋嫨鏄惁鍏抽棴鏃犱汉鐪嬬殑娴併��
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8")
+ public JSONObject onStreamNoneReader(@RequestBody OnStreamNoneReaderHookParam param) {
+
+ logger.info("[ABL HOOK]娴佹棤浜鸿鐪嬶細{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(),
+ param.getApp(), param.getStream());
+ JSONObject ret = new JSONObject();
+ ret.put("code", 0);
+ // 鍥芥爣绫诲瀷鐨勬祦
+ if ("rtp".equals(param.getApp())) {
+ ret.put("close", userSetting.getStreamOnDemand());
+ // 鍥芥爣娴侊紝 鐐规挱/褰曞儚鍥炴斁/褰曞儚涓嬭浇
+ InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
+ // 鐐规挱
+ if (inviteInfo != null) {
+ // 褰曞儚涓嬭浇
+ if (inviteInfo.getType() == InviteSessionType.DOWNLOAD) {
+ ret.put("close", false);
+ return ret;
+ }
+ // 鏀跺埌鏃犱汉瑙傜湅璇存槑娴佷篃娌℃湁鍦ㄥ線涓婄骇鎺ㄩ��
+ if (redisCatchStorage.isChannelSendingRTP(inviteInfo.getChannelId())) {
+ List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChannelId(
+ inviteInfo.getChannelId());
+ if (!sendRtpItems.isEmpty()) {
+ for (SendRtpItem sendRtpItem : sendRtpItems) {
+ ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+ try {
+ commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
+ }
+ redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
+ sendRtpItem.getCallId(), sendRtpItem.getStream());
+ if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
+ MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
+ sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
+ sendRtpItem.getPlatformId(), parentPlatform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
+ messageForPushChannel.setPlatFormIndex(parentPlatform.getId());
+ redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
+ }
+ }
+ }
+ }
+ Device device = deviceService.getDevice(inviteInfo.getDeviceId());
+ if (device != null) {
+ try {
+ // 澶氭煡璇竴娆￠槻姝㈠凡缁忚澶勭悊浜�
+ InviteInfo info = inviteStreamService.getInviteInfo(inviteInfo.getType(),
+ inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream());
+ if (info != null) {
+ cmder.streamByeCmd(device, inviteInfo.getChannelId(),
+ inviteInfo.getStream(), null);
+ } else {
+ logger.info("[鏃犱汉瑙傜湅] 鏈壘鍒拌澶囩殑鐐规挱淇℃伅锛� {}锛� 娴侊細{}", inviteInfo.getDeviceId(), param.getStream());
+ }
+ } catch (InvalidArgumentException | ParseException | SipException |
+ SsrcTransactionNotFoundException e) {
+ logger.error("[鏃犱汉瑙傜湅]鐐规挱锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
+ }
+ } else {
+ logger.info("[鏃犱汉瑙傜湅] 鏈壘鍒拌澶囷細 {}锛屾祦锛歿}", inviteInfo.getDeviceId(), param.getStream());
+ }
+
+ inviteStreamService.removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(),
+ inviteInfo.getChannelId(), inviteInfo.getStream());
+ storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
+ return ret;
+ }
+ SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);
+ if (sendRtpItem != null && "talk".equals(sendRtpItem.getApp())) {
+ ret.put("close", false);
+ return ret;
+ }
+ } else if ("talk".equals(param.getApp()) || "broadcast".equals(param.getApp())) {
+ ret.put("close", false);
+ } else {
+ // 闈炲浗鏍囨祦 鎺ㄦ祦/鎷夋祦浠ｇ悊
+ // 鎷夋祦浠ｇ悊
+ StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(param.getApp(), param.getStream());
+ if (streamProxyItem != null) {
+ if (streamProxyItem.isEnableRemoveNoneReader()) {
+ // 鏃犱汉瑙傜湅鑷姩绉婚櫎
+ ret.put("close", true);
+ streamProxyService.del(param.getApp(), param.getStream());
+ String url = streamProxyItem.getUrl() != null ? streamProxyItem.getUrl() : streamProxyItem.getSrcUrl();
+ logger.info("[{}/{}]<-[{}] 鎷夋祦浠ｇ悊鏃犱汉瑙傜湅宸茬粡绉婚櫎", param.getApp(), param.getStream(), url);
+ } else if (streamProxyItem.isEnableDisableNoneReader()) {
+ // 鏃犱汉瑙傜湅鍋滅敤
+ ret.put("close", true);
+ // 淇敼鏁版嵁
+ streamProxyService.stop(param.getApp(), param.getStream());
+ } else {
+ // 鏃犱汉瑙傜湅涓嶅仛澶勭悊
+ ret.put("close", false);
+ }
+ return ret;
+ }
+ // TODO 鎺ㄦ祦鍏锋湁涓诲姩鎬э紝鏆傛椂涓嶅仛澶勭悊
+// StreamPushItem streamPushItem = streamPushService.getPush(app, streamId);
+// if (streamPushItem != null) {
+// // TODO 鍙戦�佸仠姝�
+//
+// }
+ }
+ return ret;
+ }
+
+ /**
+ * 娴佹湭鎵惧埌浜嬩欢锛岀敤鎴峰彲浠ュ湪姝や簨浠惰Е鍙戞椂锛岀珛鍗冲幓鎷夋祦锛岃繖鏍峰彲浠ュ疄鐜版寜闇�鎷夋祦锛涙浜嬩欢瀵瑰洖澶嶄笉鏁忔劅銆�
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8")
+ public DeferredResult<HookResult> onStreamNotFound(@RequestBody OnStreamNotFoundHookParam param) {
+ logger.info("[ABL HOOK] 娴佹湭鎵惧埌锛歿}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
+
+ DeferredResult<HookResult> defaultResult = new DeferredResult<>();
+
+ MediaServer mediaInfo = mediaServerService.getOne(param.getMediaServerId());
+ if (!userSetting.isAutoApplyPlay() || mediaInfo == null) {
+ defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));
+ return defaultResult;
+ }
+
+ if ("rtp".equals(param.getApp())) {
+ String[] s = param.getStream().split("_");
+ if ((s.length != 2 && s.length != 4)) {
+ defaultResult.setResult(HookResult.SUCCESS());
+ return defaultResult;
+ }
+ String deviceId = s[0];
+ String channelId = s[1];
+ Device device = redisCatchStorage.getDevice(deviceId);
+ if (device == null || !device.isOnLine()) {
+ defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));
+ return defaultResult;
+ }
+ DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+ if (deviceChannel == null) {
+ defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));
+ return defaultResult;
+ }
+ if (s.length == 2) {
+ logger.info("[ABL HOOK] 棰勮娴佹湭鎵惧埌, 鍙戣捣鑷姩鐐规挱锛歿}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
+
+ RequestMessage msg = new RequestMessage();
+ String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;
+ boolean exist = resultHolder.exist(key, null);
+ msg.setKey(key);
+ String uuid = UUID.randomUUID().toString();
+ msg.setId(uuid);
+ DeferredResult<HookResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
+
+ result.onTimeout(() -> {
+ logger.info("[ABL HOOK] 棰勮娴佽嚜鍔ㄧ偣鎾�, 绛夊緟瓒呮椂");
+ msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "鐐规挱瓒呮椂"));
+ resultHolder.invokeAllResult(msg);
+ inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
+ storager.stopPlay(deviceId, channelId);
+ });
+
+ resultHolder.put(key, uuid, result);
+
+ if (!exist) {
+ playService.play(mediaInfo, deviceId, channelId, null, (code, message, data) -> {
+ msg.setData(new HookResult(code, message));
+ resultHolder.invokeResult(msg);
+ });
+ }
+ return result;
+ } else if (s.length == 4) {
+ // 姝ゆ椂涓哄綍鍍忓洖鏀撅紝 褰曞儚鍥炴斁鏍煎紡涓�> 璁惧ID_閫氶亾ID_寮�濮嬫椂闂確缁撴潫鏃堕棿
+ String startTimeStr = s[2];
+ String endTimeStr = s[3];
+ if (startTimeStr == null || endTimeStr == null || startTimeStr.length() != 14 || endTimeStr.length() != 14) {
+ defaultResult.setResult(HookResult.SUCCESS());
+ return defaultResult;
+ }
+ String startTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(startTimeStr);
+ String endTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(endTimeStr);
+ logger.info("[ABL HOOK] 鍥炴斁娴佹湭鎵惧埌, 鍙戣捣鑷姩鐐规挱锛歿}->{}->{}/{}-{}-{}",
+ param.getMediaServerId(), param.getSchema(),
+ param.getApp(), param.getStream(),
+ startTime, endTime
+ );
+ RequestMessage msg = new RequestMessage();
+ String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;
+ boolean exist = resultHolder.exist(key, null);
+ msg.setKey(key);
+ String uuid = UUID.randomUUID().toString();
+ msg.setId(uuid);
+ DeferredResult<HookResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
+
+ result.onTimeout(() -> {
+ logger.info("[ABL HOOK] 鍥炴斁娴佽嚜鍔ㄧ偣鎾�, 绛夊緟瓒呮椂");
+ // 閲婃斁rtpserver
+ msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "鐐规挱瓒呮椂"));
+ resultHolder.invokeResult(msg);
+ });
+
+ resultHolder.put(key, uuid, result);
+
+ if (!exist) {
+ SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaInfo, param.getStream(), null,
+ device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam());
+ playService.playBack(mediaInfo, ssrcInfo, deviceId, channelId, startTime, endTime, (code, message, data) -> {
+ msg.setData(new HookResult(code, message));
+ resultHolder.invokeResult(msg);
+ });
+ }
+ return result;
+ } else {
+ defaultResult.setResult(HookResult.SUCCESS());
+ return defaultResult;
+ }
+
+ } else {
+ // 鎷夋祦浠ｇ悊
+ StreamProxyItem streamProxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(param.getApp(), param.getStream());
+ if (streamProxyByAppAndStream != null && streamProxyByAppAndStream.isEnableDisableNoneReader()) {
+ streamProxyService.start(param.getApp(), param.getStream());
+ }
+ DeferredResult<HookResult> result = new DeferredResult<>();
+ result.setResult(HookResult.SUCCESS());
+ return result;
+ }
+ }
+
+ /**
+ * 鏈嶅姟鍣ㄥ惎鍔ㄤ簨浠讹紝鍙互鐢ㄤ簬鐩戝惉鏈嶅姟鍣ㄥ穿婧冮噸鍚紱姝や簨浠跺鍥炲涓嶆晱鎰熴��
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_server_started", produces = "application/json;charset=UTF-8")
+ public HookResult onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject) {
+
+ jsonObject.put("ip", request.getRemoteAddr());
+ ZLMServerConfig zlmServerConfig = JSON.to(ZLMServerConfig.class, jsonObject);
+ zlmServerConfig.setIp(request.getRemoteAddr());
+ logger.info("[ABL HOOK] ABL 鍚姩 " + zlmServerConfig.getGeneralMediaServerId());
+ taskExecutor.execute(() -> {
+ List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_started);
+ if (subscribes != null && !subscribes.isEmpty()) {
+ for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {
+ subscribe.response(null, zlmServerConfig);
+ }
+ }
+ });
+ try {
+ HookZlmServerStartEvent event = new HookZlmServerStartEvent(this);
+ MediaServer mediaServerItem = mediaServerService.getOne(zlmServerConfig.getMediaServerId());
+ if (mediaServerItem != null) {
+ event.setMediaServerItem(mediaServerItem);
+ applicationEventPublisher.publishEvent(event);
+ }
+ }catch (Exception e) {
+ logger.info("[ABL-HOOK-ABL鍚姩] 鍙戦�侀�氱煡澶辫触 ", e);
+ }
+
+ return HookResult.SUCCESS();
+ }
+
+ /**
+ * 鍙戦�乺tp(startSendRtp)琚姩鍏抽棴鏃跺洖璋�
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_send_rtp_stopped", produces = "application/json;charset=UTF-8")
+ public HookResult onSendRtpStopped(HttpServletRequest request, @RequestBody OnSendRtpStoppedHookParam param) {
+
+ logger.info("[ABL HOOK] rtp鍙戦�佸叧闂細{}->{}/{}", param.getMediaServerId(), param.getApp(), param.getStream());
+
+ // 鏌ユ壘瀵瑰簲鐨勪笂绾ф帹娴侊紝鍙戦�佸仠姝�
+ if (!"rtp".equals(param.getApp())) {
+ return HookResult.SUCCESS();
+ }
+ taskExecutor.execute(() -> {
+ List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(param.getStream());
+ if (sendRtpItems.size() > 0) {
+ for (SendRtpItem sendRtpItem : sendRtpItems) {
+ ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+ ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
+ try {
+ commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
+ }
+ redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
+ sendRtpItem.getCallId(), sendRtpItem.getStream());
+ }
+ }
+ });
+
+ return HookResult.SUCCESS();
+ }
+
+ /**
+ * rtpServer鏀舵祦瓒呮椂
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_rtp_server_timeout", produces = "application/json;charset=UTF-8")
+ public HookResult onRtpServerTimeout(@RequestBody OnRtpServerTimeoutHookParam
+ param) {
+ logger.info("[ABL HOOK] rtpServer鏀舵祦瓒呮椂锛歿}->{}({})", param.getMediaServerId(), param.getStream_id(), param.getSsrc());
+
+ taskExecutor.execute(() -> {
+ List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_rtp_server_timeout);
+ if (subscribes != null && !subscribes.isEmpty()) {
+ for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {
+ subscribe.response(null, param);
+ }
+ }
+ });
+
+ return HookResult.SUCCESS();
+ }
+
+ /**
+ * 褰曞儚瀹屾垚浜嬩欢
+ */
+ @ResponseBody
+ @PostMapping(value = "/on_record_mp4", produces = "application/json;charset=UTF-8")
+ public HookResult onRecordMp4(HttpServletRequest request, @RequestBody OnRecordMp4HookParam param) {
+ logger.info("[AB HOOK] 褰曞儚瀹屾垚浜嬩欢锛歿}->{}", param.getMediaServerId(), param.getFile_path());
+
+ taskExecutor.execute(() -> {
+ List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_record_mp4);
+ if (subscribes != null && !subscribes.isEmpty()) {
+ for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {
+ subscribe.response(null, param);
+ }
+ }
+ cloudRecordService.addRecord(param);
+
+ });
+
+ return HookResult.SUCCESS();
+ }
+
+ private Map<String, String> urlParamToMap(String params) {
+ HashMap<String, String> map = new HashMap<>();
+ if (ObjectUtils.isEmpty(params)) {
+ return map;
+ }
+ String[] paramsArray = params.split("&");
+ if (paramsArray.length == 0) {
+ return map;
+ }
+ for (String param : paramsArray) {
+ String[] paramArray = param.split("=");
+ if (paramArray.length == 2) {
+ map.put(paramArray[0], paramArray[1]);
+ }
+ }
+ return map;
+ }
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/media/abl/ABLMediaServerStatusManger.java b/src/main/java/com/genersoft/iot/vmp/media/abl/ABLMediaServerStatusManger.java
new file mode 100644
index 0000000..28181e8
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/media/abl/ABLMediaServerStatusManger.java
@@ -0,0 +1,305 @@
+package com.genersoft.iot.vmp.media.abl;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.conf.DynamicTask;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.media.abl.bean.AblServerConfig;
+import com.genersoft.iot.vmp.media.abl.event.HookAblServerKeepaliveEvent;
+import com.genersoft.iot.vmp.media.abl.event.HookAblServerStartEvent;
+import com.genersoft.iot.vmp.media.event.MediaServerChangeEvent;
+import com.genersoft.iot.vmp.media.event.MediaServerDeleteEvent;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
+import com.genersoft.iot.vmp.media.zlm.dto.ZLMServerConfig;
+import com.genersoft.iot.vmp.media.zlm.event.HookZlmServerKeepaliveEvent;
+import com.genersoft.iot.vmp.media.zlm.event.HookZlmServerStartEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ObjectUtils;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 绠＄悊zlm娴佸獟浣撹妭鐐圭殑鐘舵��
+ */
+@Component
+public class ABLMediaServerStatusManger {
+
+ private final static Logger logger = LoggerFactory.getLogger(ABLMediaServerStatusManger.class);
+
+ private final Map<Object, MediaServer> offlineABLPrimaryMap = new ConcurrentHashMap<>();
+ private final Map<Object, MediaServer> offlineAblsecondaryMap = new ConcurrentHashMap<>();
+ private final Map<Object, Long> offlineAblTimeMap = new ConcurrentHashMap<>();
+
+ @Autowired
+ private ABLRESTfulUtils ablResTfulUtils;
+
+ @Autowired
+ private IMediaServerService mediaServerService;
+
+ @Autowired
+ private DynamicTask dynamicTask;
+
+ @Value("${server.ssl.enabled:false}")
+ private boolean sslEnabled;
+
+ @Value("${server.port}")
+ private Integer serverPort;
+
+ @Autowired
+ private UserSetting userSetting;
+
+ private final String type = "abl";
+
+ @Async("taskExecutor")
+ @EventListener
+ public void onApplicationEvent(MediaServerChangeEvent event) {
+ if (event.getMediaServerItemList() == null
+ || event.getMediaServerItemList().isEmpty()) {
+ return;
+ }
+ for (MediaServer mediaServerItem : event.getMediaServerItemList()) {
+ if (!type.equals(mediaServerItem.getType())) {
+ continue;
+ }
+ logger.info("[ABL-娣诲姞寰呬笂绾胯妭鐐筣 ID锛�" + mediaServerItem.getId());
+ offlineABLPrimaryMap.put(mediaServerItem.getId(), mediaServerItem);
+ offlineAblTimeMap.put(mediaServerItem.getId(), System.currentTimeMillis());
+ }
+ }
+
+ @Async("taskExecutor")
+ @EventListener
+ public void onApplicationEvent(HookAblServerStartEvent event) {
+ if (event.getMediaServerItem() == null
+ || !type.equals(event.getMediaServerItem().getType())
+ || event.getMediaServerItem().isStatus()) {
+ return;
+ }
+ MediaServer serverItem = mediaServerService.getOne(event.getMediaServerItem().getId());
+ if (serverItem == null) {
+ return;
+ }
+ logger.info("[ABL-HOOK浜嬩欢-鏈嶅姟鍚姩] ID锛�" + event.getMediaServerItem().getId());
+ online(serverItem, null);
+ }
+
+ @Async("taskExecutor")
+ @EventListener
+ public void onApplicationEvent(HookAblServerKeepaliveEvent event) {
+ if (event.getMediaServerItem() == null) {
+ return;
+ }
+ MediaServer serverItem = mediaServerService.getOne(event.getMediaServerItem().getId());
+ if (serverItem == null) {
+ return;
+ }
+ logger.info("[ABL-HOOK浜嬩欢-蹇冭烦] ID锛�" + event.getMediaServerItem().getId());
+ online(serverItem, null);
+ }
+
+ @Async("taskExecutor")
+ @EventListener
+ public void onApplicationEvent(MediaServerDeleteEvent event) {
+ if (event.getMediaServerId() == null) {
+ return;
+ }
+ logger.info("[ABL-鑺傜偣琚Щ闄 ID锛�" + event.getMediaServerId());
+ offlineABLPrimaryMap.remove(event.getMediaServerId());
+ offlineAblsecondaryMap.remove(event.getMediaServerId());
+ offlineAblTimeMap.remove(event.getMediaServerId());
+ }
+
+ @Scheduled(fixedDelay = 10*1000) //姣忛殧10绉掓鏌ヤ竴娆�
+ public void execute(){
+ // 鍒濇鍔犲叆鐨勭绾胯妭鐐逛細鍦�30鍒嗛挓鍐咃紝姣忛棿闅斿崄绉掑皾璇曚竴娆★紝30鍒嗛挓鍚庡鏋滀粛鐒舵病鏈変笂绾匡紝鍒欐瘡闅�30鍒嗛挓灏濊瘯涓�娆¤繛鎺�
+ if (offlineABLPrimaryMap.isEmpty() && offlineAblsecondaryMap.isEmpty()) {
+ return;
+ }
+ if (!offlineABLPrimaryMap.isEmpty()) {
+ for (MediaServer mediaServerItem : offlineABLPrimaryMap.values()) {
+ if (offlineAblTimeMap.get(mediaServerItem.getId()) < System.currentTimeMillis() - 30*60*1000) {
+ offlineAblsecondaryMap.put(mediaServerItem.getId(), mediaServerItem);
+ offlineABLPrimaryMap.remove(mediaServerItem.getId());
+ continue;
+ }
+ logger.info("[ABL-灏濊瘯杩炴帴] ID锛歿}, 鍦板潃锛� {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ JSONObject responseJson = ablResTfulUtils.getMediaServerConfig(mediaServerItem);
+ AblServerConfig ablServerConfig = null;
+ if (responseJson == null) {
+ logger.info("[ABL-灏濊瘯杩炴帴]澶辫触, ID锛歿}, 鍦板潃锛� {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ continue;
+ }
+ JSONArray data = responseJson.getJSONArray("data");
+ if (data == null || data.isEmpty()) {
+ logger.info("[ABL-灏濊瘯杩炴帴]澶辫触, ID锛歿}, 鍦板潃锛� {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ }else {
+ ablServerConfig = JSON.parseObject(JSON.toJSONString(data.get(0)), AblServerConfig.class);
+ initPort(mediaServerItem, ablServerConfig);
+ online(mediaServerItem, ablServerConfig);
+ }
+ }
+ }
+ if (!offlineAblsecondaryMap.isEmpty()) {
+ for (MediaServer mediaServerItem : offlineAblsecondaryMap.values()) {
+ if (offlineAblTimeMap.get(mediaServerItem.getId()) < System.currentTimeMillis() - 30*60*1000) {
+ continue;
+ }
+ logger.info("[ABL-灏濊瘯杩炴帴] ID锛歿}, 鍦板潃锛� {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ JSONObject responseJson = ablResTfulUtils.getMediaServerConfig(mediaServerItem);
+ ZLMServerConfig zlmServerConfig = null;
+ if (responseJson == null) {
+ logger.info("[ABL-灏濊瘯杩炴帴]澶辫触, ID锛歿}, 鍦板潃锛� {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ offlineAblTimeMap.put(mediaServerItem.getId(), System.currentTimeMillis());
+ continue;
+ }
+ JSONArray data = responseJson.getJSONArray("data");
+ if (data == null || data.isEmpty()) {
+ logger.info("[ABL-灏濊瘯杩炴帴]澶辫触, ID锛歿}, 鍦板潃锛� {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ offlineAblTimeMap.put(mediaServerItem.getId(), System.currentTimeMillis());
+ }else {
+ zlmServerConfig = JSON.parseObject(JSON.toJSONString(data.get(0)), ZLMServerConfig.class);
+ initPort(mediaServerItem, zlmServerConfig);
+ online(mediaServerItem, zlmServerConfig);
+ }
+ }
+ }
+ }
+
+ private void online(MediaServer mediaServerItem, AblServerConfig config) {
+ offlineABLPrimaryMap.remove(mediaServerItem.getId());
+ offlineAblsecondaryMap.remove(mediaServerItem.getId());
+ offlineAblTimeMap.remove(mediaServerItem.getId());
+ if (!mediaServerItem.isStatus()) {
+ logger.info("[ABL-杩炴帴鎴愬姛] ID锛歿}, 鍦板潃锛� {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ mediaServerItem.setStatus(true);
+ mediaServerItem.setHookAliveInterval(10F);
+ mediaServerService.update(mediaServerItem);
+ if(mediaServerItem.isAutoConfig()) {
+ if (config == null) {
+ JSONObject responseJSON = ablResTfulUtils.getMediaServerConfig(mediaServerItem);
+ JSONArray data = responseJSON.getJSONArray("data");
+ if (data != null && !data.isEmpty()) {
+ config = JSON.parseObject(JSON.toJSONString(data.get(0)), AblServerConfig.class);
+ }
+ }
+ if (config != null) {
+ initPort(mediaServerItem, config);
+ setZLMConfig(mediaServerItem, "0".equals(config.getHookEnable())
+ || !Objects.equals(mediaServerItem.getHookAliveInterval(), config.getHookAliveInterval()));
+ }
+ }
+ mediaServerService.update(mediaServerItem);
+ }
+ // 璁剧疆涓ゆ蹇冭烦鏈敹鍒板垯璁や负zlm绂荤嚎
+ String key = "ABL-keepalive-" + mediaServerItem.getId();
+ dynamicTask.startDelay(key, ()->{
+ logger.warn("[ABL-蹇冭烦瓒呮椂] ID锛歿}", mediaServerItem.getId());
+ mediaServerItem.setStatus(false);
+ offlineABLPrimaryMap.put(mediaServerItem.getId(), mediaServerItem);
+ offlineAblTimeMap.put(mediaServerItem.getId(), System.currentTimeMillis());
+ // TODO 鍙戦�佺绾块�氱煡
+ mediaServerService.update(mediaServerItem);
+ }, (int)(mediaServerItem.getHookAliveInterval() * 2 * 1000));
+ }
+ private void initPort(MediaServer mediaServerItem, AblServerConfig ablServerConfig) {
+ // 绔彛鍙細浠庨厤缃腑璇诲彇涓�娆★紝涓�鏃﹁嚜宸遍厤缃垨鑰呰鍙栬繃浜嗗皢涓嶅湪閰嶇疆
+ if (mediaServerItem.getHttpSSlPort() == 0) {
+ mediaServerItem.setHttpSSlPort(ablServerConfig.getHttpSSLport());
+ }
+ if (mediaServerItem.getRtmpPort() == 0) {
+ mediaServerItem.setRtmpPort(ablServerConfig.getRtmpPort());
+ }
+ if (mediaServerItem.getRtmpSSlPort() == 0) {
+ mediaServerItem.setRtmpSSlPort(ablServerConfig.getRtmpSslPort());
+ }
+ if (mediaServerItem.getRtspPort() == 0) {
+ mediaServerItem.setRtspPort(ablServerConfig.getRtspPort());
+ }
+ if (mediaServerItem.getRtspSSLPort() == 0) {
+ mediaServerItem.setRtspSSLPort(ablServerConfig.getRtspSSlport());
+ }
+ if (mediaServerItem.getRtpProxyPort() == 0) {
+ mediaServerItem.setRtpProxyPort(ablServerConfig.getRtpProxyPort());
+ }
+ mediaServerItem.setHookAliveInterval(10F);
+ }
+
+ public void setZLMConfig(MediaServer mediaServerItem, boolean restart) {
+ logger.info("[濯掍綋鏈嶅姟鑺傜偣] 姝ｅ湪璁剧疆 锛歿} -> {}:{}",
+ mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ String protocol = sslEnabled ? "https" : "http";
+ String hookPrefix = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
+
+ Map<String, Object> param = new HashMap<>();
+ param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
+ if (mediaServerItem.getRtspPort() != 0) {
+ param.put("ffmpeg.snap", "%s -rtsp_transport tcp -i %s -y -f mjpeg -frames:v 1 %s");
+ }
+ param.put("hook.enable","1");
+ param.put("hook.on_flow_report","");
+ param.put("hook.on_play",String.format("%s/on_play", hookPrefix));
+ param.put("hook.on_http_access","");
+ param.put("hook.on_publish", String.format("%s/on_publish", hookPrefix));
+ param.put("hook.on_record_ts","");
+ param.put("hook.on_rtsp_auth","");
+ param.put("hook.on_rtsp_realm","");
+ param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrefix));
+ param.put("hook.on_shell_login","");
+ param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrefix));
+ param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrefix));
+ param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrefix));
+ param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrefix));
+ param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrefix));
+ param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrefix));
+ param.put("hook.on_record_mp4",String.format("%s/on_record_mp4", hookPrefix));
+ param.put("hook.timeoutSec","30");
+ param.put("hook.alive_interval", mediaServerItem.getHookAliveInterval());
+ // 鎺ㄦ祦鏂紑鍚庡彲浠ュ湪瓒呮椂鏃堕棿鍐呴噸鏂拌繛鎺ヤ笂缁х画鎺ㄦ祦锛岃繖鏍锋挱鏀惧櫒浼氭帴鐫�鎾斁銆�
+ // 缃�0鍏抽棴姝ょ壒鎬�(鎺ㄦ祦鏂紑浼氬鑷寸珛鍗虫柇寮�鎾斁鍣�)
+ // 姝ゅ弬鏁颁笉搴斿ぇ浜庢挱鏀惧櫒瓒呮椂鏃堕棿
+ // 浼樺寲姝ゆ秷鎭互鏇村揩鐨勬敹鍒版祦娉ㄩ攢浜嬩欢
+ param.put("protocol.continue_push_ms", "3000" );
+ // 鏈�澶氱瓑寰呮湭鍒濆鍖栫殑Track鏃堕棿锛屽崟浣嶆绉掞紝瓒呮椂涔嬪悗浼氬拷鐣ユ湭鍒濆鍖栫殑Track, 璁剧疆姝ら�夐」浼樺寲閭ｄ簺闊抽閿欒鐨勪笉瑙勮寖娴侊紝
+ // 绛墇lm鏀寔缁欐瘡涓猺tpServer璁剧疆鍏抽棴闊抽鐨勬椂鍊欏彲浠ヤ笉璁剧疆姝ら�夐」
+ if (mediaServerItem.isRtpEnable() && !ObjectUtils.isEmpty(mediaServerItem.getRtpPortRange())) {
+ param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-"));
+ }
+
+ if (!ObjectUtils.isEmpty(mediaServerItem.getRecordPath())) {
+ File recordPathFile = new File(mediaServerItem.getRecordPath());
+ param.put("protocol.mp4_save_path", recordPathFile.getParentFile().getPath());
+ param.put("protocol.downloadRoot", recordPathFile.getParentFile().getPath());
+ param.put("record.appName", recordPathFile.getName());
+ }
+
+ JSONObject responseJSON = ablResTfulUtils.setServerConfig(mediaServerItem, param);
+
+ if (responseJSON != null && responseJSON.getInteger("code") == 0) {
+ if (restart) {
+ logger.info("[濯掍綋鏈嶅姟鑺傜偣] 璁剧疆鎴愬姛,寮�濮嬮噸鍚互淇濊瘉閰嶇疆鐢熸晥 {} -> {}:{}",
+ mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ ablResTfulUtils.restartServer(mediaServerItem);
+ }else {
+ logger.info("[濯掍綋鏈嶅姟鑺傜偣] 璁剧疆鎴愬姛 {} -> {}:{}",
+ mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ }
+ }else {
+ logger.info("[濯掍綋鏈嶅姟鑺傜偣] 璁剧疆濯掍綋鏈嶅姟鑺傜偣澶辫触 {} -> {}:{}",
+ mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+ }
+ }
+
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/media/abl/ABLRESTfulUtils.java b/src/main/java/com/genersoft/iot/vmp/media/abl/ABLRESTfulUtils.java
new file mode 100755
index 0000000..ce94324
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/media/abl/ABLRESTfulUtils.java
@@ -0,0 +1,403 @@
+package com.genersoft.iot.vmp.media.abl;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
+import okhttp3.*;
+import okhttp3.logging.HttpLoggingInterceptor;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.SocketTimeoutException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+@Component
+public class ABLRESTfulUtils {
+
+ private final static Logger logger = LoggerFactory.getLogger(ABLRESTfulUtils.class);
+
+ private OkHttpClient client;
+
+ public interface RequestCallback{
+ void run(JSONObject response);
+ }
+
+ private OkHttpClient getClient(){
+ return getClient(null);
+ }
+
+ private OkHttpClient getClient(Integer readTimeOut){
+ if (client == null) {
+ if (readTimeOut == null) {
+ readTimeOut = 10;
+ }
+ OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder();
+ //todo 鏆傛椂鍐欐瓒呮椂鏃堕棿 鍧囦负5s
+ // 璁剧疆杩炴帴瓒呮椂鏃堕棿
+ httpClientBuilder.connectTimeout(8,TimeUnit.SECONDS);
+ // 璁剧疆璇诲彇瓒呮椂鏃堕棿
+ httpClientBuilder.readTimeout(readTimeOut,TimeUnit.SECONDS);
+ // 璁剧疆杩炴帴姹�
+ httpClientBuilder.connectionPool(new ConnectionPool(16, 5, TimeUnit.MINUTES));
+ if (logger.isDebugEnabled()) {
+ HttpLoggingInterceptor logging = new HttpLoggingInterceptor(message -> {
+ logger.debug("http璇锋眰鍙傛暟锛�" + message);
+ });
+ logging.setLevel(HttpLoggingInterceptor.Level.BASIC);
+ // OkHttp閫茶娣诲姞鏀旀埅鍣╨oggingInterceptor
+ httpClientBuilder.addInterceptor(logging);
+ }
+ client = httpClientBuilder.build();
+ }
+ return client;
+
+ }
+
+ public JSONObject sendPost(MediaServer mediaServerItem, String api, Map<String, Object> param, RequestCallback callback) {
+ return sendPost(mediaServerItem, api, param, callback, null);
+ }
+
+
+ public JSONObject sendPost(MediaServer mediaServerItem, String api, Map<String, Object> param, RequestCallback callback, Integer readTimeOut) {
+ OkHttpClient client = getClient(readTimeOut);
+
+ if (mediaServerItem == null) {
+ return null;
+ }
+ String url = String.format("http://%s:%s/index/api/%s", mediaServerItem.getIp(), mediaServerItem.getHttpPort(), api);
+ JSONObject responseJSON = new JSONObject();
+ //-2鑷畾涔夋祦濯掍綋 璋冪敤閿欒鐮�
+ responseJSON.put("code",-2);
+ responseJSON.put("msg","娴佸獟浣撹皟鐢ㄥけ璐�");
+
+ FormBody.Builder builder = new FormBody.Builder();
+ builder.add("secret",mediaServerItem.getSecret());
+ if (param != null && param.keySet().size() > 0) {
+ for (String key : param.keySet()){
+ if (param.get(key) != null) {
+ builder.add(key, param.get(key).toString());
+ }
+ }
+ }
+
+ FormBody body = builder.build();
+
+ Request request = new Request.Builder()
+ .post(body)
+ .url(url)
+ .build();
+ if (callback == null) {
+ try {
+ Response response = client.newCall(request).execute();
+
+ if (response.isSuccessful()) {
+ ResponseBody responseBody = response.body();
+ if (responseBody != null) {
+ String responseStr = responseBody.string();
+ responseJSON = JSON.parseObject(responseStr);
+ }
+ }else {
+ System.out.println( 2222);
+ System.out.println( response.code());
+ response.close();
+ Objects.requireNonNull(response.body()).close();
+ }
+ }catch (IOException e) {
+ logger.error(String.format("[ %s ]璇锋眰澶辫触: %s", url, e.getMessage()));
+
+ if(e instanceof SocketTimeoutException){
+ //璇诲彇瓒呮椂瓒呮椂寮傚父
+ logger.error(String.format("璇诲彇ZLM鏁版嵁瓒呮椂澶辫触: %s, %s", url, e.getMessage()));
+ }
+ if(e instanceof ConnectException){
+ //鍒ゆ柇杩炴帴寮傚父锛屾垜杩欓噷鏄姤Failed to connect to 10.7.5.144
+ logger.error(String.format("杩炴帴ZLM杩炴帴澶辫触: %s, %s", url, e.getMessage()));
+ }
+
+ }catch (Exception e){
+ logger.error(String.format("璁块棶ZLM澶辫触: %s, %s", url, e.getMessage()));
+ }
+ }else {
+ client.newCall(request).enqueue(new Callback(){
+
+ @Override
+ public void onResponse(@NotNull Call call, @NotNull Response response){
+ if (response.isSuccessful()) {
+ try {
+ String responseStr = Objects.requireNonNull(response.body()).string();
+ callback.run(JSON.parseObject(responseStr));
+ } catch (IOException e) {
+ logger.error(String.format("[ %s ]璇锋眰澶辫触: %s", url, e.getMessage()));
+ }
+
+ }else {
+ response.close();
+ Objects.requireNonNull(response.body()).close();
+ }
+ }
+
+ @Override
+ public void onFailure(@NotNull Call call, @NotNull IOException e) {
+ logger.error(String.format("杩炴帴ZLM澶辫触: %s, %s", call.request().toString(), e.getMessage()));
+
+ if(e instanceof SocketTimeoutException){
+ //璇诲彇瓒呮椂瓒呮椂寮傚父
+ logger.error(String.format("璇诲彇ZLM鏁版嵁澶辫触: %s, %s", call.request().toString(), e.getMessage()));
+ }
+ if(e instanceof ConnectException){
+ //鍒ゆ柇杩炴帴寮傚父锛屾垜杩欓噷鏄姤Failed to connect to 10.7.5.144
+ logger.error(String.format("杩炴帴ZLM澶辫触: %s, %s", call.request().toString(), e.getMessage()));
+ }
+ }
+ });
+ }
+
+
+
+ return responseJSON;
+ }
+
+ public void sendGetForImg(MediaServer mediaServerItem, String api, Map<String, Object> params, String targetPath, String fileName) {
+ String url = String.format("http://%s:%s/index/api/%s", mediaServerItem.getIp(), mediaServerItem.getHttpPort(), api);
+ HttpUrl parseUrl = HttpUrl.parse(url);
+ if (parseUrl == null) {
+ return;
+ }
+ HttpUrl.Builder httpBuilder = parseUrl.newBuilder();
+
+ httpBuilder.addQueryParameter("secret", mediaServerItem.getSecret());
+ if (params != null) {
+ for (Map.Entry<String, Object> param : params.entrySet()) {
+ httpBuilder.addQueryParameter(param.getKey(), param.getValue().toString());
+ }
+ }
+
+ Request request = new Request.Builder()
+ .url(httpBuilder.build())
+ .build();
+ logger.info(request.toString());
+ try {
+ OkHttpClient client = getClient();
+ Response response = client.newCall(request).execute();
+ if (response.isSuccessful()) {
+ if (targetPath != null) {
+ File snapFolder = new File(targetPath);
+ if (!snapFolder.exists()) {
+ if (!snapFolder.mkdirs()) {
+ logger.warn("{}璺緞鍒涘缓澶辫触", snapFolder.getAbsolutePath());
+ }
+
+ }
+ File snapFile = new File(targetPath + File.separator + fileName);
+ FileOutputStream outStream = new FileOutputStream(snapFile);
+
+ outStream.write(Objects.requireNonNull(response.body()).bytes());
+ outStream.flush();
+ outStream.close();
+ } else {
+ logger.error(String.format("[ %s ]璇锋眰澶辫触: %s %s", url, response.code(), response.message()));
+ }
+ } else {
+ logger.error(String.format("[ %s ]璇锋眰澶辫触: %s %s", url, response.code(), response.message()));
+ }
+ Objects.requireNonNull(response.body()).close();
+ } catch (ConnectException e) {
+ logger.error(String.format("杩炴帴ZLM澶辫触: %s, %s", e.getCause().getMessage(), e.getMessage()));
+ logger.info("璇锋鏌edia閰嶇疆骞剁‘璁LM宸插惎鍔�...");
+ } catch (IOException e) {
+ logger.error(String.format("[ %s ]璇锋眰澶辫触: %s", url, e.getMessage()));
+ }
+ }
+
+ public JSONObject getMediaList(MediaServer mediaServerItem, String app, String stream, String schema, RequestCallback callback){
+ Map<String, Object> param = new HashMap<>();
+ if (app != null) {
+ param.put("app",app);
+ }
+ if (stream != null) {
+ param.put("stream",stream);
+ }
+ if (schema != null) {
+ param.put("schema",schema);
+ }
+ param.put("vhost","__defaultVhost__");
+ return sendPost(mediaServerItem, "getMediaList",param, callback);
+ }
+
+ public JSONObject getMediaList(MediaServer mediaServerItem, String app, String stream){
+ return getMediaList(mediaServerItem, app, stream,null, null);
+ }
+
+ public JSONObject getMediaList(MediaServer mediaServerItem, RequestCallback callback){
+ return sendPost(mediaServerItem, "getMediaList",null, callback);
+ }
+
+ public JSONObject getMediaInfo(MediaServer mediaServerItem, String app, String schema, String stream){
+ Map<String, Object> param = new HashMap<>();
+ param.put("app",app);
+ param.put("schema",schema);
+ param.put("stream",stream);
+ param.put("vhost","__defaultVhost__");
+ return sendPost(mediaServerItem, "getMediaInfo",param, null);
+ }
+
+ public JSONObject getRtpInfo(MediaServer mediaServerItem, String stream_id){
+ Map<String, Object> param = new HashMap<>();
+ param.put("stream_id",stream_id);
+ return sendPost(mediaServerItem, "getRtpInfo",param, null);
+ }
+
+ public JSONObject addFFmpegSource(MediaServer mediaServerItem, String src_url, String dst_url, Integer timeout_ms,
+ boolean enable_audio, boolean enable_mp4, String ffmpeg_cmd_key){
+ logger.info(src_url);
+ logger.info(dst_url);
+ Map<String, Object> param = new HashMap<>();
+ param.put("src_url", src_url);
+ param.put("dst_url", dst_url);
+ param.put("timeout_ms", timeout_ms);
+ param.put("enable_mp4", enable_mp4);
+ param.put("ffmpeg_cmd_key", ffmpeg_cmd_key);
+ return sendPost(mediaServerItem, "addFFmpegSource",param, null);
+ }
+
+ public JSONObject delFFmpegSource(MediaServer mediaServerItem, String key){
+ Map<String, Object> param = new HashMap<>();
+ param.put("key", key);
+ return sendPost(mediaServerItem, "delFFmpegSource",param, null);
+ }
+
+ public JSONObject delStreamProxy(MediaServer mediaServerItem, String key){
+ Map<String, Object> param = new HashMap<>();
+ param.put("key", key);
+ return sendPost(mediaServerItem, "delStreamProxy",param, null);
+ }
+
+ public JSONObject getMediaServerConfig(MediaServer mediaServerItem){
+ return sendPost(mediaServerItem, "getServerConfig",null, null);
+ }
+
+ public JSONObject setServerConfig(MediaServer mediaServerItem, Map<String, Object> param){
+ return sendPost(mediaServerItem,"setServerConfig",param, null);
+ }
+
+ public JSONObject openRtpServer(MediaServer mediaServerItem, Map<String, Object> param){
+ return sendPost(mediaServerItem, "openRtpServer",param, null);
+ }
+
+ public JSONObject closeRtpServer(MediaServer mediaServerItem, Map<String, Object> param) {
+ return sendPost(mediaServerItem, "closeRtpServer",param, null);
+ }
+
+ public void closeRtpServer(MediaServer mediaServerItem, Map<String, Object> param, RequestCallback callback) {
+ sendPost(mediaServerItem, "closeRtpServer",param, callback);
+ }
+
+ public JSONObject listRtpServer(MediaServer mediaServerItem) {
+ return sendPost(mediaServerItem, "listRtpServer",null, null);
+ }
+
+ public JSONObject startSendRtp(MediaServer mediaServerItem, Map<String, Object> param) {
+ return sendPost(mediaServerItem, "startSendRtp",param, null);
+ }
+
+ public JSONObject startSendRtpPassive(MediaServer mediaServerItem, Map<String, Object> param) {
+ return sendPost(mediaServerItem, "startSendRtpPassive",param, null);
+ }
+
+ public JSONObject startSendRtpPassive(MediaServer mediaServerItem, Map<String, Object> param, RequestCallback callback) {
+ return sendPost(mediaServerItem, "startSendRtpPassive",param, callback);
+ }
+
+ public JSONObject stopSendRtp(MediaServer mediaServerItem, Map<String, Object> param) {
+ return sendPost(mediaServerItem, "stopSendRtp",param, null);
+ }
+
+ public JSONObject restartServer(MediaServer mediaServerItem) {
+ return sendPost(mediaServerItem, "restartServer",null, null);
+ }
+
+ public JSONObject addStreamProxy(MediaServer mediaServerItem, String app, String stream, String url, boolean enable_audio, boolean enable_mp4, String rtp_type) {
+ Map<String, Object> param = new HashMap<>();
+ param.put("vhost", "__defaultVhost__");
+ param.put("app", app);
+ param.put("stream", stream);
+ param.put("url", url);
+ param.put("enable_mp4", enable_mp4?1:0);
+ param.put("enable_audio", enable_audio?1:0);
+ param.put("rtp_type", rtp_type);
+ return sendPost(mediaServerItem, "addStreamProxy",param, null, 20);
+ }
+
+ public JSONObject closeStreams(MediaServer mediaServerItem, String app, String stream) {
+ Map<String, Object> param = new HashMap<>();
+ param.put("vhost", "__defaultVhost__");
+ param.put("app", app);
+ param.put("stream", stream);
+ param.put("force", 1);
+ return sendPost(mediaServerItem, "close_streams",param, null);
+ }
+
+ public JSONObject getAllSession(MediaServer mediaServerItem) {
+ return sendPost(mediaServerItem, "getAllSession",null, null);
+ }
+
+ public void kickSessions(MediaServer mediaServerItem, String localPortSStr) {
+ Map<String, Object> param = new HashMap<>();
+ param.put("local_port", localPortSStr);
+ sendPost(mediaServerItem, "kick_sessions",param, null);
+ }
+
+ public void getSnap(MediaServer mediaServerItem, String streamUrl, int timeout_sec, int expire_sec, String targetPath, String fileName) {
+ Map<String, Object> param = new HashMap<>(3);
+ param.put("url", streamUrl);
+ param.put("timeout_sec", timeout_sec);
+ param.put("expire_sec", expire_sec);
+ sendGetForImg(mediaServerItem, "getSnap", param, targetPath, fileName);
+ }
+
+ public JSONObject pauseRtpCheck(MediaServer mediaServerItem, String streamId) {
+ Map<String, Object> param = new HashMap<>(1);
+ param.put("stream_id", streamId);
+ return sendPost(mediaServerItem, "pauseRtpCheck",param, null);
+ }
+
+ public JSONObject resumeRtpCheck(MediaServer mediaServerItem, String streamId) {
+ Map<String, Object> param = new HashMap<>(1);
+ param.put("stream_id", streamId);
+ return sendPost(mediaServerItem, "resumeRtpCheck",param, null);
+ }
+
+ public JSONObject connectRtpServer(MediaServer mediaServerItem, String dst_url, int dst_port, String stream_id) {
+ Map<String, Object> param = new HashMap<>(1);
+ param.put("dst_url", dst_url);
+ param.put("dst_port", dst_port);
+ param.put("stream_id", stream_id);
+ return sendPost(mediaServerItem, "connectRtpServer",param, null);
+ }
+
+ public JSONObject updateRtpServerSSRC(MediaServer mediaServerItem, String streamId, String ssrc) {
+ Map<String, Object> param = new HashMap<>(1);
+ param.put("ssrc", ssrc);
+ param.put("stream_id", streamId);
+ return sendPost(mediaServerItem, "updateRtpServerSSRC",param, null);
+ }
+
+ public JSONObject deleteRecordDirectory(MediaServer mediaServerItem, String app, String stream, String date, String fileName) {
+ Map<String, Object> param = new HashMap<>(1);
+ param.put("vhost", "__defaultVhost__");
+ param.put("app", app);
+ param.put("stream", stream);
+ param.put("period", date);
+ param.put("name", fileName);
+ return sendPost(mediaServerItem, "deleteRecordDirectory",param, null);
+ }
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/media/abl/bean/AblServerConfig.java b/src/main/java/com/genersoft/iot/vmp/media/abl/bean/AblServerConfig.java
new file mode 100644
index 0000000..4898326
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/media/abl/bean/AblServerConfig.java
@@ -0,0 +1,4 @@
+package com.genersoft.iot.vmp.media.abl.bean;
+
+public class AblServerConfig {
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/media/abl/event/HookAblServerKeepaliveEvent.java b/src/main/java/com/genersoft/iot/vmp/media/abl/event/HookAblServerKeepaliveEvent.java
new file mode 100644
index 0000000..bbedd6d
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/media/abl/event/HookAblServerKeepaliveEvent.java
@@ -0,0 +1,24 @@
+package com.genersoft.iot.vmp.media.abl.event;
+
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * zlm 蹇冭烦浜嬩欢
+ */
+public class HookAblServerKeepaliveEvent extends ApplicationEvent {
+
+ public HookAblServerKeepaliveEvent(Object source) {
+ super(source);
+ }
+
+ private MediaServer mediaServerItem;
+
+ public MediaServer getMediaServerItem() {
+ return mediaServerItem;
+ }
+
+ public void setMediaServerItem(MediaServer mediaServerItem) {
+ this.mediaServerItem = mediaServerItem;
+ }
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/media/abl/event/HookAblServerStartEvent.java b/src/main/java/com/genersoft/iot/vmp/media/abl/event/HookAblServerStartEvent.java
new file mode 100644
index 0000000..4e1960e
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/media/abl/event/HookAblServerStartEvent.java
@@ -0,0 +1,24 @@
+package com.genersoft.iot.vmp.media.abl.event;
+
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * zlm server_start浜嬩欢
+ */
+public class HookAblServerStartEvent extends ApplicationEvent {
+
+ public HookAblServerStartEvent(Object source) {
+ super(source);
+ }
+
+ private MediaServer mediaServerItem;
+
+ public MediaServer getMediaServerItem() {
+ return mediaServerItem;
+ }
+
+ public void setMediaServerItem(MediaServer mediaServerItem) {
+ this.mediaServerItem = mediaServerItem;
+ }
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
index 01fc635..a247a78 100755
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
@@ -142,14 +142,6 @@
@PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) {
- taskExecutor.execute(() -> {
- List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive);
- if (subscribes != null && !subscribes.isEmpty()) {
- for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {
- subscribe.response(null, param);
- }
- }
- });
try {
HookZlmServerKeepaliveEvent event = new HookZlmServerKeepaliveEvent(this);
MediaServer mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
@@ -184,6 +176,7 @@
}
}
});
+ // TODO 姝ゅ閫昏緫閫傚悎杩佺Щ鍒癕ediaService涓�
if (!"rtp".equals(param.getApp())) {
Map<String, String> paramMap = urlParamToMap(param.getParams());
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
@@ -208,8 +201,8 @@
// TODO 鍔犲揩澶勭悊閫熷害
String mediaServerId = json.getString("mediaServerId");
- MediaServer mediaInfo = mediaServerService.getOne(mediaServerId);
- if (mediaInfo == null) {
+ MediaServer mediaServer = mediaServerService.getOne(mediaServerId);
+ if (mediaServer == null) {
return new HookResultForOnPublish(200, "success");
}
// 鎺ㄦ祦閴存潈鐨勫鐞�
@@ -257,7 +250,7 @@
taskExecutor.execute(() -> {
ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json);
if (subscribe != null) {
- subscribe.response(mediaInfo, param);
+ subscribe.response(mediaServer, param);
}
});
@@ -273,7 +266,7 @@
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream());
// 鍗曠鍙ｆā寮忎笅淇敼娴� ID
- if (!mediaInfo.isRtpEnable() && inviteInfo == null) {
+ if (!mediaServer.isRtpEnable() && inviteInfo == null) {
String ssrc = String.format("%010d", Long.parseLong(param.getStream(), 16));
inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
if (inviteInfo != null) {
--
Gitblit v1.8.0