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 e416bc3d..e3617a70 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 @@ -88,7 +88,7 @@ public class RedisRpcConfig implements MessageListener { if (userSetting.getServerId().equals(request.getFromId())) { return; } - logger.info("[redis-rpc] >> {}", request); + logger.info("[redis-rpc] << {}", request); Method method = getMethod(request.getUri()); // 没有携带目标ID的可以理解为哪个wvp有结果就哪个回复,携带目标ID,但是如果是不存在的uri则直接回复404 if (userSetting.getServerId().equals(request.getToId())) { @@ -130,6 +130,7 @@ public class RedisRpcConfig implements MessageListener { } private void sendResponse(RedisRpcResponse response){ + logger.info("[redis-rpc] >> {}", response); response.setToId(userSetting.getServerId()); RedisRpcMessage message = new RedisRpcMessage(); message.setResponse(response); @@ -137,6 +138,7 @@ public class RedisRpcConfig implements MessageListener { } private void sendRequest(RedisRpcRequest request){ + logger.info("[redis-rpc] >> {}", request); RedisRpcMessage message = new RedisRpcMessage(); message.setRequest(request); redisTemplate.convertAndSend(REDIS_REQUEST_CHANNEL_KEY, message); 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 b61615bd..0d01c946 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 @@ -164,6 +164,7 @@ public class RedisRpcController { response.setStatusCode(200); // 手动发送结果 sendResponse(response); + hookSubscribe.removeSubscribe(hook); }); return null; @@ -300,6 +301,7 @@ public class RedisRpcController { } private void sendResponse(RedisRpcResponse response){ + logger.info("[redis-rpc] >> {}", response); response.setToId(userSetting.getServerId()); RedisRpcMessage message = new RedisRpcMessage(); message.setResponse(response); 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 f4c429c2..75fc1cae 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 @@ -89,6 +89,8 @@ public class RedisRpcServiceImpl implements IRedisRpcService { // 监听流上线。 流上线直接发送sendRtpItem消息给实际的信令处理者 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中的上级点播信息,生成sendRtpItm发送出去 @@ -105,9 +107,9 @@ public class RedisRpcServiceImpl implements IRedisRpcService { 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("[请求所有WVP监听流上线] 流上线,但是未找到发流信息:{}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); @@ -118,6 +120,7 @@ public class RedisRpcServiceImpl implements IRedisRpcService { if (callback != null) { callback.run(response.getBody().toString()); } + hookSubscribe.removeSubscribe(hook); }); }