From 407a5edebfe43395d37e414604e5c5100fd605a8 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期三, 17 四月 2024 23:31:48 +0800 Subject: [PATCH] 优化多wvp国标级联推流时推流信息的清理 --- src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcController.java | 75 ++++++++++++++++++++++++++++++++++--- 1 files changed, 68 insertions(+), 7 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 7a81eab..69e942f 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 @@ -7,8 +7,10 @@ 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.gb28181.bean.ParentPlatform; 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.zlm.SendRtpPortManager; import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; @@ -18,12 +20,17 @@ 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.WVPResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; + +import javax.sip.InvalidArgumentException; +import javax.sip.SipException; +import java.text.ParseException; /** * 鍏朵粬wvp鍙戣捣鐨剅pc璋冪敤锛岃繖閲岀殑鏂规硶琚� RedisRpcConfig 閫氳繃鍙嶅皠瀵绘壘瀵瑰簲鐨勬柟娉曞悕绉拌皟鐢� @@ -59,12 +66,20 @@ private RedisTemplate<Object, Object> redisTemplate; + @Autowired + private ISIPCommanderForPlatform commanderFroPlatform; + + + @Autowired + private IVideoManagerStorage storager; + + /** * 鑾峰彇鍙戞祦鐨勪俊鎭� */ public RedisRpcResponse getSendRtpItem(RedisRpcRequest request) { SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - logger.info("[redis-rpc] 鑾峰彇鍙戞祦鐨勪俊鎭細 {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); + logger.info("[redis-rpc] 鑾峰彇鍙戞祦鐨勪俊鎭細 {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); // 鏌ヨ鏈骇鏄惁鏈夎繖涓祦 MediaServerItem mediaServerItem = mediaServerService.getMediaServerByAppAndStream(sendRtpItem.getApp(), sendRtpItem.getStream()); if (mediaServerItem == null) { @@ -99,11 +114,11 @@ */ public RedisRpcResponse waitePushStreamOnline(RedisRpcRequest request) { SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - logger.info("[redis-rpc] 鐩戝惉娴佷笂绾匡細 {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); + 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() ); + logger.info("[redis-rpc] 鐩戝惉娴佷笂绾挎椂鍙戠幇娴佸凡瀛樺湪鐩存帴杩斿洖锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); RedisRpcResponse response = request.getResponse(); response.setBody(sendRtpItem); response.setStatusCode(200); @@ -113,7 +128,7 @@ sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null); hookSubscribe.addSubscribe(hook, (MediaServerItem mediaServerItemInUse, HookParam hookParam) -> { - logger.info("[redis-rpc] 鐩戝惉娴佷笂绾匡紝娴佸凡涓婄嚎锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); + logger.info("[redis-rpc] 鐩戝惉娴佷笂绾匡紝娴佸凡涓婄嚎锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); // 璇诲彇redis涓殑涓婄骇鐐规挱淇℃伅锛岀敓鎴恠endRtpItm鍙戦�佸嚭鍘� if (sendRtpItem.getSsrc() == null) { // 涓婄骇骞冲彴鐐规挱鏃朵笉浣跨敤涓婄骇骞冲彴鎸囧畾鐨剆src锛屼娇鐢ㄨ嚜瀹氫箟鐨剆src锛屽弬鑰冨浗鏍囨枃妗�-鐐规挱澶栧煙璁惧濯掍綋娴丼SRC澶勭悊鏂瑰紡 @@ -123,6 +138,7 @@ sendRtpItem.setMediaServerId(mediaServerItemInUse.getId()); sendRtpItem.setLocalIp(mediaServerItemInUse.getSdpIp()); sendRtpItem.setServerId(userSetting.getServerId()); + RedisRpcResponse response = request.getResponse(); response.setBody(sendRtpItem); response.setStatusCode(200); @@ -133,12 +149,27 @@ return null; } + /** + * 鍋滄鐩戝惉娴佷笂绾� + */ + public RedisRpcResponse stopWaitePushStreamOnline(RedisRpcRequest request) { + SendRtpItem sendRtpItem = JSON.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); + hookSubscribe.removeSubscribe(hook); + return null; + } + /** * 寮�濮嬪彂娴� */ public RedisRpcResponse startSendRtp(RedisRpcRequest request) { SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); + logger.info("[redis-rpc] 寮�濮嬪彂娴侊細 {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); if (mediaServerItem == null) { logger.info("[redis-rpc] startSendRtp->鏈壘鍒癕ediaServer锛� {}", sendRtpItem.getMediaServerId() ); @@ -156,9 +187,11 @@ RedisRpcResponse response = request.getResponse(); response.setStatusCode(200); if (jsonObject.getInteger("code") == 0) { + logger.info("[redis-rpc] 鍙戞祦鎴愬姛锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort()); WVPResult wvpResult = WVPResult.success(); 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); } @@ -170,7 +203,7 @@ */ public RedisRpcResponse stopSendRtp(RedisRpcRequest request) { SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); - logger.info("[redis-rpc] 鍋滄鎺ㄦ祦锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); + logger.info("[redis-rpc] 鍋滄鎺ㄦ祦锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId()); if (mediaServerItem == null) { logger.info("[redis-rpc] stopSendRtp->鏈壘鍒癕ediaServer锛� {}", sendRtpItem.getMediaServerId() ); @@ -181,19 +214,47 @@ RedisRpcResponse response = request.getResponse(); response.setStatusCode(200); if (jsonObject.getInteger("code") == 0) { - logger.info("[redis-rpc] 鍋滄鎺ㄦ祦鎴愬姛锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream() ); + logger.info("[redis-rpc] 鍋滄鎺ㄦ祦鎴愬姛锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); WVPResult wvpResult = WVPResult.success(); 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 ); + logger.info("[redis-rpc] 鍋滄鎺ㄦ祦澶辫触锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}锛� code锛� {}, msg: {}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort(), code, msg ); WVPResult wvpResult = WVPResult.fail(code, msg); response.setBody(wvpResult); } return response; } + /** + * 鍏朵粬wvp閫氱煡鎺ㄦ祦宸茬粡鍋滄浜� + */ + public RedisRpcResponse rtpSendStopped(RedisRpcRequest request) { + SendRtpItem sendRtpItem = JSON.parseObject(request.getParam().toString(), SendRtpItem.class); + logger.info("[redis-rpc] 鎺ㄦ祦宸茬粡鍋滄锛� {}/{}, 鐩爣鍦板潃锛� {}锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); + SendRtpItem sendRtpItemInCatch = redisCatchStorage.querySendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), sendRtpItem.getStream(), sendRtpItem.getCallId()); + RedisRpcResponse response = request.getResponse(); + response.setStatusCode(200); + if (sendRtpItemInCatch == null) { + return response; + } + String platformId = sendRtpItem.getPlatformId(); + ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId); + if (platform == null) { + return response; + } + try { + commanderFroPlatform.streamByeCmd(platform, sendRtpItem); + redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(), + sendRtpItem.getCallId(), sendRtpItem.getStream()); + redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, platform); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍙戦�丅YE: {}", e.getMessage()); + } + return response; + } + private void sendResponse(RedisRpcResponse response){ response.setToId(userSetting.getServerId()); RedisRpcMessage message = new RedisRpcMessage(); -- Gitblit v1.8.0