实现国标录像级联播放，优化点播流程，加快点播速度

648540858

2022-02-27 e94b99d11c46246532edc93cd25cbf8c0b88f03f

实现国标录像级联播放，优化点播流程，加快点播速度

 src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java

@@ -71,6 +71,16 @@
     */
    private String mediaServerId;

    /**
     *  invite的callId
     */
    private String CallId;

    /**
     * 是否是play， false是playback
     */
    private boolean isPlay;

    public String getIp() {
        return ip;
    }
@@ -174,4 +184,20 @@
    public void setMediaServerId(String mediaServerId) {
        this.mediaServerId = mediaServerId;
    }

    public String getCallId() {
        return CallId;
    }

    public void setCallId(String callId) {
        CallId = callId;
    }

    public boolean isPlay() {
        return isPlay;
    }

    public void setPlay(boolean play) {
        isPlay = play;
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/session/SsrcConfig.java

@@ -81,7 +81,6 @@
            isUsed.remove(sn);
            notUsed.add(sn);
        }catch (NullPointerException e){
            System.out.printf("11111");
        }
    }


 src/main/java/com/genersoft/iot/vmp/gb28181/task/GPSSubscribeTask.java

@@ -36,7 +36,6 @@

        SubscribeInfo subscribe = redisCatchStorage.getSubscribe(key);
        if (subscribe != null) {
            System.out.println("发送GPS消息");
            ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
            if (parentPlatform == null || parentPlatform.isStatus()) {
                // TODO 暂时只处理视频流的回复,后续增加对国标设备的支持

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java

@@ -141,7 +141,6 @@
     */
    @Override
    public void processTimeout(TimeoutEvent timeoutEvent) {
        System.out.println("processTimeout");
        if(timeoutProcessor != null) {
            timeoutProcessor.process(timeoutEvent);
        }
@@ -173,7 +172,6 @@

    @Override
    public void processDialogTerminated(DialogTerminatedEvent dialogTerminatedEvent) {
        System.out.println("processDialogTerminated");
        CallIdHeader callId = dialogTerminatedEvent.getDialog().getCallId();
    }


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java

@@ -346,8 +346,11 @@
            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,

                    (MediaServerItem mediaServerItemInUse, JSONObject json)->{

                if (userSetup.isWaitTrack() && json.getJSONArray("tracks") == null) return;

                event.response(mediaServerItemInUse, json);

                subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);

                if (event != null) {

                    event.response(mediaServerItemInUse, json);

                }



//                subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);

            });

            //

            StringBuffer content = new StringBuffer(200);

@@ -452,9 +455,11 @@
            logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());

            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,

                    (MediaServerItem mediaServerItemInUse, JSONObject json)->{

                        System.out.println(344444);

                if (userSetup.isWaitTrack() && json.getJSONArray("tracks") == null) return;

                event.response(mediaServerItemInUse, json);

                subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);

                if (event != null) {

                    event.response(mediaServerItemInUse, json);

                }

            });



            StringBuffer content = new StringBuffer(200);

@@ -465,8 +470,6 @@
            content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");

            content.append("t="+DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime)+" "

                    +DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime) +"\r\n");







            String streamMode = device.getStreamMode().toUpperCase();



@@ -1202,7 +1205,6 @@
        if (type == null) {

            type = "all";

        }



        try {

            StringBuffer recordInfoXml = new StringBuffer(200);

            recordInfoXml.append("<?xml version=\"1.0\" encoding=\"GB2312\"?>\r\n");


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java

@@ -508,9 +508,6 @@
            // callid
            CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                    : udpSipProvider.getNewCallId();
            System.out.println(
                    recordXml.toString()
            );
            Request request = headerProviderPlarformProvider.createMessageRequest(parentPlatform, recordXml.toString(), fromTag, callIdHeader);
            transmitRequest(parentPlatform, request);


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

@@ -1,10 +1,13 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -24,6 +27,8 @@
import javax.sip.header.ToHeader;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

/**
 * SIP命令类型： ACK请求
@@ -52,6 +57,9 @@
    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private ZLMHttpHookSubscribe subscribe;


    /**   
     * 处理  ACK请求
@@ -60,6 +68,7 @@
     */
    @Override
    public void process(RequestEvent evt) {
        logger.debug("ACK请求： {}", ((System.currentTimeMillis())));
        Dialog dialog = evt.getDialog();
        if (dialog == null) return;
        if (dialog.getState()== DialogState.CONFIRMED) {
@@ -69,16 +78,17 @@
            String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
            String deviceId = sendRtpItem.getDeviceId();
            StreamInfo streamInfo = null;
            if (deviceId == null) {
            if (sendRtpItem.isPlay()) {
                streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
            }else {
                streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
            }
            System.out.println(JSON.toJSON(streamInfo));
            if (streamInfo == null) {
                streamInfo = new StreamInfo();
                streamInfo.setApp(sendRtpItem.getApp());
                streamInfo.setStreamId(sendRtpItem.getStreamId());
            }else {
                streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
                sendRtpItem.setStreamId(streamInfo.getStreamId());
                streamInfo.setApp("rtp");
            }

            redisCatchStorage.updateSendRTPSever(sendRtpItem);
            logger.info(platformGbId);
            logger.info(channelId);
@@ -90,34 +100,42 @@
            param.put("dst_url",sendRtpItem.getIp());
            param.put("dst_port", sendRtpItem.getPort());
            param.put("is_udp", is_Udp);
            //param.put ("src_port", sendRtpItem.getLocalPort());
            // 设备推流查询，成功后才能转推
            boolean rtpPushed = false;
            long startTime = System.currentTimeMillis();
            while (!rtpPushed) {
                try {
                    if (System.currentTimeMillis() - startTime < 30 * 1000) {
                        MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
                        if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStreamId())) {
                            rtpPushed = true;
                            logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
                                    streamInfo.getApp() ,streamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
                            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
                        } else {
                            logger.info("等待设备推流[{}/{}].......",
                                    streamInfo.getApp() ,streamInfo.getStreamId());
                            Thread.sleep(1000);
                            continue;
                        }
                    } else {
                        rtpPushed = true;
                        logger.info("设备推流[{}/{}]超时，终止向上级推流",
                                streamInfo.getApp() ,streamInfo.getStreamId());
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//            if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStreamId())) {
//                logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
//                        streamInfo.getApp() ,streamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
//                zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//            } else {
//                // 对hook进行订阅
//                logger.info("等待设备推流[{}/{}].......",
//                        streamInfo.getApp(), streamInfo.getStreamId());
//                Timer timer = new Timer();
//                timer.schedule(new TimerTask() {
//                    @Override
//                    public void run() {
//                        logger.info("设备推流[{}/{}]超时，终止向上级推流",
//                                finalStreamInfo.getApp() , finalStreamInfo.getStreamId());
//
//                    }
//                }, 30*1000L);
//                // 添加订阅
//                JSONObject subscribeKey = new JSONObject();
//                subscribeKey.put("app", "rtp");
//                subscribeKey.put("stream", streamInfo.getStreamId());
//                subscribeKey.put("mediaServerId", streamInfo.getMediaServerId());
//                subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, subscribeKey,
//                        (MediaServerItem mediaServerItemInUse, JSONObject json) -> {
//                            logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
//                                    finalStreamInfo.getApp(), finalStreamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
//                            timer.cancel();
//                            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//                            subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
//                        });
//            }


        }
    }
}

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

@@ -87,18 +87,29 @@
                    MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
                    zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
                    redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);
                    if (zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId) == 0) {
                    int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
                    if (totalReaderCount == 0) {
                        logger.info(streamId + "无其它观看者，通知设备停止推流");
                        cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId);
                    }else if (totalReaderCount == -1){
                        logger.warn(streamId + " 查找其它观看者失败");
                    }
                }
                // 可能是设备主动停止
                Device device = storager.queryVideoDeviceByChannelId(platformGbId);
                if (device != null) {
                    StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
                    if (streamInfo != null) {
                        redisCatchStorage.stopPlay(streamInfo);
                    if (sendRtpItem.isPlay()) {
                        StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
                        if (streamInfo != null) {
                            redisCatchStorage.stopPlay(streamInfo);
                        }
                    }else {
                        StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(device.getDeviceId(), channelId);
                        if (streamInfo != null) {
                            redisCatchStorage.stopPlayback(streamInfo);
                        }
                    }

                    storager.stopPlay(device.getDeviceId(), channelId);
                    mediaServerService.closeRTPServer(device, channelId);
                }

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

@@ -1,18 +1,28 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import gov.nist.javax.sdp.TimeDescriptionImpl;
import gov.nist.javax.sdp.fields.TimeField;
import gov.nist.javax.sip.address.AddressImpl;
import gov.nist.javax.sip.address.SipUri;
import org.slf4j.Logger;
@@ -27,10 +37,13 @@
import javax.sip.ServerTransaction;
import javax.sip.SipException;
import javax.sip.address.SipURI;
import javax.sip.header.CallIdHeader;
import javax.sip.header.FromHeader;
import javax.sip.message.Request;
import javax.sip.message.Response;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Vector;

@@ -61,6 +74,9 @@
    private IPlayService playService;

    @Autowired
    private ISIPCommander commander;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;

    @Autowired
@@ -68,6 +84,7 @@

    @Autowired
    private SIPProcessorObserver sipProcessorObserver;


    @Override
    public void afterPropertiesSet() throws Exception {
@@ -84,6 +101,7 @@
    @Override
    public void process(RequestEvent evt) {
        //  Invite Request消息实现，此消息一般为级联消息，上级给下级发送请求视频指令
        Long startTimeForInvite = System.currentTimeMillis();
        try {
            Request request = evt.getRequest();
            SipURI sipURI = (SipURI) request.getRequestURI();
@@ -91,6 +109,7 @@
            String requesterId = null;

            FromHeader fromHeader = (FromHeader)request.getHeader(FromHeader.NAME);
            CallIdHeader callIdHeader = (CallIdHeader)request.getHeader(CallIdHeader.NAME);
            AddressImpl address = (AddressImpl) fromHeader.getAddress();
            SipUri uri = (SipUri) address.getURI();
            requesterId = uri.getUser();
@@ -101,7 +120,7 @@
                return;
            }

            // 查询请求方是否上级平台
            // 查询请求是否来自上级平台\设备
            ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
            if (platform != null) {
                // 查询平台下是否有该通道
@@ -158,7 +177,21 @@
                    ssrc = ssrcDefault;
                    sdp = SdpFactory.getInstance().createSessionDescription(contentString);
                }
                String sessionName = sdp.getSessionName().getValue();

                Long startTime = null;
                Long stopTime = null;
                Date start = null;
                Date end = null;
                if (sdp.getTimeDescriptions(false) != null && sdp.getTimeDescriptions(false).size() > 0) {
                    TimeDescriptionImpl timeDescription = (TimeDescriptionImpl)(sdp.getTimeDescriptions(false).get(0));
                    TimeField startTimeFiled = (TimeField)timeDescription.getTime();
                    startTime = startTimeFiled.getStartTime();
                    stopTime = startTimeFiled.getStopTime();

                    start = new Date(startTime*1000);
                    end = new Date(stopTime*1000);
                }
                //  获取支持的格式
                Vector mediaDescriptions = sdp.getMediaDescriptions(true);
                // 查看是否支持PS 负载96
@@ -228,23 +261,31 @@
                        responseAck(evt, Response.BUSY_HERE);
                        return;
                    }

                    sendRtpItem.setCallId(callIdHeader.getCallId());
                    sendRtpItem.setPlay("Play".equals(sessionName));
                    // 写入redis， 超时时回复
                    redisCatchStorage.updateSendRTPSever(sendRtpItem);
                    // 通知下级推流，
                    PlayResult playResult = playService.play(mediaServerItem,device.getDeviceId(), channelId, (mediaServerItemInUSe, responseJSON)->{
                        // 收到推流， 回复200OK, 等待ack

                    Device finalDevice = device;
                    MediaServerItem finalMediaServerItem = mediaServerItem;
                    Long finalStartTime = startTime;
                    Long finalStopTime = stopTime;
                    ZLMHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON)->{
                        logger.info("[上级点播]收到下级开始点播订阅， {}/{}", sendRtpItem.getApp(), sendRtpItem.getStreamId());
                        // if (sendRtpItem == null) return;
                        sendRtpItem.setStatus(1);
                        redisCatchStorage.updateSendRTPSever(sendRtpItem);
                        // TODO 添加对tcp的支持

                        StringBuffer content = new StringBuffer(200);
                        content.append("v=0\r\n");
                        content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItemInUSe.getSdpIp()+"\r\n");
                        content.append("s=Play\r\n");
                        content.append("s=" + sessionName+"\r\n");
                        content.append("c=IN IP4 "+mediaServerItemInUSe.getSdpIp()+"\r\n");
                        content.append("t=0 0\r\n");
                        if ("Playback".equals(sessionName)) {
                            content.append("t=" + finalStartTime + " " + finalStopTime + "\r\n");
                        }else {
                            content.append("t=0 0\r\n");
                        }
                        content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");
                        content.append("a=sendonly\r\n");
                        content.append("a=rtpmap:96 PS/90000\r\n");
@@ -260,7 +301,11 @@
                        } catch (ParseException e) {
                            e.printStackTrace();
                        }
                    } ,((event) -> {
                        if ("Playback".equals(sessionName) && responseJSON != null) {
                            playService.onPublishHandlerForPlayBack(finalMediaServerItem, responseJSON, finalDevice.getDeviceId(), channelId, null);
                        }
                    };
                    SipSubscribe.Event errorEvent = ((event) -> {
                        // 未知错误。直接转发设备点播的错误
                        Response response = null;
                        try {
@@ -271,11 +316,27 @@
                        } catch (ParseException | SipException | InvalidArgumentException e) {
                            e.printStackTrace();
                        }
                    }));
                    if (logger.isDebugEnabled()) {
                        logger.debug(playResult.getResult().toString());
                    });
                    if ("Playback".equals(sessionName)) {
                        sendRtpItem.setPlay(false);
                        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, sendRtpItem.getSsrc(), true);
                        sendRtpItem.setStreamId(ssrc);
                        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        commander.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, format.format(start), format.format(end), hookEvent, errorEvent);
                    }else {
                        sendRtpItem.setPlay(true);
                        StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
                        if (streamInfo == null) {
                            if (mediaServerItem.isRtpEnable()) {
                                sendRtpItem.setStreamId(String.format("%s_%s", device.getDeviceId(), channelId));
                            }
                            sendRtpItem.setPlay(false);
                            playService.play(mediaServerItem,device.getDeviceId(), channelId, hookEvent,errorEvent);
                        }else {
                            sendRtpItem.setStreamId(streamInfo.getStreamId());
                            hookEvent.response(mediaServerItem, null);
                        }
                    }

                }else if (gbStream != null) {
                    SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                            gbStream.getApp(), gbStream.getStream(), channelId,
@@ -295,7 +356,6 @@

                    sendRtpItem.setStatus(1);
                    redisCatchStorage.updateSendRTPSever(sendRtpItem);
                    // TODO 添加对tcp的支持
                    StringBuffer content = new StringBuffer(200);
                    content.append("v=0\r\n");
                    content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");

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

@@ -82,9 +82,6 @@
                requestURI.setPort(event.getRemotePort());
                reqAck.setRequestURI(requestURI);
                logger.info("向 " + event.getRemoteIpAddress() + ":" + event.getRemotePort() + "回复ack");
                SipURI sipURI = (SipURI)dialog.getRemoteParty().getURI();
                String deviceId = requestURI.getUser();
                String channelId = sipURI.getUser();

                dialog.sendAck(reqAck);


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

@@ -181,7 +181,7 @@
    @PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8")

    public ResponseEntity<String> onPublish(@RequestBody JSONObject json) {



        logger.debug("[ ZLM HOOK ]on_publish API调用，参数：" + json.toString());

        logger.info("[ ZLM HOOK ]on_publish API调用，参数：" + json.toString());

        JSONObject ret = new JSONObject();

        ret.put("code", 0);

        ret.put("msg", "success");


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

@@ -77,21 +77,23 @@
        if (eventMap == null) {
            return;
        }
        Iterator<Map.Entry<JSONObject, Event>> iterator = eventMap.entrySet().iterator();
        while (iterator.hasNext()){
            Map.Entry<JSONObject, Event> next = iterator.next();
            JSONObject key = next.getKey();
            Boolean result = null;
            for (String s : key.keySet()) {
                if (result == null) {
                    result = key.getString(s).equals(hookResponse.getString(s));
                }else {
                    if (key.getString(s) == null) continue;
                    result = result && key.getString(s).equals(hookResponse.getString(s));

        Set<Map.Entry<JSONObject, Event>> entries = eventMap.entrySet();
        if (entries.size() > 0) {
            for (Map.Entry<JSONObject, Event> entry : entries) {
                JSONObject key = entry.getKey();
                Boolean result = null;
                for (String s : key.keySet()) {
                    if (result == null) {
                        result = key.getString(s).equals(hookResponse.getString(s));
                    }else {
                        if (key.getString(s) == null) continue;
                        result = result && key.getString(s).equals(hookResponse.getString(s));
                    }
                }
            }
            if (null != result && result){
                iterator.remove();
                if (null != result && result){
                    entries.remove(entry);
                }
            }
        }
    }

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

@@ -72,7 +72,6 @@
                        ResponseBody responseBody = response.body();
                        if (responseBody != null) {
                            String responseStr = responseBody.string();
                            System.out.println(responseStr);
                            responseJSON = JSON.parseObject(responseStr);
                        }
                    }else {

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

@@ -242,9 +242,18 @@
     */
    public int totalReaderCount(MediaServerItem mediaServerItem, String app, String streamId) {
        JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtmp", streamId);
        Integer code = mediaInfo.getInteger("code");
        if (mediaInfo == null) {
            return 0;
        }
        if ( code < 0) {
            logger.warn("查询流({}/{})是否有其它观看者时得到： {}", app, streamId, mediaInfo.getString("msg"));
            return -1;
        }
        if ( code == 0 && ! mediaInfo.getBoolean("online")) {
            logger.warn("查询流({}/{})是否有其它观看者时得到： {}", app, streamId, mediaInfo.getString("msg"));
            return -1;
        }
        return mediaInfo.getInteger("totalReaderCount");
    }


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

@@ -122,7 +122,6 @@
            // 点播结束时调用截图接口
            try {
                String classPath = ResourceUtils.getURL("classpath:").getPath();
                // System.out.println(classPath);
                // 兼容打包为jar的class路径
                if(classPath.contains("jar")) {
                    classPath = classPath.substring(0, classPath.lastIndexOf("."));
@@ -238,11 +237,11 @@
    }

    @Override
    public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
    public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
        RequestMessage msg = new RequestMessage();
        msg.setId(uuid);
        msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
        StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid);
        StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId, uuid);
        if (streamInfo != null) {
            DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
            if (deviceChannel != null) {

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

@@ -88,7 +88,7 @@
            "</script>"})
    int setDefaultCatalog(String platformId, String catalogId);

    @Select("select 'channel' as name, count(pgc.platformId) count from platform_gb_channel pgc  where  pgc.platformId=#{platformId} and pgc.channelId =#{gbId} " +
    @Select("select 'channel' as name, count(pgc.platformId) count from platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId where  pgc.platformId=#{platformId} and dc.channelId =#{gbId} " +
            "union " +
            "select 'stream' as name, count(pgs.platformId) count from platform_gb_stream pgs left join gb_stream gs on pgs.gbStreamId = gs.gbStreamId where  pgs.platformId=#{platformId} and gs.gbId = #{gbId}")
    List<ChannelSourceInfo> getChannelSource(String platformId, String gbId);

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

@@ -52,8 +52,8 @@
    int cleanChannelForGB(String platformId);


    @Select("SELECT * FROM device_channel WHERE deviceId = (SELECT deviceId FROM platform_gb_channel WHERE " +
            "platformId='${platformId}' AND channelId='${channelId}' ) AND channelId='${channelId}'")
    @Select("SELECT dc.* FROM platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId WHERE " +
            "pgc.platformId=#{platformId} AND dc.channelId=#{channelId}")
    DeviceChannel queryChannelInParentPlatform(String platformId, String channelId);


@@ -62,7 +62,7 @@
            "where pgc.platformId=#{platformId} and pgc.catalogId=#{catalogId}")
    List<PlatformCatalog> queryChannelInParentPlatformAndCatalog(String platformId, String catalogId);

    @Select("SELECT * FROM device WHERE deviceId = (SELECT deviceId FROM platform_gb_channel WHERE platformId='${platformId}' AND channelId='${channelId}')")
    @Select("SELECT * FROM device WHERE deviceId = (SELECT deviceId FROM platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId WHERE pgc.platformId='${platformId}' AND dc.channelId='${channelId}')")
    Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId);

    @Delete("<script> "+
@@ -71,7 +71,7 @@
    int delByCatalogId(String id);

    @Delete("<script> "+
           "DELETE FROM platform_gb_channel WHERE catalogId=#{parentId} AND platformId=#{platformId} AND channelId=#{id}"  +
           "DELETE FROM platform_gb_channel  WHERE catalogId=#{parentId} AND platformId=#{platformId} AND channelId=#{id}"  +
           "</script>")
    int delByCatalogIdAndChannelIdAndPlatformId(PlatformCatalog platformCatalog);


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

@@ -139,7 +139,6 @@

    @Override
    public StreamInfo queryPlayByDevice(String deviceId, String channelId) {
//        List<Object> playLeys = redis.keys(String.format("%S_*_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
        List<Object> playLeys = redis.scan(String.format("%S_%s_*_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
                userSetup.getServerId(),
                deviceId,

 src/test/java/com/genersoft/iot/vmp/service/impl/DeviceAlarmServiceImplTest.java

@@ -50,14 +50,7 @@
//        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, "1", null,
//                null, null).getSize());

        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
                "2021-01-01 00:00:00", null).getSize());

        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
                null, "2021-04-01 09:00:00").getSize());

        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
                "2021-02-01 01:00:00", "2021-04-01 04:00:00").getSize());
    }