From 39c3274294bce4f9dd4a231d2a74ed379af8ee34 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期四, 25 四月 2024 17:53:24 +0800 Subject: [PATCH] 合并分支 --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java | 47 ++-- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java | 348 +--------------------------------- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaNodeServerService.java | 6 src/main/java/com/genersoft/iot/vmp/conf/redis/RedisRpcConfig.java | 4 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java | 32 +- src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcController.java | 96 ++++----- src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java | 5 src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java | 12 + src/main/java/com/genersoft/iot/vmp/media/service/IMediaServerService.java | 2 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java | 2 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java | 2 src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java | 3 12 files changed, 115 insertions(+), 444 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisRpcConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisRpcConfig.java index 661e370..3fcc40c 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisRpcConfig.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisRpcConfig.java @@ -6,7 +6,6 @@ import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcMessage; import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcRequest; import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcResponse; -import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.service.redisMsg.control.RedisRpcController; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -44,9 +43,6 @@ @Autowired private RedisTemplate<Object, Object> redisTemplate; - - @Autowired - private ZlmHttpHookSubscribe hookSubscribe; private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>(); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java index 614d49e..162d232 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java @@ -15,9 +15,7 @@ import com.genersoft.iot.vmp.media.bean.MediaInfo; import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.service.IMediaServerService; -import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.service.*; -import com.genersoft.iot.vmp.service.bean.RequestStopPushStreamMsg; import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; @@ -121,22 +119,20 @@ logger.info("[鏀跺埌bye] 鍋滄鎺ㄦ祦锛歿}, 濯掍綋鑺傜偣锛� {}", streamId, sendRtpItem.getMediaServerId()); if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) { - // 鏌ヨ杩欒矾娴佹槸鍚︽槸鏈钩鍙扮殑 - StreamPushItem push = pushService.getPush(sendRtpItem.getApp(), sendRtpItem.getStream()); - if (push!= null && !push.isSelf()) { - // 涓嶆槸鏈钩鍙扮殑灏卞彂閫乺edis娑堟伅璁╁叾浠杦vp鍋滄鍙戞祦 - ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId()); - if (platform != null) { - RequestStopPushStreamMsg streamMsg = RequestStopPushStreamMsg.getInstance(sendRtpItem, platform.getName(), platform.getId()); - redisGbPlayMsgListener.sendMsgForStopSendRtpStream(sendRtpItem.getServerId(), streamMsg); - } - }else { - MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), - callIdHeader.getCallId(), null); - mediaServerService.stopSendRtp(mediaInfo, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc()); - if (userSetting.getUseCustomSsrcForParentInvite()) { - mediaServerService.releaseSsrc(mediaInfo.getId(), sendRtpItem.getSsrc()); + // 涓嶆槸鏈钩鍙扮殑灏卞彂閫乺edis娑堟伅璁╁叾浠杦vp鍋滄鍙戞祦 + ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId()); + if (platform != null) { + redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, platform); + if (!userSetting.getServerId().equals(sendRtpItem.getServerId())) { + redisRpcService.stopSendRtp(sendRtpItem.getRedisKey()); + redisCatchStorage.deleteSendRTPServer(null, null, sendRtpItem.getCallId(), null); + }else { + MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + redisCatchStorage.deleteSendRTPServer(null, null, callIdHeader.getCallId(), null); + mediaServerService.stopSendRtp(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc()); + if (userSetting.getUseCustomSsrcForParentInvite()) { + mediaServerService.releaseSsrc(mediaServer.getId(), sendRtpItem.getSsrc()); + } } }else { logger.info("[涓婄骇骞冲彴鍋滄瑙傜湅] 鏈壘鍒板钩鍙皗}鐨勪俊鎭紝鍙戦�乺edis娑堟伅澶辫触", sendRtpItem.getPlatformId()); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index f1f277f..e5c18ee 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -19,12 +19,12 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor; import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; import com.genersoft.iot.vmp.gb28181.utils.SipUtils; +import com.genersoft.iot.vmp.media.bean.MediaInfo; import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.event.hook.Hook; import com.genersoft.iot.vmp.media.event.hook.HookSubscribe; import com.genersoft.iot.vmp.media.event.hook.HookType; import com.genersoft.iot.vmp.media.service.IMediaServerService; -import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem; import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.service.IInviteStreamService; @@ -34,7 +34,6 @@ import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService; import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory; -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.OnStreamChangedHookParam; import com.genersoft.iot.vmp.service.*; @@ -491,7 +490,7 @@ String startTimeStr = DateUtil.urlFormatter.format(start); String endTimeStr = DateUtil.urlFormatter.format(end); String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr; - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0,false,!channel.isHasAudio(), false, device.getStreamModeForParam()); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0,false,!channel.getHasAudio(), false, device.getStreamModeForParam()); sendRtpItem.setStream(stream); // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -521,7 +520,7 @@ } sendRtpItem.setPlayType(InviteStreamType.DOWNLOAD); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,!channel.isHasAudio(), false, device.getStreamModeForParam()); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,!channel.getHasAudio(), false, device.getStreamModeForParam()); sendRtpItem.setStream(ssrcInfo.getStream()); // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -596,11 +595,11 @@ // 浠巖edis鏌ヨ鏄惁姝e湪鎺ユ敹杩欎釜鎺ㄦ祦 StreamPushItem pushListItem = redisCatchStorage.getPushListItem(gbStream.getApp(), gbStream.getStream()); if (pushListItem != null) { - sendRtpItem.setServerId(pushListItem.getSeverId()); + sendRtpItem.setServerId(pushListItem.getServerId()); sendRtpItem.setMediaServerId(pushListItem.getMediaServerId()); StreamPushItem transform = streamPushService.transform(pushListItem); - transform.setSelf(userSetting.getServerId().equals(pushListItem.getSeverId())); + transform.setSelf(userSetting.getServerId().equals(pushListItem.getServerId())); redisCatchStorage.updateSendRTPSever(sendRtpItem); // 寮�濮嬫帹娴� sendPushStream(sendRtpItem, mediaServerItem, platform, request); @@ -657,9 +656,10 @@ /** * 瀹夋帓鎺ㄦ祦 */ - private void sendProxyStream(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) { - Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); - if (streamReady != null && streamReady) { + private void sendProxyStream(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) { + MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); + + if (mediaInfo != null) { // 鑷钩鍙板唴瀹� int localPort = sendRtpPortManager.getNextPort(mediaServerItem); @@ -677,7 +677,7 @@ sendRtpItem.setStatus(1); sendRtpItem.setLocalIp(mediaServerItem.getSdpIp()); - SIPResponse response = sendStreamAck(mediaServer, request, sendRtpItem, platform, evt); + SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform); if (response != null) { sendRtpItem.setToTag(response.getToTag()); } @@ -685,11 +685,11 @@ } } - private void sendPushStream(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) { + private void sendPushStream(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) { // 鎺ㄦ祦 if (sendRtpItem.getServerId().equals(userSetting.getServerId())) { - Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); - if (streamReady != null && streamReady) { + MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); + if (mediaInfo != null ) { // 鑷钩鍙板唴瀹� int localPort = sendRtpPortManager.getNextPort(mediaServerItem); if (localPort == 0) { @@ -703,7 +703,7 @@ } // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� sendRtpItem.setStatus(1); - SIPResponse response = sendStreamAck(request, sendRtpItem, platform); + SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform); if (response != null) { sendRtpItem.setToTag(response.getToTag()); } @@ -726,20 +726,19 @@ /** * 閫氱煡娴佷笂绾� */ - private void notifyProxyStreamOnline(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) { + private void notifyProxyStreamOnline(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) { // TODO 鎺у埗鍚敤浠ヤ娇璁惧涓婄嚎 logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍚敤娴佸悗寮�濮嬫帹娴�", sendRtpItem.getApp(), sendRtpItem.getStream()); // 鐩戝惉娴佷笂绾� - HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", mediaServerItem.getId()); - zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, hookParam) -> { - OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam; - logger.info("[涓婄骇鐐规挱]鎷夋祦浠g悊宸茬粡灏辩华锛� {}/{}", streamChangedHookParam.getApp(), streamChangedHookParam.getStream()); + Hook hook = Hook.getInstance(HookType.on_media_arrival, sendRtpItem.getApp(), sendRtpItem.getStream(), mediaServerItem.getId()); + hookSubscribe.addSubscribe(hook, (hookData)->{ + logger.info("[涓婄骇鐐规挱]鎷夋祦浠g悊宸茬粡灏辩华锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); dynamicTask.stop(sendRtpItem.getCallId()); sendProxyStream(sendRtpItem, mediaServerItem, platform, request); }); dynamicTask.startDelay(sendRtpItem.getCallId(), () -> { logger.info("[ app={}, stream={} ] 绛夊緟鎷夋祦浠g悊娴佽秴鏃�", sendRtpItem.getApp(), sendRtpItem.getStream()); - zlmHttpHookSubscribe.removeSubscribe(hookSubscribe); + hookSubscribe.removeSubscribe(hook); }, userSetting.getPlatformPlayTimeout()); boolean start = streamProxyService.start(sendRtpItem.getApp(), sendRtpItem.getStream()); if (!start) { @@ -748,7 +747,7 @@ } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 invite 閫氶亾鏈帹娴�: {}", e.getMessage()); } - zlmHttpHookSubscribe.removeSubscribe(hookSubscribe); + hookSubscribe.removeSubscribe(hook); dynamicTask.stop(sendRtpItem.getCallId()); } } @@ -756,7 +755,7 @@ /** * 閫氱煡娴佷笂绾� */ - private void notifyPushStreamOnline(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) { + private void notifyPushStreamOnline(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) { // 鍙戦�乺edis娑堟伅浠ヤ娇璁惧涓婄嚎锛屾祦涓婄嚎鍚庤 logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍙戦�乺edis淇℃伅鎺у埗璁惧寮�濮嬫帹娴�", sendRtpItem.getApp(), sendRtpItem.getStream()); MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1, @@ -860,7 +859,7 @@ redisCatchStorage.updateSendRTPSever(sendRtpItem); } - public SIPResponse sendStreamAck(MediaServerItem mediaServerItem, SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform, RequestEvent evt) { + public SIPResponse sendStreamAck(MediaServer mediaServerItem, SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform) { String sdpIp = sendRtpItem.getLocalIp(); if (!ObjectUtils.isEmpty(platform.getSendStreamIp())) { @@ -1006,7 +1005,7 @@ logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}, {}", requesterId, addressStr, port, gb28181Sdp.getSsrc(), mediaTransmissionTCP ? (tcpActive ? "TCP涓诲姩" : "TCP琚姩") : "UDP"); - MediaServer mediaServerItem = broadcastCatch.getMediaServerItem(); + MediaServer mediaServerItem = broadcastCatch.getMediaServer(); if (mediaServerItem == null) { logger.warn("鏈壘鍒拌闊冲枈璇濅娇鐢ㄧ殑zlm"); try { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java index 9f6acc9..a8c4af8 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestForMobilePositionProcessor.java @@ -166,7 +166,7 @@ }catch (Exception e) { logger.error("[鍚戜笂绾ц浆鍙戠Щ鍔ㄤ綅缃け璐 ", e); } - if (mobilePosition.getChannelId().equals(mobilePosition.getDeviceId()) || mobilePosition.getChannelId() == null) { + if (mobilePosition.getChannelId() == null || mobilePosition.getChannelId().equals(mobilePosition.getDeviceId())) { List<DeviceChannel> channels = deviceChannelService.queryChaneListByDeviceId(mobilePosition.getDeviceId()); channels.forEach(channel -> { // 鍙戦�乺edis娑堟伅銆� 閫氱煡浣嶇疆淇℃伅鐨勫彉鍖� diff --git a/src/main/java/com/genersoft/iot/vmp/media/service/IMediaServerService.java b/src/main/java/com/genersoft/iot/vmp/media/service/IMediaServerService.java index 671790a..11d5bf5 100755 --- a/src/main/java/com/genersoft/iot/vmp/media/service/IMediaServerService.java +++ b/src/main/java/com/genersoft/iot/vmp/media/service/IMediaServerService.java @@ -147,4 +147,6 @@ SendRtpItem createSendRtpItem(MediaServer serverItem, String ip, int port, String ssrc, String platformId, String app, String stream, String channelId, boolean tcp, boolean rtcp); + + MediaServer getMediaServerByAppAndStream(String app, String stream); } diff --git a/src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java index 660fe0d..e7e79a8 100755 --- a/src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java @@ -898,4 +898,16 @@ sendRtpItem.setRtcp(rtcp); return sendRtpItem; } + + @Override + public MediaServer getMediaServerByAppAndStream(String app, String stream) { + List<MediaServer> mediaServerList = getAll(); + for (MediaServer mediaServer : mediaServerList) { + MediaInfo mediaInfo = getMediaInfo(mediaServer, app, stream); + if (mediaInfo != null) { + return mediaServer; + } + } + return null; + } } 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 030dd7d..0e46e68 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 @@ -5,25 +5,22 @@ import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager; -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.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; +import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.ResultForOnPublish; import com.genersoft.iot.vmp.media.event.hook.HookSubscribe; import com.genersoft.iot.vmp.media.event.media.*; import com.genersoft.iot.vmp.media.event.mediaServer.MediaSendRtpStoppedEvent; import com.genersoft.iot.vmp.media.service.IMediaServerService; -import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.zlm.dto.ZLMServerConfig; import com.genersoft.iot.vmp.media.zlm.dto.hook.*; import com.genersoft.iot.vmp.media.zlm.event.HookZlmServerKeepaliveEvent; import com.genersoft.iot.vmp.media.zlm.event.HookZlmServerStartEvent; import com.genersoft.iot.vmp.service.*; -import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; @@ -180,54 +177,6 @@ if (mediaServer == null) { return new HookResultForOnPublish(0, "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); - } - }); ResultForOnPublish resultForOnPublish = mediaService.authenticatePublish(mediaServer, param.getApp(), param.getStream(), param.getParams()); if (resultForOnPublish != null) { @@ -264,220 +213,6 @@ applicationEventPublisher.publishEvent(mediaDepartureEvent); } - 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) { - subscribe.response(mediaInfo, param); - } - - 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); - } 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(), tracks, 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) { - continue; - } - - if (sendRtpItem.getApp().equals(param.getApp())) { - logger.info(sendRtpItem.toString()); - if (userSetting.getServerId().equals(sendRtpItem.getServerId())) { - MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0, - sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(), - sendRtpItem.getPlatformId(), null, userSetting.getServerId(), param.getMediaServerId()); - // 閫氱煡鍏朵粬wvp鍋滄鍙戞祦 - redisCatchStorage.sendPushStreamClose(messageForPushChannel); - }else { - 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(); } @@ -491,63 +226,6 @@ logger.info("[ZLM 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()); - } - boolean close = mediaService.closeStreamOnNoneReader(param.getMediaServerId(), param.getApp(), param.getStream(), param.getSchema()); ret.put("code", close); return ret; @@ -609,22 +287,16 @@ 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()); - } + try { + MediaSendRtpStoppedEvent event = new MediaSendRtpStoppedEvent(this); + MediaServer mediaServerItem = mediaServerService.getOne(param.getMediaServerId()); + if (mediaServerItem != null) { + event.setMediaServer(mediaServerItem); + applicationEventPublisher.publishEvent(event); } - }); + }catch (Exception e) { + logger.info("[ZLM-HOOK-rtp鍙戦�佸叧闂璢 鍙戦�侀�氱煡澶辫触 ", e); + } return HookResult.SUCCESS(); } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaNodeServerService.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaNodeServerService.java index ca1eb52..74a6fce 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaNodeServerService.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaNodeServerService.java @@ -131,7 +131,11 @@ param.put("ssrc", ssrc); } JSONObject jsonObject = zlmresTfulUtils.stopSendRtp(mediaInfo, param); - return (jsonObject != null && jsonObject.getInteger("code") == 0); + if (jsonObject == null || jsonObject.getInteger("code") != 0 ) { + logger.error("鍋滄鍙戞祦澶辫触: {}, 鍙傛暟锛歿}", jsonObject.getString("msg"), JSON.toJSONString(param)); + throw new ControllerException(jsonObject.getInteger("code"), jsonObject.getString("msg")); + } + return true; } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java index 7b0bf39..4191fe5 100755 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java @@ -265,7 +265,7 @@ return result; } - public JSONObject stopSendRtpStream(MediaServerItem mediaServerItem, SendRtpItem sendRtpItem) { + public JSONObject stopSendRtpStream(MediaServer mediaServerItem, SendRtpItem sendRtpItem) { Map<String, Object> param = new HashMap<>(); param.put("vhost", "__defaultVhost__"); param.put("app", sendRtpItem.getApp()); diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcController.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcController.java index af79204..4e12670 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcController.java +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcController.java @@ -2,6 +2,7 @@ import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.conf.UserSetting; +import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.conf.redis.RedisRpcConfig; import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcMessage; import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcRequest; @@ -10,14 +11,13 @@ import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; +import com.genersoft.iot.vmp.media.bean.MediaInfo; +import com.genersoft.iot.vmp.media.bean.MediaServer; +import com.genersoft.iot.vmp.media.event.hook.Hook; +import com.genersoft.iot.vmp.media.event.hook.HookSubscribe; +import com.genersoft.iot.vmp.media.event.hook.HookType; +import com.genersoft.iot.vmp.media.service.IMediaServerService; import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; -import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory; -import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; -import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; -import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; -import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam; -import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; @@ -56,11 +56,7 @@ private UserSetting userSetting; @Autowired - private ZlmHttpHookSubscribe hookSubscribe; - - @Autowired - private ZLMServerFactory zlmServerFactory; - + private HookSubscribe hookSubscribe; @Autowired private RedisTemplate<Object, Object> redisTemplate; @@ -88,7 +84,7 @@ } logger.info("[redis-rpc] 鑾峰彇鍙戞祦鐨勪俊鎭細 {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); // 鏌ヨ鏈骇鏄惁鏈夎繖涓祦 - MediaServerItem mediaServerItem = mediaServerService.getMediaServerByAppAndStream(sendRtpItem.getApp(), sendRtpItem.getStream()); + MediaServer mediaServerItem = mediaServerService.getMediaServerByAppAndStream(sendRtpItem.getApp(), sendRtpItem.getStream()); if (mediaServerItem == null) { RedisRpcResponse response = request.getResponse(); response.setStatusCode(200); @@ -124,17 +120,17 @@ SendRtpItem sendRtpItem = JSONObject.parseObject(request.getParam().toString(), SendRtpItem.class); logger.info("[redis-rpc] 鐩戝惉娴佷笂绾匡細 {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); // 鏌ヨ鏈骇鏄惁鏈夎繖涓祦 - MediaServerItem mediaServerItem = mediaServerService.getMediaServerByAppAndStream(sendRtpItem.getApp(), sendRtpItem.getStream()); - if (mediaServerItem != null) { + MediaServer mediaServer = mediaServerService.getMediaServerByAppAndStream(sendRtpItem.getApp(), sendRtpItem.getStream()); + if (mediaServer != null) { logger.info("[redis-rpc] 鐩戝惉娴佷笂绾挎椂鍙戠幇娴佸凡瀛樺湪鐩存帴杩斿洖锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); // 璇诲彇redis涓殑涓婄骇鐐规挱淇℃伅锛岀敓鎴恠endRtpItm鍙戦�佸嚭鍘� if (sendRtpItem.getSsrc() == null) { // 涓婄骇骞冲彴鐐规挱鏃朵笉浣跨敤涓婄骇骞冲彴鎸囧畾鐨剆src锛屼娇鐢ㄨ嚜瀹氫箟鐨剆src锛屽弬鑰冨浗鏍囨枃妗�-鐐规挱澶栧煙璁惧濯掍綋娴丼SRC澶勭悊鏂瑰紡 - String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(mediaServerItem.getId()) : ssrcFactory.getPlayBackSsrc(mediaServerItem.getId()); + String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(mediaServer.getId()) : ssrcFactory.getPlayBackSsrc(mediaServer.getId()); sendRtpItem.setSsrc(ssrc); } - sendRtpItem.setMediaServerId(mediaServerItem.getId()); - sendRtpItem.setLocalIp(mediaServerItem.getSdpIp()); + sendRtpItem.setMediaServerId(mediaServer.getId()); + sendRtpItem.setLocalIp(mediaServer.getSdpIp()); sendRtpItem.setServerId(userSetting.getServerId()); redisTemplate.opsForValue().set(sendRtpItem.getRedisKey(), sendRtpItem); @@ -143,19 +139,17 @@ response.setStatusCode(200); } // 鐩戝惉娴佷笂绾裤�� 娴佷笂绾跨洿鎺ュ彂閫乻endRtpItem娑堟伅缁欏疄闄呯殑淇′护澶勭悊鑰� - HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed( - sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null); - - hookSubscribe.addSubscribe(hook, (MediaServerItem mediaServerItemInUse, HookParam hookParam) -> { + Hook hook = Hook.getInstance(HookType.on_media_arrival, sendRtpItem.getApp(), sendRtpItem.getStream(), null); + hookSubscribe.addSubscribe(hook, (hookData) -> { logger.info("[redis-rpc] 鐩戝惉娴佷笂绾匡紝娴佸凡涓婄嚎锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); // 璇诲彇redis涓殑涓婄骇鐐规挱淇℃伅锛岀敓鎴恠endRtpItm鍙戦�佸嚭鍘� if (sendRtpItem.getSsrc() == null) { // 涓婄骇骞冲彴鐐规挱鏃朵笉浣跨敤涓婄骇骞冲彴鎸囧畾鐨剆src锛屼娇鐢ㄨ嚜瀹氫箟鐨剆src锛屽弬鑰冨浗鏍囨枃妗�-鐐规挱澶栧煙璁惧濯掍綋娴丼SRC澶勭悊鏂瑰紡 - String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(mediaServerItemInUse.getId()) : ssrcFactory.getPlayBackSsrc(mediaServerItemInUse.getId()); + String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(hookData.getMediaServer().getId()) : ssrcFactory.getPlayBackSsrc(hookData.getMediaServer().getId()); sendRtpItem.setSsrc(ssrc); } - sendRtpItem.setMediaServerId(mediaServerItemInUse.getId()); - sendRtpItem.setLocalIp(mediaServerItemInUse.getSdpIp()); + sendRtpItem.setMediaServerId(hookData.getMediaServer().getId()); + sendRtpItem.setLocalIp(hookData.getMediaServer().getSdpIp()); sendRtpItem.setServerId(userSetting.getServerId()); redisTemplate.opsForValue().set(sendRtpItem.getRedisKey(), sendRtpItem); @@ -177,8 +171,7 @@ SendRtpItem sendRtpItem = JSONObject.parseObject(request.getParam().toString(), SendRtpItem.class); logger.info("[redis-rpc] 鍋滄鐩戝惉娴佷笂绾匡細 {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); // 鐩戝惉娴佷笂绾裤�� 娴佷笂绾跨洿鎺ュ彂閫乻endRtpItem娑堟伅缁欏疄闄呯殑淇′护澶勭悊鑰� - HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed( - sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null); + Hook hook = Hook.getInstance(HookType.on_media_arrival, sendRtpItem.getApp(), sendRtpItem.getStream(), null); hookSubscribe.removeSubscribe(hook); RedisRpcResponse response = request.getResponse(); response.setStatusCode(200); @@ -201,31 +194,31 @@ return response; } logger.info("[redis-rpc] 寮�濮嬪彂娴侊細 {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); - MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - if (mediaServerItem == null) { + MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + if (mediaServer == null) { logger.info("[redis-rpc] startSendRtp->鏈壘鍒癕ediaServer锛� {}", sendRtpItem.getMediaServerId() ); WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "鏈壘鍒癕ediaServer"); response.setBody(wvpResult); return response; } - - Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); - if (!streamReady) { + MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream()); + if (mediaInfo != null) { logger.info("[redis-rpc] startSendRtp->娴佷笉鍦ㄧ嚎锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "娴佷笉鍦ㄧ嚎"); response.setBody(wvpResult); return response; } - JSONObject jsonObject = zlmServerFactory.startSendRtp(mediaServerItem, sendRtpItem); - if (jsonObject.getInteger("code") == 0) { - logger.info("[redis-rpc] 鍙戞祦鎴愬姛锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); - WVPResult wvpResult = WVPResult.success(); + try { + mediaServerService.startSendRtp(mediaServer, null, sendRtpItem); + }catch (ControllerException exception) { + logger.info("[redis-rpc] 鍙戞祦澶辫触锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}锛� {}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort(), exception.getMsg()); + WVPResult wvpResult = WVPResult.fail(exception.getCode(), exception.getMsg()); response.setBody(wvpResult); - }else { - logger.info("[redis-rpc] 鍙戞祦澶辫触锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}锛� {}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort(), jsonObject); - WVPResult wvpResult = WVPResult.fail(jsonObject.getInteger("code"), jsonObject.getString("msg")); - response.setBody(wvpResult); + return response; } + logger.info("[redis-rpc] 鍙戞祦鎴愬姛锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); + WVPResult wvpResult = WVPResult.success(); + response.setBody(wvpResult); return response; } @@ -244,25 +237,24 @@ return response; } logger.info("[redis-rpc] 鍋滄鎺ㄦ祦锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); - MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - if (mediaServerItem == null) { + MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId()); + if (mediaServer == null) { logger.info("[redis-rpc] stopSendRtp->鏈壘鍒癕ediaServer锛� {}", sendRtpItem.getMediaServerId() ); WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "鏈壘鍒癕ediaServer"); response.setBody(wvpResult); return response; } - JSONObject jsonObject = zlmServerFactory.stopSendRtpStream(mediaServerItem, sendRtpItem); - if (jsonObject.getInteger("code") == 0) { - logger.info("[redis-rpc] 鍋滄鎺ㄦ祦鎴愬姛锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); - response.setBody(WVPResult.success()); - return response; - }else { - int code = jsonObject.getInteger("code"); - String msg = jsonObject.getString("msg"); - logger.info("[redis-rpc] 鍋滄鎺ㄦ祦澶辫触锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}锛� code锛� {}, msg: {}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort(), code, msg ); - response.setBody(WVPResult.fail(code, msg)); + try { + mediaServerService.stopSendRtp(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc()); + }catch (ControllerException exception) { + logger.info("[redis-rpc] 鍋滄鎺ㄦ祦澶辫触锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}锛� code锛� {}, msg: {}", sendRtpItem.getApp(), + sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort(), exception.getCode(), exception.getMsg() ); + response.setBody(WVPResult.fail(exception.getCode(), exception.getMsg())); return response; } + logger.info("[redis-rpc] 鍋滄鎺ㄦ祦鎴愬姛锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); + response.setBody(WVPResult.success()); + return response; } /** diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index 21911b9..d87c342 100755 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -2,10 +2,7 @@ import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.common.SystemAllInfo; -import com.genersoft.iot.vmp.gb28181.bean.AlarmChannelMessage; -import com.genersoft.iot.vmp.gb28181.bean.Device; -import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch; -import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; +import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.media.bean.MediaInfo; import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent; diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java index 10d0ee1..4f12746 100755 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java @@ -357,7 +357,8 @@ "<if test='item.gpsTime != null'>, gps_time=#{item.gpsTime}</if>" + "<if test='item.streamIdentification != null'>, stream_identification=#{item.streamIdentification}</if>" + "<if test='item.id > 0'>WHERE id=#{item.id}</if>" + - "<if test='item.id == 0'>WHERE device_id=#{item.deviceId} AND channel_id=#{item.channelId}</if>" + + "<if test='item.id == 0 and item.channelId != null '>WHERE device_id=#{item.deviceId} AND channel_id=#{item.channelId}</if>" + + "<if test='item.id == 0 and item.channelId == null '>WHERE device_id=#{item.deviceId}</if>" + "</foreach>" + "</script>"}) int batchUpdate(List<DeviceChannel> updateChannels); -- Gitblit v1.8.0