From 764d04b497356ba6bcbb75fd42b51eca750f7223 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期三, 29 五月 2024 15:02:51 +0800 Subject: [PATCH] 调整上级观看消息的发送 --- src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcController.java | 204 ++++++++++++++++++++++++++++++--------------------- 1 files changed, 120 insertions(+), 84 deletions(-) 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 682f564..adb2318 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 @@ -1,8 +1,8 @@ package com.genersoft.iot.vmp.service.redisMsg.control; -import com.alibaba.fastjson2.JSON; 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; @@ -11,16 +11,16 @@ 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; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -56,11 +56,7 @@ private UserSetting userSetting; @Autowired - private ZlmHttpHookSubscribe hookSubscribe; - - @Autowired - private ZLMServerFactory zlmServerFactory; - + private HookSubscribe hookSubscribe; @Autowired private RedisTemplate<Object, Object> redisTemplate; @@ -78,10 +74,17 @@ * 鑾峰彇鍙戞祦鐨勪俊鎭� */ public RedisRpcResponse getSendRtpItem(RedisRpcRequest request) { - SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - logger.info("[redis-rpc] 鑾峰彇鍙戞祦鐨勪俊鎭細 {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); + String sendRtpItemKey = request.getParam().toString(); + SendRtpItem sendRtpItem = (SendRtpItem) redisTemplate.opsForValue().get(sendRtpItemKey); + if (sendRtpItem == null) { + logger.info("[redis-rpc] 鑾峰彇鍙戞祦鐨勪俊鎭�, 鏈壘鍒皉edis涓殑鍙戞祦淇℃伅锛� key锛歿}", sendRtpItemKey); + RedisRpcResponse response = request.getResponse(); + response.setStatusCode(200); + return response; + } + 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); @@ -103,9 +106,10 @@ sendRtpItem.setSsrc(ssrc); } redisCatchStorage.updateSendRTPSever(sendRtpItem); + redisTemplate.opsForValue().set(sendRtpItemKey, sendRtpItem); RedisRpcResponse response = request.getResponse(); response.setStatusCode(200); - response.setBody(sendRtpItem); + response.setBody(sendRtpItemKey); return response; } @@ -113,36 +117,48 @@ * 鐩戝惉娴佷笂绾� */ public RedisRpcResponse waitePushStreamOnline(RedisRpcRequest request) { - SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - logger.info("[redis-rpc] 鐩戝惉娴佷笂绾匡細 {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); + 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) { - logger.info("[redis-rpc] 鐩戝惉娴佷笂绾挎椂鍙戠幇娴佸凡瀛樺湪鐩存帴杩斿洖锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); - RedisRpcResponse response = request.getResponse(); - response.setBody(sendRtpItem); - response.setStatusCode(200); - } - // 鐩戝惉娴佷笂绾裤�� 娴佷笂绾跨洿鎺ュ彂閫乻endRtpItem娑堟伅缁欏疄闄呯殑淇′护澶勭悊鑰� - HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed( - sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null); - - hookSubscribe.addSubscribe(hook, (MediaServerItem mediaServerItemInUse, HookParam hookParam) -> { - logger.info("[redis-rpc] 鐩戝惉娴佷笂绾匡紝娴佸凡涓婄嚎锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); + 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(mediaServerItemInUse.getId()) : ssrcFactory.getPlayBackSsrc(mediaServerItemInUse.getId()); + String ssrc = "Play".equalsIgnoreCase(sendRtpItem.getSessionName()) ? ssrcFactory.getPlaySsrc(mediaServer.getId()) : ssrcFactory.getPlayBackSsrc(mediaServer.getId()); sendRtpItem.setSsrc(ssrc); } - sendRtpItem.setMediaServerId(mediaServerItemInUse.getId()); - sendRtpItem.setLocalIp(mediaServerItemInUse.getSdpIp()); + sendRtpItem.setMediaServerId(mediaServer.getId()); + sendRtpItem.setLocalIp(mediaServer.getSdpIp()); sendRtpItem.setServerId(userSetting.getServerId()); + + redisTemplate.opsForValue().set(sendRtpItem.getRedisKey(), sendRtpItem); RedisRpcResponse response = request.getResponse(); - response.setBody(sendRtpItem); + response.setBody(sendRtpItem.getRedisKey()); + response.setStatusCode(200); + } + // 鐩戝惉娴佷笂绾裤�� 娴佷笂绾跨洿鎺ュ彂閫乻endRtpItem娑堟伅缁欏疄闄呯殑淇′护澶勭悊鑰� + 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(hookData.getMediaServer().getId()) : ssrcFactory.getPlayBackSsrc(hookData.getMediaServer().getId()); + sendRtpItem.setSsrc(ssrc); + } + sendRtpItem.setMediaServerId(hookData.getMediaServer().getId()); + sendRtpItem.setLocalIp(hookData.getMediaServer().getSdpIp()); + sendRtpItem.setServerId(userSetting.getServerId()); + + redisTemplate.opsForValue().set(sendRtpItem.getRedisKey(), sendRtpItem); + RedisRpcResponse response = request.getResponse(); + response.setBody(sendRtpItem.getRedisKey()); response.setStatusCode(200); // 鎵嬪姩鍙戦�佺粨鏋� sendResponse(response); + hookSubscribe.removeSubscribe(hook); }); return null; @@ -152,14 +168,14 @@ * 鍋滄鐩戝惉娴佷笂绾� */ public RedisRpcResponse stopWaitePushStreamOnline(RedisRpcRequest request) { - SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - logger.info("[redis-rpc] 鍋滄鐩戝惉娴佷笂绾匡細 {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); - + 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); - return null; + RedisRpcResponse response = request.getResponse(); + response.setStatusCode(200); + return response; } @@ -167,30 +183,42 @@ * 寮�濮嬪彂娴� */ public RedisRpcResponse startSendRtp(RedisRpcRequest request) { - SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - if (mediaServerItem == null) { - logger.info("[redis-rpc] startSendRtp->鏈壘鍒癕ediaServer锛� {}", sendRtpItem.getMediaServerId() ); - RedisRpcResponse response = request.getResponse(); - response.setStatusCode(200); - } - - Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); - if (!streamReady) { - logger.info("[redis-rpc] startSendRtp->娴佷笉鍦ㄧ嚎锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); - RedisRpcResponse response = request.getResponse(); - response.setStatusCode(200); - } - JSONObject jsonObject = zlmServerFactory.startSendRtp(mediaServerItem, sendRtpItem); + String sendRtpItemKey = request.getParam().toString(); + SendRtpItem sendRtpItem = (SendRtpItem) redisTemplate.opsForValue().get(sendRtpItemKey); RedisRpcResponse response = request.getResponse(); response.setStatusCode(200); - if (jsonObject.getInteger("code") == 0) { - WVPResult wvpResult = WVPResult.success(); + if (sendRtpItem == null) { + logger.info("[redis-rpc] 寮�濮嬪彂娴�, 鏈壘鍒皉edis涓殑鍙戞祦淇℃伅锛� key锛歿}", sendRtpItemKey); + WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "鏈壘鍒皉edis涓殑鍙戞祦淇℃伅"); response.setBody(wvpResult); - }else { - 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()); + 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; + } + 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; + } + try { + mediaServerService.startSendRtp(mediaServer, 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); + return response; + } + logger.info("[redis-rpc] 鍙戞祦鎴愬姛锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); + WVPResult wvpResult = WVPResult.success(); + response.setBody(wvpResult); return response; } @@ -198,28 +226,34 @@ * 鍋滄鍙戞祦 */ public RedisRpcResponse stopSendRtp(RedisRpcRequest request) { - SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - logger.info("[redis-rpc] 鍋滄鎺ㄦ祦锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); - MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); - if (mediaServerItem == null) { - logger.info("[redis-rpc] stopSendRtp->鏈壘鍒癕ediaServer锛� {}", sendRtpItem.getMediaServerId() ); - RedisRpcResponse response = request.getResponse(); - response.setStatusCode(200); - } - JSONObject jsonObject = zlmServerFactory.stopSendRtpStream(mediaServerItem, sendRtpItem); + String sendRtpItemKey = request.getParam().toString(); + SendRtpItem sendRtpItem = (SendRtpItem) redisTemplate.opsForValue().get(sendRtpItemKey); RedisRpcResponse response = request.getResponse(); response.setStatusCode(200); - if (jsonObject.getInteger("code") == 0) { - logger.info("[redis-rpc] 鍋滄鎺ㄦ祦鎴愬姛锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); - WVPResult wvpResult = WVPResult.success(); + if (sendRtpItem == null) { + logger.info("[redis-rpc] 鍋滄鎺ㄦ祦, 鏈壘鍒皉edis涓殑鍙戞祦淇℃伅锛� key锛歿}", sendRtpItemKey); + WVPResult wvpResult = WVPResult.fail(ErrorCode.ERROR100.getCode(), "鏈壘鍒皉edis涓殑鍙戞祦淇℃伅"); response.setBody(wvpResult); - }else { - int code = jsonObject.getInteger("code"); - String msg = jsonObject.getString("msg"); - logger.info("[redis-rpc] 鍋滄鎺ㄦ祦澶辫触锛� {}/{}, code锛� {}, msg: {}", sendRtpItem.getApp(), sendRtpItem.getStream(),code, msg ); - WVPResult wvpResult = WVPResult.fail(code, msg); - response.setBody(wvpResult); + return response; } + logger.info("[redis-rpc] 鍋滄鎺ㄦ祦锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); + 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; + } + 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; } @@ -227,14 +261,15 @@ * 鍏朵粬wvp閫氱煡鎺ㄦ祦宸茬粡鍋滄浜� */ public RedisRpcResponse rtpSendStopped(RedisRpcRequest request) { - SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - logger.info("[redis-rpc] 鎺ㄦ祦宸茬粡鍋滄锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); - SendRtpItem sendRtpItemInCatch = redisCatchStorage.querySendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), sendRtpItem.getStream(), sendRtpItem.getCallId()); + String sendRtpItemKey = request.getParam().toString(); + SendRtpItem sendRtpItem = (SendRtpItem) redisTemplate.opsForValue().get(sendRtpItemKey); RedisRpcResponse response = request.getResponse(); response.setStatusCode(200); - if (sendRtpItemInCatch == null) { + if (sendRtpItem == null) { + logger.info("[redis-rpc] 鎺ㄦ祦宸茬粡鍋滄, 鏈壘鍒皉edis涓殑鍙戞祦淇℃伅锛� key锛歿}", sendRtpItemKey); return response; } + logger.info("[redis-rpc] 鎺ㄦ祦宸茬粡鍋滄锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); String platformId = sendRtpItem.getPlatformId(); ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId); if (platform == null) { @@ -252,6 +287,7 @@ } private void sendResponse(RedisRpcResponse response){ + logger.info("[redis-rpc] >> {}", response); response.setToId(userSetting.getServerId()); RedisRpcMessage message = new RedisRpcMessage(); message.setResponse(response); -- Gitblit v1.8.0