优化级联时的异常处理

panlinlin

2021-04-26 39078225f1104eeaed21b24a024457b892fa7f7a

优化级联时的异常处理

 src/main/java/com/genersoft/iot/vmp/conf/ApplicationCheckRunner.java

@@ -55,7 +55,7 @@
            System.exit(1);
        }

        if (mediaIp.equals("localhost") || mediaIp.equals("127.0.0.1")) {
        if (mediaIp.equals("localhost") || (mediaIp.equals("127.0.0.1") && mediaWanIp == null)) {
            logger.warn("mediaIp.ip使用 {} ,将无法收到网络内其他设备的推流!!!", mediaIp );
        }


 src/main/java/com/genersoft/iot/vmp/conf/SipDeviceRunner.java

New file
@@ -0,0 +1,33 @@
package com.genersoft.iot.vmp.conf;

import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 系统启动时控制设备离线
 */
@Component
@Order(value=4)
public class SipDeviceRunner implements CommandLineRunner {

    @Autowired
    private IVideoManagerStorager storager;

    @Autowired
    private IRedisCatchStorage redisCatchStorage;

    @Override
    public void run(String... args) throws Exception {
        // 设置所有设备离线
        storager.outlineForAll();
    }
}

 src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java

@@ -3,6 +3,7 @@
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import org.springframework.beans.factory.annotation.Autowired;
@@ -28,6 +29,10 @@
    @Autowired
    private EventPublisher publisher;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;


    @Override
    public void run(String... args) throws Exception {
        // 设置所有平台离线
@@ -36,6 +41,9 @@
        // 清理所有平台注册缓存
        redisCatchStorage.cleanPlatformRegisterInfos();

        // 停止所有推流
//        zlmrtpServerFactory.closeAllSendRtpStream();

        List<ParentPlatform> parentPlatforms = storager.queryEnableParentPlatformList(true);

        for (ParentPlatform parentPlatform : parentPlatforms) {

 src/main/java/com/genersoft/iot/vmp/conf/security/WebSecurityConfig.java

@@ -111,7 +111,7 @@
        http.headers().contentTypeOptions().disable();
        http.authorizeRequests()
                // 放行接口
                .antMatchers("/api/user/login","/index/hook/**").permitAll()
                .antMatchers("/#/**", "/api/user/login","/index/hook/**").permitAll()
                // 除上面外的所有请求全部需要鉴权认证
                .anyRequest().authenticated()
                // 异常处理(权限拒绝、登录失效等)

 src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepaliveTimeoutListenerForPlatform.java

@@ -39,8 +39,8 @@
        //  获取失效的key
        String expiredKey = message.toString();
        logger.info(expiredKey);
        if(!expiredKey.startsWith(VideoManagerConstants.PLATFORM_PREFIX)){
            logger.info("收到redis过期监听，但开头不是"+VideoManagerConstants.PLATFORM_PREFIX+"，忽略");
        if(!expiredKey.startsWith(VideoManagerConstants.PLATFORM_KEEPLIVEKEY_PREFIX)){
            logger.debug("收到redis过期监听，但开头不是"+VideoManagerConstants.PLATFORM_KEEPLIVEKEY_PREFIX+"，忽略");
            return;
        }
        // 平台心跳到期,需要重发, 判断是否已经多次未收到心跳回复, 多次未收到,则重新发起注册, 注册尝试多次未得到回复,则认为平台离线
@@ -49,7 +49,6 @@

            publisher.platformKeepaliveExpireEventPublish(platformGBId);
        }else if (expiredKey.startsWith(VideoManagerConstants.PLATFORM_REGISTER_PREFIX)) {
            logger.info("11111111111111");
            String platformGBId = expiredKey.substring(VideoManagerConstants.PLATFORM_REGISTER_PREFIX.length(),expiredKey.length());

            publisher.platformNotRegisterEventPublish(platformGBId);

 src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepliveTimeoutListener.java

@@ -38,7 +38,7 @@
        //  获取失效的key

        String expiredKey = message.toString();

        if(!expiredKey.startsWith(VideoManagerConstants.KEEPLIVEKEY_PREFIX)){

            logger.info("收到redis过期监听，但开头不是"+VideoManagerConstants.KEEPLIVEKEY_PREFIX+"，忽略");

            logger.debug("收到redis过期监听，但开头不是"+VideoManagerConstants.KEEPLIVEKEY_PREFIX+"，忽略");

            return;

        }

        


 src/main/java/com/genersoft/iot/vmp/gb28181/event/platformKeepaliveExpire/PlatformKeepaliveExpireEventLister.java

@@ -63,6 +63,7 @@
        if (parentPlatformCatch.getKeepAliveReply() >= 3) {
            // 有3次未收到心跳回复, 设置平台状态为离线, 开始重新注册
            logger.warn("有3次未收到心跳回复,标记设置平台状态为离线, 并重新注册 平台国标ID：" + event.getPlatformGbID());
            storager.updateParentPlatformStatus(event.getPlatformGbID(), false);
            publisher.platformNotRegisterEventPublish(event.getPlatformGbID());
            parentPlatformCatch.setKeepAliveReply(0);
        }else {

 src/main/java/com/genersoft/iot/vmp/gb28181/event/platformNotRegister/PlatformNotRegisterEventLister.java

@@ -1,13 +1,20 @@
package com.genersoft.iot.vmp.gb28181.event.platformNotRegister;

import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Description: 平台未注册事件,来源有二:
@@ -23,9 +30,14 @@

    @Autowired
    private IVideoManagerStorager storager;
    @Autowired
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private SIPCommanderFroPlatform sipCommanderFroPlatform;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;

    // @Autowired
    // private RedisUtil redis;
@@ -33,13 +45,38 @@
    @Override
    public void onApplicationEvent(PlatformNotRegisterEvent event) {

        logger.debug("平台未注册事件触发，平台国标ID：" + event.getPlatformGbID());
        logger.info("平台未注册事件触发，平台国标ID：" + event.getPlatformGbID());

        ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformGbID());
        if (parentPlatform == null) {
            logger.debug("平台未注册事件触发，但平台已经删除!!! 平台国标ID：" + event.getPlatformGbID());
            logger.info("平台未注册事件触发，但平台已经删除!!! 平台国标ID：" + event.getPlatformGbID());
            return;
        }
        // 查询是否有推流， 如果有则都停止
        List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(event.getPlatformGbID());
        logger.info("停止[ {} ]的所有推流size", sendRtpItems.size());
        if (sendRtpItems != null && sendRtpItems.size() > 0) {
            logger.info("停止[ {} ]的所有推流", event.getPlatformGbID());
            StringBuilder app = new StringBuilder();
            StringBuilder stream = new StringBuilder();
            for (int i = 0; i < sendRtpItems.size(); i++) {
                if (app.length() != 0) {
                    app.append(",");
                }
                app.append(sendRtpItems.get(i).getApp());
                if (stream.length() != 0) {
                    stream.append(",");
                }
                stream.append(sendRtpItems.get(i).getStreamId());
                redisCatchStorage.deleteSendRTPServer(event.getPlatformGbID(), sendRtpItems.get(i).getChannelId());
            }
            Map<String, Object> param = new HashMap<>();
            param.put("vhost","__defaultVhost__");
            param.put("app", app.toString());
            param.put("stream", stream.toString());
            zlmrtpServerFactory.stopSendRtpStream(param);

        }
        sipCommanderFroPlatform.register(parentPlatform);
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/CheckForAllRecordsThread.java

@@ -43,7 +43,7 @@
            if (totalRecordList.size() < this.recordInfo.getSumNum()) {
                logger.info("已获取" + totalRecordList.size() + "项录像数据，共" + this.recordInfo.getSumNum() + "项");
            } else {
                logger.info("录像数据已全部获取，共" + this.recordInfo.getSumNum() + "项");
                logger.info("录像数据已全部获取，共 {} 项", this.recordInfo.getSumNum());
                this.recordInfo.setRecordList(totalRecordList);
                for (int i = 0; i < cacheKeys.size(); i++) {
                    redis.del(cacheKeys.get(i).toString());

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java

@@ -58,6 +58,7 @@
                param.put("vhost","__defaultVhost__");

                param.put("app",sendRtpItem.getApp());

                param.put("stream",streamId);

                param.put("ssrc",sendRtpItem.getSsrc());

                logger.info("停止向上级推流：" + streamId);

                zlmrtpServerFactory.stopSendRtpStream(param);

                redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java

@@ -93,6 +93,11 @@
                GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);

                // 不是通道可能是直播流

                if (channel != null || gbStream != null ) {

                    if (channel.getStatus() == 0) {

                        logger.info("通道离线，返回400");

                        responseAck(evt, Response.BAD_REQUEST, "channel [" + channel.getChannelId() + "] offline");

                        return;

                    }

                    responseAck(evt, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中

                }else {

                    logger.info("通道不存在，返回404");

@@ -367,6 +372,12 @@
        getServerTransaction(evt).sendResponse(response);

    }



    private void responseAck(RequestEvent evt, int statusCode, String msg) throws SipException, InvalidArgumentException, ParseException {

        Response response = getMessageFactory().createResponse(statusCode, evt.getRequest());

        response.setReasonPhrase(msg);

        getServerTransaction(evt).sendResponse(response);

    }



    /**

     * 回复带sdp的200

     * @param evt


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java

@@ -770,14 +770,17 @@
        try {

            Element rootElement = getRootElement(evt);

            String deviceId = XmlUtil.getText(rootElement, "DeviceID");

            // 检查设备是否存在， 不存在则不回复

            if (storager.exists(deviceId)) {

            Device device = storager.queryVideoDevice(deviceId);

            // 检查设备是否存在并在线， 不存在则不回复

            if (device != null && device.getOnline() == 1) {

                // 回复200 OK

                responseAck(evt);

                if (offLineDetector.isOnline(deviceId)) {

                    publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);

                } else {

                }

            }else {

                logger.warn("收到[ "+deviceId+" ]心跳信息, 但是设备" + (device == null? "不存在":"离线") + ", 心跳信息不予以回复");

            }

        } catch (ParseException | SipException | InvalidArgumentException | DocumentException e) {

            e.printStackTrace();


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/RegisterRequestProcessor.java

@@ -146,7 +146,7 @@
            // 注册成功

            // 保存到redis

            // 下发catelog查询目录

            if (registerFlag == 1 && device != null) {

            if (registerFlag == 1 ) {

                logger.info("注册成功! deviceId:" + device.getDeviceId());

                // boolean exists = storager.exists(device.getDeviceId());

                device.setRegisterTimeMillis(System.currentTimeMillis());


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/RegisterResponseProcessor.java

@@ -80,12 +80,13 @@
            // 注册/注销成功
            logger.info(String.format("%s %s成功", platformGBId, action));
            redisCatchStorage.delPlatformRegisterInfo(callId);
            parentPlatform.setStatus(true);
            parentPlatform.setStatus("注册".equals(action));
            // 取回Expires设置，避免注销过程中被置为0
            ParentPlatform parentPlatformTmp = storager.queryParentPlatByServerGBId(platformGBId);
            String expires = parentPlatformTmp.getExpires();
            parentPlatform.setExpires(expires);
            storager.updateParentPlatform(parentPlatform);
            parentPlatform.setId(parentPlatformTmp.getId());
            storager.updateParentPlatformStatus(platformGBId, "注册".equals(action));

            redisCatchStorage.updatePlatformRegister(parentPlatform);


 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java

@@ -1,6 +1,9 @@
package com.genersoft.iot.vmp.media.zlm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
@@ -119,4 +122,33 @@
            storager.mediaOutline(app, streamId);
        }
    }

    public void clearAllSessions() {
        logger.info("清空所有国标相关的session");
        JSONObject allSessionJSON = zlmresTfulUtils.getAllSession();
        MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
        HashSet<String> allLocalPorts = new HashSet();
        if (allSessionJSON.getInteger("code") == 0) {
            JSONArray data = allSessionJSON.getJSONArray("data");
            if (data.size() > 0) {
                for (int i = 0; i < data.size(); i++) {
                    JSONObject sessionJOSN = data.getJSONObject(i);
                    Integer local_port = sessionJOSN.getInteger("local_port");
                    if (!local_port.equals(Integer.valueOf(mediaInfo.getHttpPort())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getHttpSSLport())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getRtmpPort())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getRtspPort())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getRtspSSlport())) &&
                        !local_port.equals(Integer.valueOf(mediaInfo.getHookOnFlowReport()))){
                        allLocalPorts.add(sessionJOSN.getInteger("local_port") + "");
                     }
                }
            }
        }
        if (allLocalPorts.size() > 0) {
            List<String> result = new ArrayList<>(allLocalPorts);
            String localPortSStr = String.join(",", result);
            zlmresTfulUtils.kickSessions(localPortSStr);
        }
    }
}

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java

@@ -194,4 +194,14 @@
        param.put("force", 1);
        return sendPost("close_streams",param, null);
    }

    public JSONObject getAllSession() {
        return sendPost("getAllSession",null, null);
    }

    public void kickSessions(String localPortSStr) {
        Map<String, Object> param = new HashMap<>();
        param.put("local_port", localPortSStr);
        sendPost("kick_sessions",param, null);
    }
}

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java

@@ -18,13 +18,13 @@

    private Logger logger = LoggerFactory.getLogger("ZLMRTPServerFactory");

    @Value("${media.rtp.udpPortRange}")
    private String udpPortRange;
    @Value("${media.rtp.portRange}")
    private String portRange;

    @Autowired
    private ZLMRESTfulUtils zlmresTfulUtils;

    private int[] udpPortRangeArray = new int[2];
    private int[] portRangeArray = new int[2];

    private int currentPort = 0;

@@ -52,7 +52,7 @@

        Map<String, Object> param = new HashMap<>();
        int result = -1;
        int newPort = getPortFromUdpPortRange();
        int newPort = getPortFromportRange();
        param.put("port", newPort);
        param.put("enable_tcp", 1);
        param.put("stream_id", streamId);
@@ -101,16 +101,16 @@
        return result;
    }

    private int getPortFromUdpPortRange() {
    private int getPortFromportRange() {
        if (currentPort == 0) {
            String[] udpPortRangeStrArray = udpPortRange.split(",");
            udpPortRangeArray[0] = Integer.parseInt(udpPortRangeStrArray[0]);
            udpPortRangeArray[1] = Integer.parseInt(udpPortRangeStrArray[1]);
            String[] portRangeStrArray = portRange.split(",");
            portRangeArray[0] = Integer.parseInt(portRangeStrArray[0]);
            portRangeArray[1] = Integer.parseInt(portRangeStrArray[1]);
        }

        if (currentPort == 0 || currentPort++ > udpPortRangeArray[1]) {
            currentPort = udpPortRangeArray[0];
            return udpPortRangeArray[0];
        if (currentPort == 0 || currentPort++ > portRangeArray[1]) {
            currentPort = portRangeArray[0];
            return portRangeArray[0];
        } else {
            if (currentPort % 2 == 1) {
                currentPort++;
@@ -244,4 +244,8 @@
        }
        return result;
    }

    public void closeAllSendRtpStream() {

    }
}

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java

@@ -164,6 +164,10 @@
        mediaServerConfig.setLocalIP(mediaIp);
        mediaServerConfig.setWanIp(StringUtils.isEmpty(mediaWanIp)? mediaIp: mediaWanIp);
        redisCatchStorage.updateMediaInfo(mediaServerConfig);

        // 清空所有session
//        zlmMediaListManager.clearAllSessions();

        // 更新流列表
        zlmMediaListManager.updateMediaList();
        // 恢复流代理

 src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java

@@ -6,6 +6,7 @@
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;

import java.util.List;
import java.util.Map;

public interface IRedisCatchStorage {
@@ -91,6 +92,8 @@
     */
    SendRtpItem querySendRTPServer(String platformGbId, String channelId);

    List<SendRtpItem> querySendRTPServer(String platformGbId);

    /**
     * 删除RTP推送信息缓存
     * @param platformGbId

 src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java

@@ -135,6 +135,13 @@
     */

    public boolean outline(String deviceId);



    /**

     * 更新所有设备离线

     *

     * @return true：更新成功  false：更新失败

     */

    public boolean outlineForAll();





    /**

     * 查询子设备

@@ -352,4 +359,10 @@
     * @param streamId

     */

    void mediaOutline(String app, String streamId);



    /**

     * 设置平台在线/离线

     * @param online

     */

    void updateParentPlatformStatus(String platformGbID, boolean online);

}


 src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java

@@ -65,4 +65,7 @@

    @Delete("DELETE FROM device WHERE deviceId=#{deviceId}")
    int del(String deviceId);

    @Update("UPDATE device SET online=0")
    int outlineForAll();
}

 src/main/java/com/genersoft/iot/vmp/storager/dao/ParentPlatformMapper.java

@@ -66,5 +66,8 @@
    ParentPlatform getParentPlatById(int id);

    @Update("UPDATE parent_platform SET status=false" )
    void outlineForAllParentPlatform();
    int outlineForAllParentPlatform();

    @Update("UPDATE parent_platform SET status=#{online} WHERE serverGBId=#{platformGbID}" )
    int updateParentPlatformStatus(String platformGbID, boolean online);
}

 src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java

@@ -233,6 +233,20 @@
        return (SendRtpItem)redis.get(key);
    }

    @Override
    public List<SendRtpItem> querySendRTPServer(String platformGbId) {
        String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + platformGbId + "_*";
        List<Object> queryResult = redis.scan(key);
        List<SendRtpItem> result= new ArrayList<>();

        for (int i = 0; i < queryResult.size(); i++) {
            String keyItem = (String) queryResult.get(i);
            result.add((SendRtpItem)redis.get(keyItem));
        }

        return result;
    }

    /**
     * 删除RTP推送信息缓存
     * @param platformGbId

 src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java

@@ -257,6 +257,18 @@
    }

    /**
     * 更新所有设备离线
     *
     * @return true：更新成功  false：更新失败
     */
    @Override
    public synchronized boolean outlineForAll() {
        logger.info("更新所有设备离线");
        int result = deviceMapper.outlineForAll();
        return result > 0;
    }

    /**
     * 清空通道
     * @param deviceId
     */
@@ -575,5 +587,8 @@
        gbStreamMapper.setStatus(app, streamId, false);
    }


    @Override
    public void updateParentPlatformStatus(String platformGbID, boolean online) {
        platformMapper.updateParentPlatformStatus(platformGbID, online);
    }
}

 src/main/resources/application-dev.yml

@@ -78,12 +78,12 @@
    autoApplyPlay: false
    # [可选] 部分设备需要扩展SDP，需要打开此设置
    seniorSdp: false
    # 启用udp多端口模式
    # 启用多端口模式, 多端口模式使用端口区分每路流，兼容性更好。 单端口使用流的ssrc区分， 点播超时建议使用多端口测试
    rtp:
        # [可选] 是否启用udp多端口模式, 开启后会在udpPortRange范围内选择端口用于媒体流传输
        # [可选] 是否启用多端口模式, 开启后会在portRange范围内选择端口用于媒体流传输
        enable: true
        # [可选] 在此范围内选择端口用于媒体流传输, 不只是udp, 使用TCP被动传输模式时,也是从这个范围内选择端口
        udpPortRange: 30000,30500 # 端口范围
        # [可选] 在此范围内选择端口用于媒体流传输,
        portRange: 30000,30500 # 端口范围

# [可选] 日志配置, 一般不需要改
logging: