diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 090de960..ff84fc43 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -771,7 +771,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel); // 设置超时 dynamicTask.startDelay(sendRtpItem.getCallId(), () -> { - redisRpcService.stopWaitePushStreamOnline(sendRtpItem); + redisRpcService.stopWaitePushStreamOnline(sendRtpItem.getRedisKey(), sendRtpItem); logger.info("[ app={}, stream={} ] 等待设备开始推流超时", sendRtpItem.getApp(), sendRtpItem.getStream()); try { responseAck(request, Response.REQUEST_TIMEOUT); // 超时 @@ -834,7 +834,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements redisPushStreamResponseListener.addEvent(sendRtpItem.getApp(), sendRtpItem.getStream(), response -> { if (response.getCode() != 0) { dynamicTask.stop(sendRtpItem.getCallId()); - redisRpcService.stopWaitePushStreamOnline(sendRtpItem); + redisRpcService.stopWaitePushStreamOnline(sendRtpItem.getRedisKey(), sendRtpItem); try { responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg()); } catch (SipException | InvalidArgumentException | ParseException e) { diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java index d3a8acde..70d53bc8 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java @@ -14,7 +14,7 @@ public interface IRedisRpcService { void waitePushStreamOnline(SendRtpItem sendRtpItem, CommonCallback callback); - void stopWaitePushStreamOnline(SendRtpItem sendRtpItem); + void stopWaitePushStreamOnline(String sendRtpItemKey, SendRtpItem sendRtpItem); void rtpSendStopped(String sendRtpItemKey); 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 250d9a2c..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 @@ -174,7 +174,14 @@ public class RedisRpcController { * 停止监听流上线 */ public RedisRpcResponse stopWaitePushStreamOnline(RedisRpcRequest request) { - SendRtpItem sendRtpItem = JSONObject.parseObject(request.getParam().toString(), SendRtpItem.class); + String sendRtpItemKey = request.getParam().toString(); + SendRtpItem sendRtpItem = (SendRtpItem) redisTemplate.opsForValue().get(sendRtpItemKey); + if (sendRtpItem == null) { + logger.info("[redis-rpc] 停止监听流上线, 未找到redis中的发流信息, key:{}", sendRtpItemKey); + RedisRpcResponse response = request.getResponse(); + response.setStatusCode(200); + return response; + } logger.info("[redis-rpc] 停止监听流上线: {}/{}, 目标地址: {}:{}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort() ); // 监听流上线。 流上线直接发送sendRtpItem消息给实际的信令处理者 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 c4406cdf..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 @@ -126,12 +126,12 @@ public class RedisRpcServiceImpl implements IRedisRpcService { } @Override - public void stopWaitePushStreamOnline(SendRtpItem sendRtpItem) { - logger.info("[停止WVP监听流上线] {}/{}, key:{}", sendRtpItem.getApp(), sendRtpItem.getStream(), 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); }