修复收到catalog消息是更新导致是否有音频的设置失效的BUG

648540858

2024-04-24 c21d973977a9f1d00d26179de764687ddd0ec56c

 src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java

old mode 100644
new mode 100755
@@ -4,6 +4,7 @@
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
@@ -20,7 +21,7 @@
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.*;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaceInfo;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import org.slf4j.Logger;
@@ -36,6 +37,7 @@
import java.util.stream.Collectors;

@Service
@DS("master")
public class StreamPushServiceImpl implements IStreamPushService {

    private final static Logger logger = LoggerFactory.getLogger(StreamPushServiceImpl.class);
@@ -183,8 +185,11 @@

    @Override
    public boolean stop(String app, String streamId) {
        logger.info("[推流 ] 停止流： {}/{}", app, streamId);
        StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId);
        gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
        if (streamPushItem != null) {
            gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
        }

        platformGbStreamMapper.delByAppAndStream(app, streamId);
        gbStreamMapper.del(app, streamId);
@@ -279,6 +284,8 @@
                    redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
                    // 移除redis内流的信息
                    redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineOnStreamChangedHookParam.getApp(), offlineOnStreamChangedHookParam.getStream());
                    // 冗余数据，自己系统中自用
                    redisCatchStorage.removePushListItem(offlineOnStreamChangedHookParam.getApp(), offlineOnStreamChangedHookParam.getStream(), mediaServerItem.getId());
                }
            }

@@ -316,6 +323,9 @@
                jsonObject.put("register", false);
                jsonObject.put("mediaServerId", mediaServerId);
                redisCatchStorage.sendStreamChangeMsg(type, jsonObject);

                // 冗余数据，自己系统中自用
                redisCatchStorage.removePushListItem(onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream(), mediaServerId);
            }
        }
    }
@@ -364,7 +374,7 @@
        // 存储数据到stream_push表
        streamPushMapper.addAll(streamPushItems);
        List<StreamPushItem> streamPushItemForGbStream = streamPushItems.stream()
                .filter(streamPushItem-> streamPushItem.getId() != null)
                .filter(streamPushItem-> streamPushItem.getGbId() != null)
                .collect(Collectors.toList());
        // 存储数据到gb_stream表， id会返回到streamPushItemForGbStream里
        if (streamPushItemForGbStream.size() > 0) {
@@ -437,7 +447,7 @@

                }
            }
            if (streamPushItemListFroPlatform.size() > 0) {
            if (!streamPushItemListFroPlatform.isEmpty()) {
                platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
                // 发送通知
                for (String platformId : platformForEvent.keySet()) {
@@ -503,6 +513,9 @@
        stream.setUpdateTime(DateUtil.getNow());
        stream.setCreateTime(DateUtil.getNow());
        stream.setServerId(userSetting.getServerId());
        stream.setMediaServerId(mediaConfig.getId());
        stream.setSelf(true);
        stream.setPushIng(true);

        // 放在事务内执行
        boolean result = false;
@@ -529,7 +542,32 @@
    }

    @Override
    public ResourceBaceInfo getOverview() {
        return streamPushMapper.getOverview(userSetting.isUsePushingAsStatus());
    public ResourceBaseInfo getOverview() {
        int total = streamPushMapper.getAllCount();
        int online = streamPushMapper.getAllOnline(userSetting.isUsePushingAsStatus());

        return new ResourceBaseInfo(total, online);
    }

    @Override
    public Map<String, StreamPushItem> getAllAppAndStreamMap() {
        return streamPushMapper.getAllAppAndStreamMap();
    }

    @Override
    public void updatePush(OnStreamChangedHookParam param) {
        StreamPushItem transform = transform(param);
        StreamPushItem pushInDb = getPush(param.getApp(), param.getStream());
        transform.setPushIng(param.isRegist());
        transform.setUpdateTime(DateUtil.getNow());
        transform.setPushTime(DateUtil.getNow());
        transform.setSelf(userSetting.getServerId().equals(param.getSeverId()));
        if (pushInDb == null) {
            transform.setCreateTime(DateUtil.getNow());
            streamPushMapper.add(transform);
        }else {
            streamPushMapper.update(transform);
            gbStreamMapper.updateMediaServer(param.getApp(), param.getStream(), param.getMediaServerId());
        }
    }
}