Merge branch '648540858:wvp-28181-2.0' into wvp-28181-2.0

xu-bin-bin

2023-10-24 ae9976bb3a59cc3d9952dda6d9388bb412baa158

 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java

old mode 100644
new mode 100755
@@ -2,18 +2,17 @@

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.*;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import org.slf4j.Logger;
@@ -22,11 +21,13 @@
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -68,27 +69,21 @@


    @Autowired
    private ZLMMediaListManager zlmMediaListManager;
    private RedisTemplate<Object, Object> redisTemplate;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;
    private ZLMServerFactory zlmServerFactory;

    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private DynamicTask dynamicTask;

    @Autowired
    private ZLMMediaListManager mediaListManager;

    @Autowired
    private ZlmHttpHookSubscribe subscribe;

    private boolean taskQueueHandlerRun = false;

    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();

@@ -111,107 +106,106 @@

    @Override
    public void onMessage(Message message, byte[] bytes) {

        boolean isEmpty = taskQueue.isEmpty();
        taskQueue.offer(message);
        if (!taskQueueHandlerRun) {
            taskQueueHandlerRun = true;
        if (isEmpty) {
            taskExecutor.execute(() -> {
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
                    WvpRedisMsg wvpRedisMsg = JSON.toJavaObject(msgJSON, WvpRedisMsg.class);
                    if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
                        continue;
                    }
                    if (WvpRedisMsg.isRequest(wvpRedisMsg)) {
                        logger.info("[收到REDIS通知] 请求： {}", new String(msg.getBody()));

                        switch (wvpRedisMsg.getCmd()){
                            case WvpRedisMsgCmd.GET_SEND_ITEM:
                                RequestSendItemMsg content = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestSendItemMsg.class);
                                requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
                                break;
                            case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
                                RequestPushStreamMsg param = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestPushStreamMsg.class);;
                                requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
                                break;
                            default:
                                break;
                    try {
                        JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
                        WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
                        if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
                            continue;
                        }
                        if (WvpRedisMsg.isRequest(wvpRedisMsg)) {
                            logger.info("[收到REDIS通知] 请求： {}", new String(msg.getBody()));

                    }else {
                        logger.info("[收到REDIS通知] 回复： {}", new String(msg.getBody()));
                        switch (wvpRedisMsg.getCmd()){
                            case WvpRedisMsgCmd.GET_SEND_ITEM:
                            switch (wvpRedisMsg.getCmd()){
                                case WvpRedisMsgCmd.GET_SEND_ITEM:
                                    RequestSendItemMsg content = JSON.to(RequestSendItemMsg.class, wvpRedisMsg.getContent());
                                    requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
                                    break;
                                case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
                                    RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());
                                    requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());

                                WVPResult content  = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class);
                                    break;
                                default:
                                    break;
                            }

                                String key = wvpRedisMsg.getSerial();
                                switch (content.getCode()) {
                                    case 0:
                                        ResponseSendItemMsg responseSendItemMsg =JSON.toJavaObject((JSONObject)content.getData(), ResponseSendItemMsg.class);
                                        PlayMsgCallback playMsgCallback = callbacks.get(key);
                                        if (playMsgCallback != null) {
                                            callbacksForError.remove(key);
                                            try {
                                                playMsgCallback.handler(responseSendItemMsg);
                                            } catch (ParseException e) {
                                                logger.error("[REDIS消息处理异常] ", e);
                        }else {
                            logger.info("[收到REDIS通知] 回复： {}", new String(msg.getBody()));
                            switch (wvpRedisMsg.getCmd()){
                                case WvpRedisMsgCmd.GET_SEND_ITEM:

                                   WVPResult content  = JSON.to(WVPResult.class, wvpRedisMsg.getContent());

                                    String key = wvpRedisMsg.getSerial();
                                    switch (content.getCode()) {
                                        case 0:
                                           ResponseSendItemMsg responseSendItemMsg =JSON.to(ResponseSendItemMsg.class, content.getData());
                                            PlayMsgCallback playMsgCallback = callbacks.get(key);
                                            if (playMsgCallback != null) {
                                                callbacksForError.remove(key);
                                                try {
                                                    playMsgCallback.handler(responseSendItemMsg);
                                                } catch (ParseException e) {
                                                    logger.error("[REDIS消息处理异常] ", e);
                                                }
                                            }
                                        }
                                        break;
                                    case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
                                    case ERROR_CODE_OFFLINE:
                                    case ERROR_CODE_TIMEOUT:
                                        PlayMsgErrorCallback errorCallback = callbacksForError.get(key);
                                        if (errorCallback != null) {
                                            callbacks.remove(key);
                                            errorCallback.handler(content);
                                        }
                                        break;
                                    default:
                                        break;
                                }
                                break;
                            case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
                                WVPResult wvpResult  = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class);
                                String serial = wvpRedisMsg.getSerial();
                                switch (wvpResult.getCode()) {
                                    case 0:
                                        JSONObject jsonObject = (JSONObject)wvpResult.getData();
                                        PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial);
                                        if (playMsgCallback != null) {
                                            callbacksForError.remove(serial);
                                            playMsgCallback.handler(jsonObject);
                                        }
                                        break;
                                    case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
                                    case ERROR_CODE_OFFLINE:
                                    case ERROR_CODE_TIMEOUT:
                                        PlayMsgErrorCallback errorCallback = callbacksForError.get(serial);
                                        if (errorCallback != null) {
                                            callbacks.remove(serial);
                                            errorCallback.handler(wvpResult);
                                        }
                                        break;
                                    default:
                                        break;
                                }
                                break;
                            default:
                                break;
                                            break;
                                        case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
                                        case ERROR_CODE_OFFLINE:
                                        case ERROR_CODE_TIMEOUT:
                                            PlayMsgErrorCallback errorCallback = callbacksForError.get(key);
                                            if (errorCallback != null) {
                                                callbacks.remove(key);
                                                errorCallback.handler(content);
                                            }
                                            break;
                                        default:
                                            break;
                                    }
                                    break;
                                case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
                                    WVPResult wvpResult  = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
                                    String serial = wvpRedisMsg.getSerial();
                                    switch (wvpResult.getCode()) {
                                        case 0:
                                            JSONObject jsonObject = (JSONObject)wvpResult.getData();
                                            PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial);
                                            if (playMsgCallback != null) {
                                                callbacksForError.remove(serial);
                                                playMsgCallback.handler(jsonObject);
                                            }
                                            break;
                                        case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
                                        case ERROR_CODE_OFFLINE:
                                        case ERROR_CODE_TIMEOUT:
                                            PlayMsgErrorCallback errorCallback = callbacksForError.get(serial);
                                            if (errorCallback != null) {
                                                callbacks.remove(serial);
                                                errorCallback.handler(wvpResult);
                                            }
                                            break;
                                        default:
                                            break;
                                    }
                                    break;
                                default:
                                    break;
                            }

                        }
                    }catch (Exception e) {
                        logger.warn("[RedisGbPlayMsg] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
                        logger.error("[RedisGbPlayMsg] 异常内容： ", e);
                    }
                }
                taskQueueHandlerRun = false;
            });
        }






    }

    /**
@@ -236,7 +230,7 @@
        param.put("pt", requestPushStreamMsg.getPt());
        param.put("use_ps", requestPushStreamMsg.isPs() ? "1" : "0");
        param.put("only_audio", requestPushStreamMsg.isOnlyAudio() ? "1" : "0");
        JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
        JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaInfo, param);
        // 回复消息
        responsePushStream(jsonObject, fromId, serial);
    }
@@ -250,7 +244,7 @@
        WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, result);
        JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
        RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
    }

    /**
@@ -269,12 +263,12 @@
                    WvpRedisMsgCmd.GET_SEND_ITEM, serial, result);

            JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
            RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
            redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
            return;
        }
        // 确定流是否在线
        boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream());
        if (streamReady) {
        Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream());
        if (streamReady != null && streamReady) {
            logger.info("[回复推流信息]  {}/{}", content.getApp(), content.getStream());
            responseSendItem(mediaServerItem, content, toId, serial);
        }else {
@@ -292,13 +286,13 @@
                        userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
                );
                JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
                RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
                redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
            }, userSetting.getPlatformPlayTimeout());

            // 添加订阅
            HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(content.getApp(), content.getStream(), true, "rtsp", mediaServerItem.getId());

            subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
            subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam)->{
                        dynamicTask.stop(taskKey);
                        responseSendItem(mediaServerItem, content, toId, serial);
                    });
@@ -306,8 +300,10 @@
            MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1, content.getApp(), content.getStream(),
                    content.getChannelId(), content.getPlatformId(), content.getPlatformName(), content.getServerId(),
                    content.getMediaServerId());
            redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);

            String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_REQUESTED;
            logger.info("[redis发送通知] 推流被请求 {}: {}/{}", key, messageForPushChannel.getApp(), messageForPushChannel.getStream());
            redisTemplate.convertAndSend(key, JSON.toJSON(messageForPushChannel));
        }
    }

@@ -315,7 +311,7 @@
     * 将获取到的sendItem发送出去
     */
    private void responseSendItem(MediaServerItem mediaServerItem, RequestSendItemMsg content, String toId, String serial) {
        SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, content.getIp(),
        SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, content.getIp(),
                content.getPort(), content.getSsrc(), content.getPlatformId(),
                content.getApp(), content.getStream(), content.getChannelId(),
                content.getTcp(), content.getRtcp());
@@ -331,7 +327,7 @@
                userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
        );
        JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
        RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
    }

    /**
@@ -368,7 +364,7 @@
            wvpResult.setMsg("timeout");
            errorCallback.handler(wvpResult);
        }, userSetting.getPlatformPlayTimeout());
        RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
    }

    /**
@@ -393,6 +389,33 @@
            callbacksForStartSendRtpStream.remove(key);
            callbacksForError.remove(key);
        });
        RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
    }

    private SendRtpItem querySendRTPServer(String platformGbId, String channelId, String streamId, String callId) {
        if (platformGbId == null) {
            platformGbId = "*";
        }
        if (channelId == null) {
            channelId = "*";
        }
        if (streamId == null) {
            streamId = "*";
        }
        if (callId == null) {
            callId = "*";
        }
        String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX
                + userSetting.getServerId() + "_*_"
                + platformGbId + "_"
                + channelId + "_"
                + streamId + "_"
                + callId;
        List<Object> scan = RedisUtil.scan(redisTemplate, key);
        if (scan.size() > 0) {
            return (SendRtpItem)redisTemplate.opsForValue().get(scan.get(0));
        }else {
            return null;
        }
    }
}