修复多wvp国标级联机制

2024-04-17 12:56:22 +08:00
parent f357bece62
commit bf6e09d231
12 changed files with 237 additions and 76 deletions
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java
@@ -13,4 +13,9 @@ public interface IRedisRpcService {
    void waitePushStreamOnline(SendRtpItem sendRtpItem, CommonCallback<SendRtpItem> callback);

    WVPResult stopSendRtp(SendRtpItem sendRtpItem);
+
+    void stopWaitePushStreamOnline(SendRtpItem sendRtpItem);
+
+    void rtpSendStopped(SendRtpItem sendRtpItem);
+
 }
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java
@@ -0,0 +1,76 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson2.JSON;
+import com.genersoft.iot.vmp.service.bean.MessageForPushChannelResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ObjectUtils;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * 接收redis返回的推流结果
+ * @author lin
+ */
+@Component
+public class RedisPushStreamResponseListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamResponseListener.class);
+
+    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+
+    private Map<String, PushStreamResponseEvent> responseEvents = new ConcurrentHashMap<>();
+
+    public interface PushStreamResponseEvent{
+        void run(MessageForPushChannelResponse response);
+    }
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        logger.info("[REDIS消息-请求推流结果]： {}", new String(message.getBody()));
+        boolean isEmpty = taskQueue.isEmpty();
+        taskQueue.offer(message);
+        if (isEmpty) {
+            taskExecutor.execute(() -> {
+                while (!taskQueue.isEmpty()) {
+                    Message msg = taskQueue.poll();
+                    try {
+                        MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class);
+                        if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){
+                            logger.info("[REDIS消息-请求推流结果]：参数不全");
+                            continue;
+                        }
+                        // 查看正在等待的invite消息
+                        if (responseEvents.get(response.getApp() + response.getStream()) != null) {
+                            responseEvents.get(response.getApp() + response.getStream()).run(response);
+                        }
+                    }catch (Exception e) {
+                        logger.warn("[REDIS消息-请求推流结果] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
+                        logger.error("[REDIS消息-请求推流结果] 异常内容： ", e);
+                    }
+                }
+            });
+        }
+    }
+
+    public void addEvent(String app, String stream, PushStreamResponseEvent callback) {
+        responseEvents.put(app + stream, callback);
+    }
+
+    public void removeEvent(String app, String stream) {
+        responseEvents.remove(app + stream);
+    }
+}
--- 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.RedisRpcConfig;
 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,6 +20,7 @@ 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.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;
@@ -25,6 +28,10 @@ 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发起的rpc调用，这里的方法被 RedisRpcConfig 通过反射寻找对应的方法名称调用
 */
@@ -59,6 +66,14 @@ public class RedisRpcController {
    private RedisTemplate<Object, Object> redisTemplate;


+    @Autowired
+    private ISIPCommanderForPlatform commanderFroPlatform;
+
+
+    @Autowired
+    private IVideoManagerStorage storager;
+
+
    /**
     * 获取发流的信息
     */
@@ -133,6 +148,20 @@ public class RedisRpcController {
        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消息给实际的信令处理者
+        HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed(
+                sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null);
+        hookSubscribe.removeSubscribe(hook);
+        return null;
+    }
+

    /**
     * 开始发流
@@ -194,6 +223,34 @@ public class RedisRpcController {
        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 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("[命令发送失败] 发送BYE: {}", e.getMessage());
+        }
+        return response;
+    }
+
    private void sendResponse(RedisRpcResponse response){
        response.setToId(userSetting.getServerId());
        RedisRpcMessage message = new RedisRpcMessage();
--- 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
@@ -97,4 +97,21 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
        });

    }
+
+    @Override
+    public void stopWaitePushStreamOnline(SendRtpItem sendRtpItem) {
+        HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed(
+                sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", null);
+        hookSubscribe.removeSubscribe(hook);
+        RedisRpcRequest request = buildRequest("stopWaitePushStreamOnline", sendRtpItem);
+        request.setToId(sendRtpItem.getServerId());
+        redisRpcConfig.request(request, 10);
+    }
+
+    @Override
+    public void rtpSendStopped(SendRtpItem sendRtpItem) {
+        RedisRpcRequest request = buildRequest("rtpSendStopped", sendRtpItem);
+        request.setToId(sendRtpItem.getServerId());
+        redisRpcConfig.request(request, 10);
+    }
 }