From 55a240bb4504baa9a75e44bc6cc597c96b80705d Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期四, 18 四月 2024 15:52:34 +0800
Subject: [PATCH] 临时提交

---
 src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java |   58 +++++++++++++++++++++++++++++++++++++++++-----------------
 1 files changed, 41 insertions(+), 17 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java
index 9dcfadf..f4c429c 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java
@@ -14,10 +14,12 @@
 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.redisMsg.IRedisRpcService;
+import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 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.Service;
 
 @Service
@@ -37,6 +39,9 @@
     @Autowired
     private SSRCFactory ssrcFactory;
 
+    @Autowired
+    private RedisTemplate<Object, Object> redisTemplate;
+
     private RedisRpcRequest buildRequest(String uri, Object param) {
         RedisRpcRequest request = new RedisRpcRequest();
         request.setFromId(userSetting.getServerId());
@@ -46,32 +51,40 @@
     }
 
     @Override
-    public SendRtpItem getSendRtpItem(SendRtpItem sendRtpItem) {
-
-        RedisRpcRequest request = buildRequest("getSendRtpItem", sendRtpItem);
+    public SendRtpItem getSendRtpItem(String sendRtpItemKey) {
+        RedisRpcRequest request = buildRequest("getSendRtpItem", sendRtpItemKey);
         RedisRpcResponse response = redisRpcConfig.request(request, 10);
-        return JSON.parseObject(response.getBody().toString(), SendRtpItem.class);
+        if (response.getBody() == null) {
+            return null;
+        }
+        return (SendRtpItem)redisTemplate.opsForValue().get(response.getBody().toString());
     }
 
     @Override
-    public WVPResult startSendRtp(SendRtpItem sendRtpItem) {
+    public WVPResult startSendRtp(String sendRtpItemKey, SendRtpItem sendRtpItem) {
         logger.info("[璇锋眰鍏朵粬WVP] 寮�濮嬫帹娴侊紝wvp锛歿}锛� {}/{}", sendRtpItem.getServerId(), sendRtpItem.getApp(), sendRtpItem.getStream());
-        RedisRpcRequest request = buildRequest("startSendRtp", sendRtpItem);
+        RedisRpcRequest request = buildRequest("startSendRtp", sendRtpItemKey);
         request.setToId(sendRtpItem.getServerId());
         RedisRpcResponse response = redisRpcConfig.request(request, 10);
         return JSON.parseObject(response.getBody().toString(), WVPResult.class);
     }
 
     @Override
-    public WVPResult stopSendRtp(SendRtpItem sendRtpItem) {
-        RedisRpcRequest request = buildRequest("stopSendRtp", sendRtpItem);
+    public WVPResult stopSendRtp(String sendRtpItemKey) {
+        SendRtpItem sendRtpItem = (SendRtpItem)redisTemplate.opsForValue().get(sendRtpItemKey);
+        if (sendRtpItem == null) {
+            logger.info("[璇锋眰鍏朵粬WVP] 鍋滄鎺ㄦ祦, 鏈壘鍒皉edis涓殑鍙戞祦淇℃伅锛� key锛歿}", sendRtpItemKey);
+            return WVPResult.fail(ErrorCode.ERROR100.getCode(), "鏈壘鍒板彂娴佷俊鎭�");
+        }
+        logger.info("[璇锋眰鍏朵粬WVP] 鍋滄鎺ㄦ祦锛寃vp锛歿}锛� {}/{}", sendRtpItem.getServerId(), sendRtpItem.getApp(), sendRtpItem.getStream());
+        RedisRpcRequest request = buildRequest("stopSendRtp", sendRtpItemKey);
         request.setToId(sendRtpItem.getServerId());
         RedisRpcResponse response = redisRpcConfig.request(request, 10);
         return JSON.parseObject(response.getBody().toString(), WVPResult.class);
     }
 
     @Override
-    public void waitePushStreamOnline(SendRtpItem sendRtpItem, CommonCallback<SendRtpItem> callback) {
+    public void waitePushStreamOnline(SendRtpItem sendRtpItem, CommonCallback<String> callback) {
         logger.info("[璇锋眰鎵�鏈塛VP鐩戝惉娴佷笂绾縘 {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
         // 鐩戝惉娴佷笂绾裤�� 娴佷笂绾跨洿鎺ュ彂閫乻endRtpItem娑堟伅缁欏疄闄呯殑淇′护澶勭悊鑰�
         HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed(
@@ -87,36 +100,47 @@
             sendRtpItem.setMediaServerId(mediaServerItemInUse.getId());
             sendRtpItem.setLocalIp(mediaServerItemInUse.getSdpIp());
             sendRtpItem.setServerId(userSetting.getServerId());
+            redisTemplate.opsForValue().set(sendRtpItem.getRedisKey(), sendRtpItem);
             if (callback != null) {
-                callback.run(sendRtpItem);
+                callback.run(sendRtpItem.getRedisKey());
             }
             hookSubscribe.removeSubscribe(hook);
         });
         RedisRpcRequest request = buildRequest("waitePushStreamOnline", sendRtpItem);
         request.setToId(sendRtpItem.getServerId());
         redisRpcConfig.request(request, response -> {
-            SendRtpItem sendRtpItemFromOther = JSON.parseObject(response.getBody().toString(), SendRtpItem.class);
-            logger.info("[璇锋眰鎵�鏈塛VP鐩戝惉娴佷笂绾縘 娴佷笂绾� {}/{}->{}", sendRtpItemFromOther.getApp(), sendRtpItemFromOther.getStream(), sendRtpItemFromOther);
+            if (response.getBody() == null) {
+                logger.info("[璇锋眰鎵�鏈塛VP鐩戝惉娴佷笂绾縘 娴佷笂绾�,浣嗘槸鏈壘鍒板彂娴佷俊鎭細{}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
+                return;
+            }
+            logger.info("[璇锋眰鎵�鏈塛VP鐩戝惉娴佷笂绾縘 娴佷笂绾� {}/{}->{}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.toString());
+
             if (callback != null) {
-                callback.run(sendRtpItemFromOther);
+                callback.run(response.getBody().toString());
             }
         });
 
     }
 
     @Override
-    public void stopWaitePushStreamOnline(SendRtpItem sendRtpItem) {
+    public void stopWaitePushStreamOnline(String sendRtpItemKey, SendRtpItem sendRtpItem) {
+        logger.info("[鍋滄WVP鐩戝惉娴佷笂绾縘 {}/{}锛� key锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItemKey);
         HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed(
                 sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null);
         hookSubscribe.removeSubscribe(hook);
-        RedisRpcRequest request = buildRequest("stopWaitePushStreamOnline", sendRtpItem);
+        RedisRpcRequest request = buildRequest("stopWaitePushStreamOnline", sendRtpItemKey);
         request.setToId(sendRtpItem.getServerId());
         redisRpcConfig.request(request, 10);
     }
 
     @Override
-    public void rtpSendStopped(SendRtpItem sendRtpItem) {
-        RedisRpcRequest request = buildRequest("rtpSendStopped", sendRtpItem);
+    public void rtpSendStopped(String sendRtpItemKey) {
+        SendRtpItem sendRtpItem = (SendRtpItem)redisTemplate.opsForValue().get(sendRtpItemKey);
+        if (sendRtpItem == null) {
+            logger.info("[鍋滄WVP鐩戝惉娴佷笂绾縘 鏈壘鍒皉edis涓殑鍙戞祦淇℃伅锛� key锛歿}", sendRtpItemKey);
+            return;
+        }
+        RedisRpcRequest request = buildRequest("rtpSendStopped", sendRtpItemKey);
         request.setToId(sendRtpItem.getServerId());
         redisRpcConfig.request(request, 10);
     }

--
Gitblit v1.8.0