From c7d15150237c946fbb4f5547dab0018a5f573d88 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期二, 27 六月 2023 15:34:38 +0800 Subject: [PATCH] 增加redis通道 --- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 127 +++++++++++++++++++++++------------------ 1 files changed, 71 insertions(+), 56 deletions(-) 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 ca5bb13..9ffb524 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -7,6 +7,7 @@ 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.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; @@ -116,7 +117,7 @@ @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8") public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) { - logger.info("[ZLM HOOK] 鏀跺埌zlm蹇冭烦锛�" + param.getMediaServerId()); +// logger.info("[ZLM HOOK] 鏀跺埌zlm蹇冭烦锛�" + param.getMediaServerId()); taskExecutor.execute(() -> { List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive); @@ -281,7 +282,7 @@ @ResponseBody @PostMapping(value = "/on_stream_changed", produces = "application/json;charset=UTF-8") public HookResult onStreamChanged(@RequestBody OnStreamChangedHookParam param) { - + System.out.println(JSON.toJSONString(param)); if (param.isRegist()) { logger.info("[ZLM HOOK] 娴佹敞鍐�, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream()); } else { @@ -292,22 +293,22 @@ JSONObject json = (JSONObject) JSON.toJSON(param); taskExecutor.execute(() -> { ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json); + MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId()); + if (mediaInfo == null) { + logger.info("[ZLM HOOK] 娴佸彉鍖栨湭鎵惧埌ZLM, {}", param.getMediaServerId()); + return; + } if (subscribe != null) { - MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId()); - if (mediaInfo != null) { - subscribe.response(mediaInfo, json); - } + subscribe.response(mediaInfo, json); } List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks(); // 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); @@ -321,7 +322,7 @@ redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream()); } - if ("rtsp".equals(param.getSchema())) { + if ("rtmp".equals(param.getSchema())) { // 鏇存柊娴佸獟浣撹礋杞戒俊鎭� if (param.isRegist()) { mediaServerService.addCount(param.getMediaServerId()); @@ -329,7 +330,10 @@ mediaServerService.removeCount(param.getMediaServerId()); } // 璁剧疆鎷夋祦浠g悊涓婄嚎/绂荤嚎 - streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream()); + int updateStatusResult = streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream()); + if (updateStatusResult > 0) { + + } if ("rtp".equals(param.getApp()) && !param.isRegist()) { StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream()); @@ -337,7 +341,8 @@ redisCatchStorage.stopPlay(streamInfo); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); } else { - streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null); + streamInfo = redisCatchStorage.queryPlayback(null, null, + param.getStream(), null); if (streamInfo != null) { redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream(), null); @@ -346,48 +351,52 @@ } else { if (!"rtp".equals(param.getApp())) { String type = OriginType.values()[param.getOriginType()].getType(); - MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId()); + 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(), tracks, callId); + param.setStreamInfo(new StreamContent(streamInfoByAppAndStream)); + // 濡傛灉鏄媺娴佷唬鐞嗕骇鐢熺殑锛屼笉闇�瑕佸啓鍏ユ帹娴� - if (mediaServerItem != null) { - if (param.isRegist()) { - StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream()); - String callId = null; - if (streamAuthorityInfo != null) { - callId = streamAuthorityInfo.getCallId(); - } - StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem, - param.getApp(), param.getStream(), tracks, callId); - param.setStreamInfo(new StreamContent(streamInfoByAppAndStream)); - redisCatchStorage.addStream(mediaServerItem, 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); - } - } else { - // 鍏煎娴佹敞閿�鏃剁被鍨嬩粠redis璁板綍鑾峰彇 - OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(param.getApp(), param.getStream(), param.getMediaServerId()); - if (onStreamChangedHookParam != null) { - type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType(); - redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream()); - } - GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream()); - if (gbStream != null) { + 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); + } + } 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()); + } + GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream()); + if (gbStream != null) { // eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF); - } - zlmMediaListManager.removeMedia(param.getApp(), param.getStream()); } - 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); - } + zlmMediaListManager.removeMedia(param.getApp(), param.getStream()); + } + GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream()); + if (gbStream != null) { + 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); } } } @@ -395,7 +404,7 @@ List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(param.getStream()); if (sendRtpItems.size() > 0) { for (SendRtpItem sendRtpItem : sendRtpItems) { - if (sendRtpItem.getApp().equals(param.getApp())) { + if (sendRtpItem != null && sendRtpItem.getApp().equals(param.getApp())) { String platformId = sendRtpItem.getPlatformId(); ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId); Device device = deviceService.getDevice(platformId); @@ -403,6 +412,8 @@ try { if (platform != null) { commanderFroPlatform.streamByeCmd(platform, sendRtpItem); + redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(), + sendRtpItem.getCallId(), sendRtpItem.getStreamId()); } else { cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId()); } @@ -427,7 +438,8 @@ @PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8") public JSONObject onStreamNoneReader(@RequestBody OnStreamNoneReaderHookParam param) { - logger.info("[ZLM HOOK]娴佹棤浜鸿鐪嬶細{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream()); + logger.info("[ZLM HOOK]娴佹棤浜鸿鐪嬶細{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), + param.getApp(), param.getStream()); JSONObject ret = new JSONObject(); ret.put("code", 0); // 鍥芥爣绫诲瀷鐨勬祦 @@ -439,7 +451,8 @@ if (streamInfoForPlayCatch != null) { // 鏀跺埌鏃犱汉瑙傜湅璇存槑娴佷篃娌℃湁鍦ㄥ線涓婄骇鎺ㄩ�� if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) { - List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(streamInfoForPlayCatch.getChannelId()); + List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId( + streamInfoForPlayCatch.getChannelId()); if (sendRtpItems.size() > 0) { for (SendRtpItem sendRtpItem : sendRtpItems) { ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); @@ -469,7 +482,8 @@ return ret; } // 褰曞儚鍥炴斁 - StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, param.getStream(), null); + StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, + param.getStream(), null); if (streamInfoForPlayBackCatch != null) { if (streamInfoForPlayBackCatch.isPause()) { ret.put("close", false); @@ -490,7 +504,8 @@ return ret; } // 褰曞儚涓嬭浇 - StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, param.getStream(), null); + StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, + param.getStream(), null); // 杩涜褰曞儚涓嬭浇鏃舵棤浜鸿鐪嬩笉鏂祦 if (streamInfoForDownload != null) { ret.put("close", false); @@ -518,7 +533,7 @@ } return ret; } - // 鎺ㄦ祦鍏锋湁涓诲姩鎬э紝鏆傛椂涓嶅仛澶勭悊 + // TODO 鎺ㄦ祦鍏锋湁涓诲姩鎬э紝鏆傛椂涓嶅仛澶勭悊 // StreamPushItem streamPushItem = streamPushService.getPush(app, streamId); // if (streamPushItem != null) { // // TODO 鍙戦�佸仠姝� -- Gitblit v1.8.0