From d70bfb53dd5d92d0405f7ce3d2c9bbabce26184b Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期一, 20 五月 2024 11:37:07 +0800 Subject: [PATCH] 修复兼容zlm-pro转码时无人管看自动移除失败的BUG --- src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java | 22 +++++++++++++++------- 1 files changed, 15 insertions(+), 7 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 f4c429c..8ffb562 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 @@ -84,11 +84,13 @@ } @Override - public void waitePushStreamOnline(SendRtpItem sendRtpItem, CommonCallback<String> callback) { + public long waitePushStreamOnline(SendRtpItem sendRtpItem, CommonCallback<String> callback) { logger.info("[璇锋眰鎵�鏈塛VP鐩戝惉娴佷笂绾縘 {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); // 鐩戝惉娴佷笂绾裤�� 娴佷笂绾跨洿鎺ュ彂閫乻endRtpItem娑堟伅缁欏疄闄呯殑淇′护澶勭悊鑰� HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed( sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null); + RedisRpcRequest request = buildRequest("waitePushStreamOnline", sendRtpItem); + request.setToId(sendRtpItem.getServerId()); hookSubscribe.addSubscribe(hook, (MediaServerItem mediaServerItemInUse, HookParam hookParam) -> { // 璇诲彇redis涓殑涓婄骇鐐规挱淇℃伅锛岀敓鎴恠endRtpItm鍙戦�佸嚭鍘� @@ -105,9 +107,9 @@ callback.run(sendRtpItem.getRedisKey()); } hookSubscribe.removeSubscribe(hook); + redisRpcConfig.removeCallback(request.getSn()); }); - RedisRpcRequest request = buildRequest("waitePushStreamOnline", sendRtpItem); - request.setToId(sendRtpItem.getServerId()); + redisRpcConfig.request(request, response -> { if (response.getBody() == null) { logger.info("[璇锋眰鎵�鏈塛VP鐩戝惉娴佷笂绾縘 娴佷笂绾�,浣嗘槸鏈壘鍒板彂娴佷俊鎭細{}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); @@ -118,17 +120,18 @@ if (callback != null) { callback.run(response.getBody().toString()); } + hookSubscribe.removeSubscribe(hook); }); - + return request.getSn(); } @Override - public void stopWaitePushStreamOnline(String sendRtpItemKey, SendRtpItem sendRtpItem) { - logger.info("[鍋滄WVP鐩戝惉娴佷笂绾縘 {}/{}锛� key锛歿}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItemKey); + public void stopWaitePushStreamOnline(SendRtpItem sendRtpItem) { + logger.info("[鍋滄WVP鐩戝惉娴佷笂绾縘 {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed( sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null); hookSubscribe.removeSubscribe(hook); - RedisRpcRequest request = buildRequest("stopWaitePushStreamOnline", sendRtpItemKey); + RedisRpcRequest request = buildRequest("stopWaitePushStreamOnline", sendRtpItem); request.setToId(sendRtpItem.getServerId()); redisRpcConfig.request(request, 10); } @@ -144,4 +147,9 @@ request.setToId(sendRtpItem.getServerId()); redisRpcConfig.request(request, 10); } + + @Override + public void removeCallback(long key) { + redisRpcConfig.removeCallback(key); + } } -- Gitblit v1.8.0