优化CivilCode缓存

648540858

2024-03-14 d4f6ec39b7e0421757a6b9d1a68b1c4610ea2e8c

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

old mode 100644
new mode 100755
@@ -6,13 +6,14 @@
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.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
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;
@@ -72,10 +73,13 @@
    private RedisTemplate<Object, Object> redisTemplate;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;
    private ZLMServerFactory zlmServerFactory;

    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private IRedisCatchStorage redisCatchStorage;


    @Autowired
@@ -113,8 +117,8 @@
                while (!taskQueue.isEmpty()) {
                    Message msg = taskQueue.poll();
                    try {
                        JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class);
                        WvpRedisMsg wvpRedisMsg = JSON.to(WvpRedisMsg.class, msgJSON);
                        WvpRedisMsg wvpRedisMsg = JSON.parseObject(msg.getBody(), WvpRedisMsg.class);
                        logger.info("[收到REDIS通知] 消息： {}", JSON.toJSONString(wvpRedisMsg));
                        if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
                            continue;
                        }
@@ -123,12 +127,13 @@

                            switch (wvpRedisMsg.getCmd()){
                                case WvpRedisMsgCmd.GET_SEND_ITEM:
                                    RequestSendItemMsg content = JSON.to(RequestSendItemMsg.class, wvpRedisMsg.getContent());
                                    RequestSendItemMsg content = JSON.parseObject(wvpRedisMsg.getContent(), RequestSendItemMsg.class);
                                    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());

                                    break;
                                default:
                                    break;
@@ -229,7 +234,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);
    }
@@ -241,7 +246,7 @@
        result.setData(content);

        WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, result);
                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, JSON.toJSONString(result));
        JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
    }
@@ -259,15 +264,15 @@
            result.setMsg("流媒体不存在");

            WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
                    WvpRedisMsgCmd.GET_SEND_ITEM, serial, result);
                    WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result));

            JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
            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 {
@@ -282,7 +287,7 @@
                WVPResult<SendRtpItem> result = new WVPResult<>();
                result.setCode(ERROR_CODE_TIMEOUT);
                WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
                        userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
                        userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
                );
                JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
                redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
@@ -291,7 +296,7 @@
            // 添加订阅
            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);
                    });
@@ -303,9 +308,6 @@
            String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_REQUESTED;
            logger.info("[redis发送通知] 推流被请求 {}: {}/{}", key, messageForPushChannel.getApp(), messageForPushChannel.getStream());
            redisTemplate.convertAndSend(key, JSON.toJSON(messageForPushChannel));

//            redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);

        }
    }

@@ -313,12 +315,10 @@
     * 将获取到的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(), ssrcFromCallback -> {
                    return querySendRTPServer(content.getPlatformId(), content.getChannelId(), content.getStream(), null) != null;
                });
                content.getTcp(), content.getRtcp());

        WVPResult<ResponseSendItemMsg> result = new WVPResult<>();
        result.setCode(0);
@@ -326,9 +326,10 @@
        responseSendItemMsg.setSendRtpItem(sendRtpItem);
        responseSendItemMsg.setMediaServerItem(mediaServerItem);
        result.setData(responseSendItemMsg);
        redisCatchStorage.updateSendRTPSever(sendRtpItem);

        WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
                userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
                userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
        );
        JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
@@ -354,7 +355,7 @@
        requestSendItemMsg.setServerId(serverId);
        String key = UUID.randomUUID().toString();
        WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId, WvpRedisMsgCmd.GET_SEND_ITEM,
                key, requestSendItemMsg);
                key, JSON.toJSONString(requestSendItemMsg));

        JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
        logger.info("[请求推流SendItem] {}: {}", serverId, jsonObject);
@@ -379,7 +380,7 @@
    public void sendMsgForStartSendRtpStream(String serverId, RequestPushStreamMsg param, PlayMsgCallbackForStartSendRtpStream callback) {
        String key = UUID.randomUUID().toString();
        WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId,
                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, key, param);
                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, key, JSON.toJSONString(param));

        JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
        logger.info("[REDIS 请求其他平台推流] {}: {}", serverId, jsonObject);